fix: route store-resume catch-up through the sync ingress

Resume-fetched history is older than the archive's 20 s live-traffic
freshness filter by definition, so the whole catch-up was silently
rejected: a node past the sync window fetched its gap from a store node
and archived none of it (1,662/1,662 dropped in the mechanism probe;
stuck at ~71% delivery in the 150-node reproduction). Catch-up now
enters via syncMessageIngress - the same filter-free door the sync
transfer uses - and also feeds the reconciliation index so the next
sync round does not re-request what the store just provided.

Also repairs tests/waku_store/test_resume.nim: its node/waku_node
import stopped exporting the store api after the rebase (compile
error), and the legacy test assumed node startup blocks on resume,
which stopped being true when catch-up moved to the background. Adds
a regression test that messages older than the freshness window
survive resume.

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

logos_delivery/waku/node/waku_node/store.nim

@@ -23,6 +23,7 @@ import
   ../../waku_store/client as store_client,
   ../../waku_store/common as store_common,
   ../../waku_store/resume,
+  ../../waku_store_sync/reconciliation,
   ../peer_manager,
   ../../common/rate_limit/setting,
   ../../waku_archive
@@ -151,8 +152,21 @@ proc query*(
   return ok(response)
 
 proc setupStoreResume*(node: WakuNode) =
+  # Resume-fetched messages also feed reconciliation, so the next sync
+  # round does not re-request history the store already provided. The
+  # reconciliation protocol may mount after resume is set up; check at
+  # call time.
+  let reconIngress: ReconciliationIngress = proc(
+      msgHash: WakuMessageHash, pubsubTopic: PubsubTopic, msg: WakuMessage
+  ) {.gcsafe, raises: [].} =
+    if not node.wakuStoreReconciliation.isNil():
+      node.wakuStoreReconciliation.messageIngress(msgHash, pubsubTopic, msg)
+
   node.wakuStoreResume = StoreResume.new(
-    node.peerManager, node.wakuArchive, node.wakuStoreClient
+    node.peerManager,
+    node.wakuArchive,
+    node.wakuStoreClient,
+    reconIngress = Opt.some(reconIngress),
   ).valueOr:
     error "Failed to setup Store Resume", error = $error
     return

logos_delivery/waku/waku_store/resume.nim

@@ -14,6 +14,7 @@ import
 import
   ../common/databases/db_sqlite,
   ../waku_core,
+  ../waku_core/message/digest,
   ../waku_archive,
   ../common/nimchronos,
   ../waku_store/[client, common],
@@ -32,6 +33,12 @@ type
     timestamp: Timestamp, peer: RemotePeerInfo
   ): Future[Result[void, string]] {.async: (raises: []), closure.}
 
+  ReconciliationIngress* = proc(
+    msgHash: WakuMessageHash, pubsubTopic: PubsubTopic, msg: WakuMessage
+  ) {.gcsafe, raises: [].}
+    ## Feeds a resume-fetched message into the reconciliation storage so
+    ## peers are not asked again for messages the store already provided.
+
   StoreResume* = ref object
     handle: Future[void]
 
@@ -62,7 +69,10 @@ proc setupLastOnlineDB(): Result[SqliteDatabase, string] =
   return ok(db)
 
 proc initTransferHandler(
-    self: StoreResume, wakuArchive: WakuArchive, wakuStoreClient: WakuStoreClient
+    self: StoreResume,
+    wakuArchive: WakuArchive,
+    wakuStoreClient: WakuStoreClient,
+    reconIngress: Opt[ReconciliationIngress],
 ) =
   # guard clauses to prevent faulty callback
   if self.peerManager.isNil():
