fix: clamp straddling reconciliation ranges to the receiver's window

preProcessPayload only skipped ranges ENTIRELY below the receiver's sync
window, so when a wide-window peer (e.g. a store node) reconciled with a
narrow-window peer (e.g. a core node), the recursive split bottomed out
in a range straddling the narrow node's window floor and its below-window
items were still offered for transfer. Now a straddling range is split:
the below-window portion becomes Skip so the wide peer no longer offers
history older than the narrow node retains. Only triggers on a window
mismatch; equal windows never straddle, so store<->store and core<->core
sync are unchanged (full suite 37/37).

Added a reconciliation test proving a 2h-window initiator does not push a
45-min-old message to a 30-min-window peer.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

logos_delivery/waku/waku_store_sync/reconciliation.nim

@@ -155,6 +155,21 @@ proc preProcessPayload(self: SyncReconciliation, payload: RangesData): Opt[Range
       elif rangeType == RangeType.ItemSet:
         payload.itemSets.delete(0)
     else:
+      # This range straddles our window's lower bound: mark the portion below
+      # our window as skip so a peer with a wider window does not reconcile
+      # (and offer us) messages older than we retain. Only triggers on a
+      # window mismatch; equal windows never straddle.
+      let lowerBound = payload.ranges[i][0].a.time
+      if lowerBound < selfLowerBound:
+        let clampId = SyncID(time: selfLowerBound, hash: EmptyFingerprint)
+        let belowSkip = (payload.ranges[i][0].a .. clampId, RangeType.Skip)
+
+        if rangeType == RangeType.ItemSet:
+          payload.itemSets[0].elements.keepItIf(it.time >= selfLowerBound)
+
+        payload.ranges[i][0].a = clampId
+        payload.ranges.insert(belowSkip, i)
+
       break
 
   return Opt.some(payload)

tests/waku_store_sync/test_protocol.nim

@@ -160,6 +160,44 @@ suite "Waku Sync: reconciliation":
       remoteNeeds.contains((serverPeerInfo.peerId, hash2)) == true
       remoteNeeds.contains((serverPeerInfo.peerId, hash3)) == true
 
+  asyncTest "wide-window peer does not push below-window messages to a narrow-window peer":
+    ## A store-like node (wide sync range) reconciling with a core-like node
+    ## (narrow range) must not flood the narrow node with history older than
+    ## its own window: preProcessPayload clamps incoming ranges to the
+    ## receiver's window, so only in-window messages are offered for transfer.
+    const
+      narrowRange = 30.minutes
+      wideRange = 2.hours
+
+    # server = narrow (core-like) ; client = wide (store-like), initiates
+    server = await newTestWakuRecon(
+      serverSwitch, @[], @[], narrowRange, idsChannel, localWants, remoteNeeds
+    )
+    client = await newTestWakuRecon(
+      clientSwitch, @[], @[], wideRange, idsChannel, localWants, remoteNeeds
+    )
+
+    let
+      tNow = now()
+      oldTs = tNow - 2_700_000_000_000 # 45 min ago: below the narrow window
+      recentTs = tNow - 300_000_000_000 # 5 min ago: inside both windows
+      oldMsg = fakeWakuMessage(ts = oldTs, contentTopic = DefaultContentTopic)
+      recentMsg = fakeWakuMessage(ts = recentTs, contentTopic = DefaultContentTopic)
+      oldHash = computeMessageHash(DefaultPubsubTopic, oldMsg)
+      recentHash = computeMessageHash(DefaultPubsubTopic, recentMsg)
+
+    client.messageIngress(oldHash, DefaultPubsubTopic, oldMsg)
+    client.messageIngress(recentHash, DefaultPubsubTopic, recentMsg)
+
+    let res = await client.storeSynchronization(Opt.some(serverPeerInfo))
+    assert res.isOk(), res.error
+
+    check:
+      # the recent, in-window message is offered to the narrow server
+      remoteNeeds.contains((serverPeerInfo.peerId, recentHash)) == true
+      # the old, below-window message is clamped out, never offered
+      remoteNeeds.contains((serverPeerInfo.peerId, oldHash)) == false
+
   asyncTest "sync 2 nodes different hashes":
     server = await newTestWakuRecon(
       serverSwitch, @[], @[], DefaultSyncRange, idsChannel, localWants, remoteNeeds