feat: message transfer mechanism & tests (#2688)

Makefile

@@ -37,10 +37,10 @@ endif
 ##########
 ## Main ##
 ##########
-.PHONY: all test update clean negentropy
+.PHONY: all test update clean
 
 # default target, because it's the first one that doesn't start with '.'
-all: | negentropy wakunode2 example2 chat2 chat2bridge libwaku
+all: | wakunode2 example2 chat2 chat2bridge libwaku
 
 test: | testcommon testwaku
 
@@ -51,7 +51,7 @@ update: | update-common
 	rm -rf waku.nims && \
 		$(MAKE) waku.nims $(HANDLE_OUTPUT)
 
-clean: | negentropy-clean
+clean:
 	rm -rf build
 
 # must be included after the default target
@@ -84,8 +84,6 @@ endif
 endif
 ## end of Heaptracker options
 
-## Pass libnegentropy to linker.
-NIM_PARAMS := $(NIM_PARAMS) --passL:./libnegentropy.so
 
 ##################
 ## Dependencies ##
@@ -412,9 +410,22 @@ release-notes:
 			sed -E 's@#([0-9]+)@[#\1](https://github.com/waku-org/nwaku/issues/\1)@g'
 # I could not get the tool to replace issue ids with links, so using sed for now,
 # asked here: https://github.com/bvieira/sv4git/discussions/101
+
+######################
+###   NEGENTROPY   ###
+######################
+.PHONY: negentropy
+
+## Pass libnegentropy to linker.
+NIM_PARAMS := $(NIM_PARAMS) --passL:./libnegentropy.so
+
+all: | negentropy
+
+clean: | negentropy-clean
+
 negentropy:
 	$(MAKE) -C vendor/negentropy/cpp && \
 		cp vendor/negentropy/cpp/libnegentropy.so ./
 negentropy-clean:
 	$(MAKE) -C vendor/negentropy/cpp clean && \
-		rm libnegentropy.so
+		rm libnegentropy.so

tests/node/test_wakunode_sync.nim

@@ -0,0 +1,194 @@
+{.used.}
+
+import std/net, testutils/unittests, chronos, libp2p/crypto/crypto
+
+import
+  ../../../waku/
+    [node/waku_node, node/peer_manager, waku_core, waku_store, waku_archive, waku_sync],
+  ../waku_store/store_utils,
+  ../waku_archive/archive_utils,
+  ../testlib/[wakucore, wakunode, testasync]
+
+suite "Store Sync - End to End":
+  var server {.threadvar.}: WakuNode
+  var client {.threadvar.}: WakuNode
+
+  asyncSetup:
+    let timeOrigin = now()
+
+    let messages =
+      @[
+        fakeWakuMessage(@[byte 00], ts = ts(-90, timeOrigin)),
+        fakeWakuMessage(@[byte 01], ts = ts(-80, timeOrigin)),
+        fakeWakuMessage(@[byte 02], ts = ts(-70, timeOrigin)),
+        fakeWakuMessage(@[byte 03], ts = ts(-60, timeOrigin)),
+        fakeWakuMessage(@[byte 04], ts = ts(-50, timeOrigin)),
+        fakeWakuMessage(@[byte 05], ts = ts(-40, timeOrigin)),
+        fakeWakuMessage(@[byte 06], ts = ts(-30, timeOrigin)),
+        fakeWakuMessage(@[byte 07], ts = ts(-20, timeOrigin)),
+        fakeWakuMessage(@[byte 08], ts = ts(-10, timeOrigin)),
+        fakeWakuMessage(@[byte 09], ts = ts(00, timeOrigin)),
+      ]
+
+    let
+      serverKey = generateSecp256k1Key()
+      clientKey = generateSecp256k1Key()
+
+    server = newTestWakuNode(serverKey, IPv4_any(), Port(0))
+    client = newTestWakuNode(clientKey, IPv4_any(), Port(0))
+
+    let serverArchiveDriver = newArchiveDriverWithMessages(DefaultPubsubTopic, messages)
+    let clientArchiveDriver = newArchiveDriverWithMessages(DefaultPubsubTopic, messages)
+
+    let mountServerArchiveRes = server.mountArchive(serverArchiveDriver)
+    let mountClientArchiveRes = client.mountArchive(clientArchiveDriver)
+
+    assert mountServerArchiveRes.isOk()
+    assert mountClientArchiveRes.isOk()
+
+    await server.mountStore()
+    await client.mountStore()
+
+    client.mountStoreClient()
+    server.mountStoreClient()
+
+    let mountServerSync = await server.mountWakuSync(
+      maxFrameSize = 0,
+      syncInterval = 1.hours,
+      relayJitter = 0.seconds,
+      enablePruning = false,
+    )
+    let mountClientSync = await client.mountWakuSync(
+      maxFrameSize = 0,
+      syncInterval = 2.milliseconds,
+      relayJitter = 0.seconds,
+      enablePruning = false,
+    )
+
+    assert mountServerSync.isOk(), mountServerSync.error
+    assert mountClientSync.isOk(), mountClientSync.error
+
+    # messages are retreived when mounting Waku sync
+    # but based on interval so this is needed for client only
+    for msg in messages:
+      client.wakuSync.ingessMessage(DefaultPubsubTopic, msg)
+
+    await allFutures(server.start(), client.start())
+
+    let serverRemotePeerInfo = server.peerInfo.toRemotePeerInfo()
+    let clientRemotePeerInfo = client.peerInfo.toRemotePeerInfo()
+
+    client.peerManager.addServicePeer(serverRemotePeerInfo, WakuSyncCodec)
+    server.peerManager.addServicePeer(clientRemotePeerInfo, WakuSyncCodec)
+
+    client.peerManager.addServicePeer(serverRemotePeerInfo, WakuStoreCodec)
+    server.peerManager.addServicePeer(clientRemotePeerInfo, WakuStoreCodec)
+
+  asyncTeardown:
+    # prevent premature channel shutdown 
+    await sleepAsync(10.milliseconds)
+
+    await allFutures(client.stop(), server.stop())
+
+  asyncTest "no message set differences":
+    check:
+      client.wakuSync.storageSize() == server.wakuSync.storageSize()
+
+    await sleepAsync(10.milliseconds)
+
+    check:
+      client.wakuSync.storageSize() == server.wakuSync.storageSize()
+
+  asyncTest "client message set differences":
+    let msg = fakeWakuMessage(@[byte 10])
+
+    client.wakuSync.ingessMessage(DefaultPubsubTopic, msg)
+    await client.wakuArchive.handleMessage(DefaultPubsubTopic, msg)
+
+    check:
+      client.wakuSync.storageSize() != server.wakuSync.storageSize()
+
+    await sleepAsync(10.milliseconds)
+
+    check:
+      client.wakuSync.storageSize() == server.wakuSync.storageSize()
+
+  asyncTest "server message set differences":
+    let msg = fakeWakuMessage(@[byte 10])
+
+    server.wakuSync.ingessMessage(DefaultPubsubTopic, msg)
+    await server.wakuArchive.handleMessage(DefaultPubsubTopic, msg)
+
+    check:
+      client.wakuSync.storageSize() != server.wakuSync.storageSize()
+
+    await sleepAsync(10.milliseconds)
+
+    check:
+      client.wakuSync.storageSize() == server.wakuSync.storageSize()
+
+suite "Waku Sync - Pruning":
+  var server {.threadvar.}: WakuNode
+  var client {.threadvar.}: WakuNode
+
+  asyncSetup:
+    let
+      serverKey = generateSecp256k1Key()
+      clientKey = generateSecp256k1Key()
+
+    server = newTestWakuNode(serverKey, IPv4_any(), Port(0))
+    client = newTestWakuNode(clientKey, IPv4_any(), Port(0))
+
+    let serverArchiveDriver = newSqliteArchiveDriver()
+    let clientArchiveDriver = newSqliteArchiveDriver()
+
+    let mountServerArchiveRes = server.mountArchive(serverArchiveDriver)
+    let mountClientArchiveRes = client.mountArchive(clientArchiveDriver)
+
+    assert mountServerArchiveRes.isOk()
+    assert mountClientArchiveRes.isOk()
+
+    await server.mountStore()
+    await client.mountStore()
+
+    client.mountStoreClient()
+    server.mountStoreClient()
+
+    let mountServerSync = await server.mountWakuSync(
+      maxFrameSize = 0,
+      relayJitter = 0.seconds,
+      syncInterval = 1.hours,
+      enablePruning = false,
+    )
+    let mountClientSync = await client.mountWakuSync(
+      maxFrameSize = 0,
+      syncInterval = 10.milliseconds,
+      relayJitter = 0.seconds,
+      enablePruning = true,
+    )
+
+    assert mountServerSync.isOk(), mountServerSync.error
+    assert mountClientSync.isOk(), mountClientSync.error
+
+    await allFutures(server.start(), client.start())
+
+  asyncTeardown:
+    await sleepAsync(10.milliseconds)
+
+    await allFutures(client.stop(), server.stop())
+
+  asyncTest "pruning":
+    for _ in 0 ..< 4:
+      for _ in 0 ..< 10:
+        let msg = fakeWakuMessage()
+        client.wakuSync.ingessMessage(DefaultPubsubTopic, msg)
+        await client.wakuArchive.handleMessage(DefaultPubsubTopic, msg)
+
+        server.wakuSync.ingessMessage(DefaultPubsubTopic, msg)
+        await server.wakuArchive.handleMessage(DefaultPubsubTopic, msg)
+
+      await sleepAsync(10.milliseconds)
+
+    check:
+      client.wakuSync.storageSize() == 10
+      server.wakuSync.storageSize() == 40

tests/waku_sync/sync_utils.nim

@@ -5,14 +5,23 @@ import std/options, chronos, chronicles, libp2p/crypto/crypto
 import ../../../waku/[node/peer_manager, waku_core, waku_sync], ../testlib/wakucore
 
 proc newTestWakuSync*(
-    switch: Switch, handler: SyncCallback
+    switch: Switch,
+    transfer: Option[TransferCallback] = none(TransferCallback),
+    prune: Option[PruneCallback] = none(PruneCallback),
+    interval: Duration = DefaultSyncInterval,
 ): Future[WakuSync] {.async.} =
-  let
-    peerManager = PeerManager.new(switch)
-    proto = WakuSync.new(
-      peerManager = peerManager, relayJitter = 0.seconds, syncCB = some(handler)
-    )
-  assert proto != nil
+  let peerManager = PeerManager.new(switch)
+
+  let res = await WakuSync.new(
+    peerManager = peerManager,
+    relayJitter = 0.seconds,
+    syncInterval = interval,
+    pruning = false,
+    wakuArchive = nil,
+    wakuStoreClient = nil,
+  )
+
+  let proto = res.get()
 
   proto.start()
   switch.mount(proto)

tests/waku_sync/test_all.nim

@@ -1,4 +1,3 @@
 {.used.}
 
-import 
-  ./test_protocol
+import ./test_protocol, ./test_bindings