From 94551413b13f853ab581f776557baaad87962ca6 Mon Sep 17 00:00:00 2001 From: darshankabariya Date: Sat, 18 Jul 2026 02:57:41 +0530 Subject: [PATCH] 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 --- .../waku/waku_store_sync/reconciliation.nim | 15 ++++++++ tests/waku_store_sync/test_protocol.nim | 38 +++++++++++++++++++ 2 files changed, 53 insertions(+) diff --git a/logos_delivery/waku/waku_store_sync/reconciliation.nim b/logos_delivery/waku/waku_store_sync/reconciliation.nim index db3574a5e..87f5fe3e6 100644 --- a/logos_delivery/waku/waku_store_sync/reconciliation.nim +++ b/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) diff --git a/tests/waku_store_sync/test_protocol.nim b/tests/waku_store_sync/test_protocol.nim index 2d2eed707..aaf6da1c3 100644 --- a/tests/waku_store_sync/test_protocol.nim +++ b/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