diff --git a/logos_delivery/waku/node/waku_node/store.nim b/logos_delivery/waku/node/waku_node/store.nim index dca89d5df..73dd1c634 100644 --- a/logos_delivery/waku/node/waku_node/store.nim +++ b/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 diff --git a/logos_delivery/waku/waku_store/resume.nim b/logos_delivery/waku/waku_store/resume.nim index 4d380b30f..06963bc3c 100644 --- a/logos_delivery/waku/waku_store/resume.nim +++ b/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) diff --git a/tests/waku_store/test_resume.nim b/tests/waku_store/test_resume.nim index c8bf1bbad..c0712a6dc 100644 --- a/tests/waku_store/test_resume.nim +++ b/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