@@ -102,13 +112,29 @@ proc initTransferHandler(
         req.paginationCursor = response.paginationCursor
 
         for kv in response.messages:
-          let handleRes = catch:
-            await wakuArchive.handleMessage(kv.pubsubTopic.get(), kv.message.get())
+          let
+            pubsubTopic = kv.pubsubTopic.get()
+            msg = kv.message.get()
+            msgHash = computeMessageHash(pubsubTopic, msg)
 
-          handleRes.isOkOr:
+          # Catch-up messages are older than the archive's live-traffic
+          # freshness window by definition, so they must enter through the
+          # sync ingress, which skips that validation (same path the
+          # store-sync transfer uses).
+          let handleRes = catch:
+            await wakuArchive.syncMessageIngress(msgHash, pubsubTopic, msg)
+
+          let res = handleRes.valueOr:
             error "message transfer failed", error = error.msg
             continue
 
+          res.isOkOr:
+            error "message transfer failed", error = error
+            continue
+
+          if reconIngress.isSome():
+            reconIngress.get()(msgHash, pubsubTopic, msg)
+
         if req.paginationCursor.isNone():
           break
 
@@ -120,6 +146,7 @@ proc new*(
     peerManager: PeerManager,
     wakuArchive: WakuArchive,
     wakuStoreClient: WakuStoreClient,
+    reconIngress: Opt[ReconciliationIngress] = Opt.none(ReconciliationIngress),
 ): Result[T, string] =
   info "initializing store resume"
 
@@ -132,7 +159,7 @@ proc new*(
 
   let resume = StoreResume(db: db, replaceStmt: replaceStmt, peerManager: peerManager)
 
-  resume.initTransferHandler(wakuArchive, wakuStoreClient)
+  resume.initTransferHandler(wakuArchive, wakuStoreClient, reconIngress)
 
   return ok(resume)
 

tests/waku_store/test_resume.nim

@@ -5,7 +5,7 @@ import std/net, testutils/unittests, chronos, results
 import
   logos_delivery/waku/[
     node/peer_manager,
-    node/waku_node,
+    waku_node,
     waku_core,
     waku_store/resume,
     waku_store/common,
@@ -105,8 +105,46 @@ suite "Store Resume - End to End":
 
     await client.start()
 
-    countRes = await clientDriver.getMessagesCount()
+    # catch-up runs in the background (store is a startup-only dependency;
+    # node startup no longer blocks on it), so poll for its completion
+    var count = 0'i64
+    for _ in 0 ..< 75:
+      await sleepAsync(200.milliseconds)
+      count = (await clientDriver.getMessagesCount()).valueOr:
+        continue
+      if count == 10:
+        break
+
+    check:
+      count == 10
+
+  asyncTest "resume archives messages older than the archive freshness window":
+    ## Catch-up messages are older than the archive's 20 s live-traffic
+    ## freshness filter by definition, so resume must feed them through the
+    ## sync ingress; through the validated live path the whole catch-up
+    ## would be silently dropped.
+    let hourAgo = Timestamp(getNowInNanosecondTime() - 3_600_000_000_000)
+    let oldMessages = @[
+      fakeWakuMessage(@[byte 10], ts = hourAgo),
+      fakeWakuMessage(@[byte 11], ts = hourAgo + 1),
+      fakeWakuMessage(@[byte 12], ts = hourAgo + 2),
+      fakeWakuMessage(@[byte 13], ts = hourAgo + 3),
+      fakeWakuMessage(@[byte 14], ts = hourAgo + 4),
+    ]
+    serverDriver = serverDriver.put(DefaultPubsubTopic, oldMessages)
+
+    await client.start()
+
+    # resume from a deep, beyond-freshness-window gap: the server holds
+    # 10 fresh messages plus 5 from an hour ago, all inside the gap
+    let twoHoursAgo = Timestamp(getNowInNanosecondTime() - 7_200_000_000_000)
+    let serverPeer = server.peerInfo.toRemotePeerInfo()
+
+    let res = await client.wakuStoreResume.startStoreResume(twoHoursAgo, serverPeer)
+    assert res.isOk(), $res.error
+
+    let countRes = await clientDriver.getMessagesCount()
     assert countRes.isOk(), $countRes.error
 
     check:
-      countRes.get() == 10
+      countRes.get() == 15