test: full-node to full-node miss recovery via store sync

Integration test wiring two archive-backed nodes exactly like
mountStoreSync (shared peer manager + the three sync channels, no store
service): a missed message is recovered end to end through
reconciliation and transfer into the receiving node's archive, and one
session converges both nodes to the union of their message sets.

Part of the "Store as a startup-only dependency" experiment (step 2).

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

tests/waku_store_sync/test_all.nim

@@ -1,3 +1,3 @@
 {.used.}
 
-import ./test_protocol, ./test_storage, ./test_codec
+import ./test_protocol, ./test_storage, ./test_codec, ./test_full_node_sync

tests/waku_store_sync/test_full_node_sync.nim

@@ -0,0 +1,155 @@
+{.used.}
+
+## Integration test for the "store as a startup-only dependency" experiment:
+## two full nodes (no store service) wired exactly like `mountStoreSync` —
+## reconciliation + transfer sharing one peer manager and the three async
+## queues, backed by small in-memory archives — must recover missed
+## messages from each other end to end.
+
+import results, testutils/unittests, chronos, chronicles
+import
+  ../../logos_delivery/waku/[
+    node/peer_manager,
+    waku_core,
+    waku_core/message/digest,
+    waku_store_sync/common,
+    waku_store_sync/reconciliation,
+    waku_store_sync/transfer,
+    waku_archive/archive,
+    waku_archive/driver,
+    waku_archive/common,
+  ],
+  ../testlib/[wakucore, testasync],
+  ../waku_archive/archive_utils
+
+type FullNode = object
+  switch: Switch
+  driver: ArchiveDriver
+  archive: WakuArchive
+  recon: SyncReconciliation
+  transfer: SyncTransfer
+  peerInfo: RemotePeerInfo
+  peerManager: PeerManager
+
+proc newFullNode(): Future[FullNode] {.async.} =
+  ## Mirrors the wiring of WakuNode.mountStoreSync: one peer manager and
+  ## three shared channels connect reconciliation and transfer.
+  let switch = newTestSwitch()
+  await switch.start()
+
+  let driver = newSqliteArchiveDriver()
+  let archive = newWakuArchive(driver)
+  let peerManager = PeerManager.new(switch)
+
+  let idsChannel = newAsyncQueue[(SyncID, PubsubTopic, ContentTopic)]()
+  let wantsChannel = newAsyncQueue[PeerId]()
+  let needsChannel = newAsyncQueue[(PeerId, WakuMessageHash)]()
+
+  let recon = (
+    await SyncReconciliation.new(
+      pubsubTopics = @[],
+      contentTopics = @[],
+      peerManager = peerManager,
+      wakuArchive = archive,
+      relayJitter = 0.seconds,
+      idsRx = idsChannel,
+      localWantsTx = wantsChannel,
+      remoteNeedsTx = needsChannel,
+    )
+  ).valueOr:
+    raiseAssert error
+
+  await recon.start()
+  switch.mount(recon)
+
+  let transfer = SyncTransfer.new(
+    peerManager = peerManager,
+    wakuArchive = archive,
+    idsTx = idsChannel,
+    localWantsRx = wantsChannel,
+    remoteNeedsRx = needsChannel,
+  )
+
+  await transfer.start()
+  switch.mount(transfer)
+
+  return FullNode(
+    switch: switch,
+    driver: driver,
+    archive: archive,
+    recon: recon,
+    transfer: transfer,
+    peerInfo: switch.peerInfo.toRemotePeerInfo(),
+    peerManager: peerManager,
+  )
+
+proc stop(node: FullNode) {.async.} =
+  await node.transfer.stop()
+  await node.recon.stop()
+  await node.switch.stop()
+
+proc insertMessage(node: FullNode, msg: WakuMessage) =
+  ## Live-ingest path: relay would archive the message and feed the
+  ## reconciliation storage (subscription_manager archive + sync handlers).
+  discard node.driver.put(DefaultPubsubTopic, @[msg])
+  node.recon.messageIngress(DefaultPubsubTopic, msg)
+
+proc hasMessage(node: FullNode, hash: WakuMessageHash): Future[bool] {.async.} =
+  var query = ArchiveQuery()
+  query.includeData = true
+  query.hashes = @[hash]
+
+  let response = (await node.archive.findMessages(query)).valueOr:
+    raiseAssert $error
+
+  return response.messages.len > 0
+
+suite "Waku Sync: full node miss recovery":
+  var nodeA {.threadvar.}: FullNode
+  var nodeB {.threadvar.}: FullNode
+
+  asyncSetup:
+    nodeA = await newFullNode()
+    nodeB = await newFullNode()
+
+    nodeA.peerManager.addPeer(nodeB.peerInfo)
+    nodeB.peerManager.addPeer(nodeA.peerInfo)
+
+  asyncTeardown:
+    await nodeA.stop()
+    await nodeB.stop()
+
+  asyncTest "node recovers a missed message from a full-node peer":
+    let msg = fakeWakuMessage(contentTopic = DefaultContentTopic)
+    let hash = computeMessageHash(DefaultPubsubTopic, msg)
+
+    nodeA.insertMessage(msg)
+
+    check not await nodeB.hasMessage(hash)
+
+    let res = await nodeB.recon.storeSynchronization(Opt.some(nodeA.peerInfo))
+    assert res.isOk(), $res.error
+
+    # transfer of the missing message happens asynchronously after the
+    # reconciliation session ends
+    await sleepAsync(1.seconds)
+
+    check await nodeB.hasMessage(hash)
+
+  asyncTest "one session converges both nodes to the union":
+    let msgA = fakeWakuMessage(payload = @[byte 1], contentTopic = DefaultContentTopic)
+    let msgB = fakeWakuMessage(payload = @[byte 2], contentTopic = DefaultContentTopic)
+    let hashA = computeMessageHash(DefaultPubsubTopic, msgA)
+    let hashB = computeMessageHash(DefaultPubsubTopic, msgB)
+
+    nodeA.insertMessage(msgA)
+    nodeB.insertMessage(msgB)
+
+    let res = await nodeB.recon.storeSynchronization(Opt.some(nodeA.peerInfo))
+    assert res.isOk(), $res.error
+
+    await sleepAsync(1.seconds)
+
+    check:
+      await nodeA.hasMessage(hashB)
+      await nodeB.hasMessage(hashA)