From e57546b66f8479a6b32d742fa1a499059369744f Mon Sep 17 00:00:00 2001 From: darshankabariya Date: Fri, 17 Jul 2026 16:35:07 +0530 Subject: [PATCH] feat: startup offline-gap catch-up (store only at fresh start / long offline) At startup the node estimates its offline gap from the persisted last-online timestamp: within the sync window it catches up via full-node reconciliation (retrying until a sync peer is found, with the periodic sync loop as safety net); beyond the window, or on fresh start, it falls back to a bounded store resume, waiting for a store peer before consuming retry attempts. The catch-up runs as a background future so node startup no longer blocks up to 90 s on store peers, and shutdown bounds its cancellation wait. StoreResume is also mounted on sync-enabled full nodes so they track last-online. Part of the "Store as a startup-only dependency" experiment (step 10). Co-Authored-By: Claude Fable 5 --- logos_delivery/waku/factory/node_factory.nim | 5 +- logos_delivery/waku/node/waku_node.nim | 74 ++++++++++++++++++- logos_delivery/waku/waku_store/resume.nim | 17 +---- tests/waku_store_sync/test_all.nim | 7 +- .../waku_store_sync/test_startup_catchup.nim | 32 ++++++++ 5 files changed, 119 insertions(+), 16 deletions(-) create mode 100644 tests/waku_store_sync/test_startup_catchup.nim diff --git a/logos_delivery/waku/factory/node_factory.nim b/logos_delivery/waku/factory/node_factory.nim index 891d2602a..6eb6ad011 100644 --- a/logos_delivery/waku/factory/node_factory.nim +++ b/logos_delivery/waku/factory/node_factory.nim @@ -279,7 +279,10 @@ proc setupProtocols( return err("failed to set node waku store peer: " & error) node.peerManager.addServicePeer(storeNode, WakuStoreCodec) - if conf.storeServiceConf.isSome and conf.storeServiceConf.get().resume: + if (conf.storeServiceConf.isSome and conf.storeServiceConf.get().resume) or + conf.storeSyncConf.isSome(): + # sync-enabled full nodes track last-online so the startup catch-up can + # choose between neighbour reconciliation and store resume node.setupStoreResume() if conf.shardingConf.kind == AutoSharding: diff --git a/logos_delivery/waku/node/waku_node.nim b/logos_delivery/waku/node/waku_node.nim index 3dd9a6a32..47fbfd89c 100644 --- a/logos_delivery/waku/node/waku_node.nim +++ b/logos_delivery/waku/node/waku_node.nim @@ -136,6 +136,8 @@ type wakuKademlia*: WakuKademlia ports*: BoundPorts relayReconnectFut*: Future[void] + storeCatchUpFut*: Future[void] + storeSyncRange*: timer.Duration SubscriptionManager* = ref object of RootObj node*: WakuNode @@ -379,6 +381,8 @@ proc mountStoreSync*( let pubsubTopics = shards.mapIt($RelayShard(clusterId: cluster, shardId: it)) + node.storeSyncRange = storeSyncRange.seconds + let recon = ?await SyncReconciliation.new( pubsubTopics, contentTopics, node.peerManager, node.wakuArchive, storeSyncRange.seconds, storeSyncInterval.seconds, storeSyncRelayJitter.seconds, @@ -613,6 +617,66 @@ proc stopProvidersAndListeners*(node: WakuNode) = RequestContentTopicsHealth.clearProvider(node.brokerCtx) RequestShardTopicsHealth.clearProvider(node.brokerCtx) +func useSyncCatchUp*( + hasReconciliation: bool, + lastOnline: Timestamp, + now: Timestamp, + syncRange: timer.Duration, +): bool = + ## Startup catch-up decision: inside the sync window neighbour + ## reconciliation repairs the gap; beyond it (or on fresh start, when no + ## last-online timestamp exists) store resume is needed. + hasReconciliation and lastOnline > 0 and now - lastOnline <= syncRange.nanos + +proc storeStartupCatchUp(node: WakuNode) {.async.} = + ## Store as a startup-only dependency: estimate the offline gap from the + ## persisted last-online timestamp and catch up accordingly. Runs in the + ## background; node startup must not block on store or sync peers. + let lastOnline = node.wakuStoreResume.getLastOnlineTimestamp().valueOr: + error "catch-up: failed to read last online timestamp", error = error + Timestamp(0) + + let now = getNowInNanosecondTime() + + if useSyncCatchUp( + not node.wakuStoreReconciliation.isNil(), lastOnline, now, node.storeSyncRange + ): + info "offline gap within sync window, catching up via reconciliation", + gapSeconds = (now - lastOnline) div 1_000_000_000 + # sleep first: the switch may not be started yet and discovery may still + # be warming up; retry failed attempts, the periodic sync loop remains + # the safety net if none succeeds within the budget + for _ in 0 ..< 24: + await sleepAsync(5.seconds) + if node.peerManager.selectPeer(WakuReconciliationCodec).isNone(): + continue + (await node.wakuStoreReconciliation.storeSynchronization()).isOkOr: + error "startup reconciliation attempt failed", error = error + continue + return + warn "startup reconciliation did not complete; periodic sync remains the safety net" + else: + info "offline gap beyond sync window or fresh start, using store resume", + gapSeconds = (now - lastOnline) div 1_000_000_000 + # don't burn resume tries against an empty peer store: wait (bounded) + # for a store peer to be discovered first + var peerWait = 24 + var tries = 3 + while tries > 0: + if node.peerManager.selectPeer(WakuStoreCodec).isNone(): + peerWait -= 1 + if peerWait <= 0: + warn "no store peer found for startup resume" + return + await sleepAsync(5.seconds) + continue + (await node.wakuStoreResume.autoStoreResume()).isOkOr: + tries -= 1 + error "store resume failed", triesLeft = tries, error = $error + await sleepAsync(30.seconds) + continue + break + proc start*(node: WakuNode) {.async.} = ## Starts a created Waku Node and ## all its mounted protocols. @@ -626,7 +690,9 @@ proc start*(node: WakuNode) {.async.} = zeroPortPresent = true if not node.wakuStoreResume.isNil(): - await node.wakuStoreResume.start() + # catch up in the background: startup must not block on store peers + node.storeCatchUpFut = node.storeStartupCatchUp() + node.wakuStoreResume.startPeriodicOnly() if not node.wakuRendezvousClient.isNil(): await node.wakuRendezvousClient.start() @@ -699,6 +765,12 @@ proc stop*(node: WakuNode) {.async.} = if not node.wakuArchive.isNil(): await node.wakuArchive.stopWait() + if not node.storeCatchUpFut.isNil(): + # in-flight store queries wrap awaits in results.catch, which can swallow + # the one cancellation delivery; bound shutdown instead of waiting for + # their natural completion + discard await node.storeCatchUpFut.cancelAndWait().withTimeout(5.seconds) + if not node.wakuStoreResume.isNil(): await node.wakuStoreResume.stopWait() diff --git a/logos_delivery/waku/waku_store/resume.nim b/logos_delivery/waku/waku_store/resume.nim index 0f45b8a6b..4d380b30f 100644 --- a/logos_delivery/waku/waku_store/resume.nim +++ b/logos_delivery/waku/waku_store/resume.nim @@ -197,19 +197,10 @@ proc periodicSetLastOnline(self: StoreResume) {.async.} = self.setLastOnlineTimestamp(ts).isOkOr: error "failed to set last online timestamp", error, time = ts -proc start*(self: StoreResume) {.async.} = - # start resume process, will try thrice. - var tries = 3 - while tries > 0: - (await self.autoStoreResume()).isOkOr: - tries -= 1 - error "store resume failed", triesLeft = tries, error = $error - await sleepAsync(30.seconds) - continue - - break - - # starting periodic storage of last online timestamp +proc startPeriodicOnly*(self: StoreResume) = + ## Starts only the periodic last-online bookkeeping; the catch-up decision + ## (neighbour sync within the window vs store resume beyond it) is made by + ## the node's startup catch-up. self.handle = self.periodicSetLastOnline() proc stopWait*(self: StoreResume) {.async.} = diff --git a/tests/waku_store_sync/test_all.nim b/tests/waku_store_sync/test_all.nim index dfb83166a..bb53db6e5 100644 --- a/tests/waku_store_sync/test_all.nim +++ b/tests/waku_store_sync/test_all.nim @@ -1,3 +1,8 @@ {.used.} -import ./test_protocol, ./test_storage, ./test_codec, ./test_full_node_sync +import + ./test_protocol, + ./test_storage, + ./test_codec, + ./test_full_node_sync, + ./test_startup_catchup diff --git a/tests/waku_store_sync/test_startup_catchup.nim b/tests/waku_store_sync/test_startup_catchup.nim new file mode 100644 index 000000000..a837f3fbf --- /dev/null +++ b/tests/waku_store_sync/test_startup_catchup.nim @@ -0,0 +1,32 @@ +{.used.} + +import testutils/unittests, chronos +import + ../../logos_delivery/waku/node/waku_node, ../../logos_delivery/waku/waku_core/time + +suite "Startup catch-up decision (store as a startup-only dependency)": + const SyncRange = 1800.seconds # 30 min window + let now = getNowInNanosecondTime() + + test "gap within the sync window uses reconciliation": + let lastOnline = now - 300 * 1_000_000_000'i64 # 5 min ago + + check useSyncCatchUp(true, lastOnline, now, SyncRange) + + test "gap at the window boundary uses reconciliation": + let lastOnline = now - 1800 * 1_000_000_000'i64 + + check useSyncCatchUp(true, lastOnline, now, SyncRange) + + test "gap beyond the sync window falls back to store resume": + let lastOnline = now - 2700 * 1_000_000_000'i64 # 45 min ago + + check not useSyncCatchUp(true, lastOnline, now, SyncRange) + + test "fresh start (no last-online record) falls back to store resume": + check not useSyncCatchUp(true, Timestamp(0), now, SyncRange) + + test "without reconciliation mounted always falls back to store resume": + let lastOnline = now - 300 * 1_000_000_000'i64 + + check not useSyncCatchUp(false, lastOnline, now, SyncRange)