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 <noreply@anthropic.com>

Author: darshankabariya

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:

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()
 

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.} =

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

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)