feat: negentropy c integ (#2448)

* chore: move negentropy to a submodule

* chore: add negentropy folder in vendor dir

* moved submodule to c-wrapper branch

* chore: updated negentropy

* chore: udpate submodule URL to use https

* chore: started integrating negetropy C wrapper

* chore: fixed all compilation errors wrt C-wrapper integration

* chore: include sync peers to be returned as part of REST API

* chore: tested insert into storage and changes done for it.

* chore: experimenting with callback

* chore: first test for sync

* chore: revert callback changes

* chore: revert temp changes

* chore: write tests to verify c integration

* draft: in progress changes to integrate callback based method from C

* chore: in progress callback integration

* chore: first working sync example with c bindings

* feat: added few tests for sync protocol

* chore: copy negentropy so for build to work

* chore: add negentropy as dependency for test targets

* chore: try to fix CI compilation issue of negentropy

* chore: apply suggestions from code review

Co-authored-by: Ivan FB <128452529+Ivansete-status@users.noreply.github.com>

* chore: fix naming convention changes

---------

Co-authored-by: Ivan FB <128452529+Ivansete-status@users.noreply.github.com>

.gitmodules

@@ -164,3 +164,8 @@
 	branch = master
 	path = vendor/nim-results
 	url = https://github.com/arnetheduck/nim-results.git
+[submodule "vendor/negentropy"]
+	ignore = untracked
+	branch = feat/c-wrapper
+	path = vendor/negentropy
+	url = https://github.com/waku-org/negentropy.git

Makefile

@@ -32,10 +32,10 @@ else # "variables.mk" was included. Business as usual until the end of this file
 ##########
 ## Main ##
 ##########
-.PHONY: all test update clean
+.PHONY: all test update clean negentropy
 
 # default target, because it's the first one that doesn't start with '.'
-all: | wakunode2 example2 chat2 chat2bridge libwaku
+all: | negentropy wakunode2 example2 chat2 chat2bridge libwaku
 
 test: | testcommon testwaku
 
@@ -46,7 +46,7 @@ update: | update-common
 	rm -rf waku.nims && \
 		$(MAKE) waku.nims $(HANDLE_OUTPUT)
 
-clean:
+clean: | negentropy-clean
 	rm -rf build
 
 # must be included after the default target
@@ -166,15 +166,15 @@ testcommon: | build deps
 ##########
 .PHONY: testwaku wakunode2 testwakunode2 example2 chat2 chat2bridge
 
-testwaku: | build deps librln
+testwaku: | build deps librln negentropy
 	echo -e $(BUILD_MSG) "build/$@" && \
 		$(ENV_SCRIPT) nim test -d:os=$(shell uname) $(NIM_PARAMS) waku.nims
 
-wakunode2: | build deps librln
+wakunode2: | build deps librln negentropy
 	echo -e $(BUILD_MSG) "build/$@" && \
 		$(ENV_SCRIPT) nim wakunode2 $(NIM_PARAMS) waku.nims
 
-testwakunode2: | build deps librln
+testwakunode2: | build deps librln negentropy
 	echo -e $(BUILD_MSG) "build/$@" && \
 		$(ENV_SCRIPT) nim testwakunode2 $(NIM_PARAMS) waku.nims
 
@@ -314,3 +314,9 @@ 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:
+	$(MAKE) -C vendor/negentropy/cpp && \
+		cp vendor/negentropy/cpp/libnegentropy.so ./
+negentropy-clean:
+	$(MAKE) -C vendor/negentropy/cpp clean && \
+		rm libnegentropy.so

apps/wakunode2/app.nim

@@ -508,6 +508,11 @@ proc setupProtocols(node: WakuNode,
     except CatchableError:
       return err("failed to mount waku store protocol: " & getCurrentExceptionMsg())
 
+    # Waku Sync setup
+    node.mountWakuSync().isOkOr:
+      return err("failed to mount waku sync protocol: " & error)
+
+
   mountStoreClient(node)
   if conf.storenode != "":
     let storeNode = parsePeerInfo(conf.storenode)

tests/waku_sync/sync_utils.nim

@@ -0,0 +1,29 @@
+{.used.}
+
+import
+  std/options,
+  chronos,
+  chronicles,
+  libp2p/crypto/crypto
+
+import
+  ../../../waku/[
+    node/peer_manager,
+    waku_core,
+    waku_sync,
+  ],
+  ../testlib/[
+    common,
+    wakucore
+  ]
+
+proc newTestWakuSync*(switch: Switch, handler: WakuSyncCallback): Future[WakuSync] {.async.} =
+  const DefaultFrameSize = 153600
+  let
+    peerManager = PeerManager.new(switch)
+    proto = WakuSync.new(peerManager, DefaultFrameSize, 2.seconds, some(handler))
+
+  proto.start()
+  switch.mount(proto)
+
+  return proto

tests/waku_sync/test_all.nim

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

tests/waku_sync/test_protocol.nim

@@ -0,0 +1,141 @@
+{.used.}
+
+import
+  std/options,
+  testutils/unittests,
+  chronos,
+  chronicles,
+  libp2p/crypto/crypto,
+  stew/byteutils
+from std/os import sleep
+
+import
+  ../../../waku/[
+    common/paging,
+    node/peer_manager,
+    waku_core,
+    waku_core/message/digest,
+    waku_sync,
+    waku_sync/raw_bindings,
+  ],
+  ../testlib/[
+    common,
+    wakucore
+  ],
+  ./sync_utils
+
+
+suite "Waku Sync - Protocol Tests":
+
+    asyncTest "test c integration":
+        let 
+            s1 = negentropyNewStorage()
+            s2 = negentropyNewStorage()
+            ng1 = negentropyNew(s1,10000)
+            ng2 = negentropyNew(s2,10000)
+
+        let msg1 = fakeWakuMessage(contentTopic=DefaultContentTopic)
+        let msgHash: WakuMessageHash = computeMessageHash(pubsubTopic=DefaultPubsubTopic, msg1)
+        var ret = negentropyStorageInsert(s1, msg1.timestamp, msgHash)
+        check:
+          ret == true
+
+        ret = negentropyStorageInsert(s2, msg1.timestamp, msgHash)
+        check:
+          ret == true
+
+        let msg2 = fakeWakuMessage(contentTopic=DefaultContentTopic)
+        let msgHash2: WakuMessageHash = computeMessageHash(pubsubTopic=DefaultPubsubTopic, msg2)
+        ret = negentropyStorageInsert(s2, msg2.timestamp, msgHash2)
+        check:
+          ret == true
+
+        let ng1_q1 = negentropyInitiate(ng1)
+        check:
+          ng1_q1.len > 0
+
+        let ng2_q1 = negentropyServerReconcile(ng2, ng1_q1)
+        check:
+          ng2_q1.len > 0
+
+        var
+          haveHashes: seq[WakuMessageHash]
+          needHashes: seq[WakuMessageHash]
+        let ng1_q2 = negentropyClientReconcile(ng1, ng2_q1, haveHashes, needHashes)
+
+        check:
+          needHashes.len() == 1
+          haveHashes.len() == 0
+          ng1_q2.len == 0
+          needHashes[0] == msgHash2
+
+        ret = negentropyStorageErase(s1, msg1.timestamp, msgHash)
+        check:
+          ret == true
+
+    asyncTest "sync 2 nodes different hashes":
+        ## Setup
+        let
+            serverSwitch = newTestSwitch()
+            clientSwitch = newTestSwitch()
+
+        await allFutures(serverSwitch.start(), clientSwitch.start())
+
+        let serverPeerInfo = serverSwitch.peerInfo.toRemotePeerInfo()
+        let msg1 = fakeWakuMessage(contentTopic=DefaultContentTopic)
+        let msg2 = fakeWakuMessage(contentTopic=DefaultContentTopic)
+
+        let protoHandler:WakuSyncCallback = proc(hashes: seq[WakuMessageHash]) {.async: (raises: []), closure, gcsafe.} = 
+            debug "Received needHashes from peer:", len = hashes.len
+            for hash in hashes:
+                debug "Hash received from peer:", hash=hash.to0xHex()
+
+        let
+            server = await newTestWakuSync(serverSwitch, handler=protoHandler)
+            client = await newTestWakuSync(clientSwitch, handler=protoHandler)
+        server.ingessMessage(DefaultPubsubTopic, msg1)
+        client.ingessMessage(DefaultPubsubTopic, msg1)
+        server.ingessMessage(DefaultPubsubTopic, msg2)
+
+        var hashes = await client.sync(serverPeerInfo)
+        require (hashes.isOk())
+        check:
+          hashes.value.len == 1
+          hashes.value[0] == computeMessageHash(pubsubTopic=DefaultPubsubTopic, msg2)
+        #Assuming message is fetched from peer
+#[         client.ingessMessage(DefaultPubsubTopic, msg2)
+        sleep(1000)
+        hashes = await client.sync(serverPeerInfo)
+        require (hashes.isOk())
+        check:
+          hashes.value.len == 0 ]#
+
+    asyncTest "sync 2 nodes same hashes":
+        ## Setup
+        let
+            serverSwitch = newTestSwitch()
+            clientSwitch = newTestSwitch()
+
+        await allFutures(serverSwitch.start(), clientSwitch.start())
+
+        let serverPeerInfo = serverSwitch.peerInfo.toRemotePeerInfo()
+        let msg1 = fakeWakuMessage(contentTopic=DefaultContentTopic)
+        let msg2 = fakeWakuMessage(contentTopic=DefaultContentTopic)
+
+        let protoHandler:WakuSyncCallback = proc(hashes: seq[WakuMessageHash]) {.async: (raises: []), closure, gcsafe.} = 
+            debug "Received needHashes from peer:", len = hashes.len
+            for hash in hashes:
+                debug "Hash received from peer:", hash=hash.to0xHex()
+
+        let
+            server = await newTestWakuSync(serverSwitch, handler=protoHandler)
+            client = await newTestWakuSync(clientSwitch, handler=protoHandler)
+        server.ingessMessage(DefaultPubsubTopic, msg1)
+        client.ingessMessage(DefaultPubsubTopic, msg1)
+        server.ingessMessage(DefaultPubsubTopic, msg2)
+        client.ingessMessage(DefaultPubsubTopic, msg2)
+
+        let hashes = await client.sync(serverPeerInfo)
+        assert hashes.isOk(), $hashes.error
+        check:
+          hashes.value.len == 0

waku/waku_api/rest/admin/handlers.nim

@@ -19,6 +19,7 @@ import
   ../../../waku_lightpush/common,
   ../../../waku_relay,
   ../../../waku_node,
+  ../../../waku_sync,
   ../../../node/peer_manager,
   ../responses,
   ../serdes,
@@ -89,6 +90,15 @@ proc installAdminV1GetPeersHandler(router: var RestRouter, node: WakuNode) =
                                    connected: it.connectedness == Connectedness.Connected))
       tuplesToWakuPeers(peers, lightpushPeers)
 
+    if not node.wakuSync.isNil():
+      # Map WakuStore peers to WakuPeers and add to return list
+      let syncPeers = node.peerManager.peerStore
+                           .peers(WakuSyncCodec)
+                           .mapIt((multiaddr: constructMultiaddrStr(it),
+                                   protocol: WakuSyncCodec,
+                                   connected: it.connectedness == Connectedness.Connected))
+      tuplesToWakuPeers(peers, syncPeers)
+
     let resp = RestApiResponse.jsonResponse(peers, status=Http200)
     if resp.isErr():
       error "An error ocurred while building the json respose: ", error=resp.error