Improve rdv advertise (#951)

Co-authored-by: Ludovic Chenut <[email protected]>

libp2p/discovery/discoverymngr.nim

@@ -122,20 +122,15 @@ proc request*[T](dm: DiscoveryManager, value: T): DiscoveryQuery =
   pa.add(value)
   return dm.request(pa)
 
-proc advertise*(dm: DiscoveryManager, pa: PeerAttributes) =
+proc advertise*[T](dm: DiscoveryManager, value: T) =
   for i in dm.interfaces:
-    i.toAdvertise = pa
+    i.toAdvertise.add(value)
     if i.advertiseLoop.isNil:
       i.advertisementUpdated = newAsyncEvent()
       i.advertiseLoop = i.advertise()
     else:
       i.advertisementUpdated.fire()
 
-proc advertise*[T](dm: DiscoveryManager, value: T) =
-  var pa: PeerAttributes
-  pa.add(value)
-  dm.advertise(pa)
-
 template forEach*(query: DiscoveryQuery, code: untyped) =
   ## Will execute `code` for each discovered peer. The
   ## peer attritubtes are available through the variable

libp2p/discovery/rendezvousinterface.nim

@@ -19,6 +19,7 @@ type
     rdv*: RendezVous
     timeToRequest: Duration
     timeToAdvertise: Duration
+    ttl: Duration
 
   RdvNamespace* = distinct string
 
@@ -62,12 +63,16 @@ method advertise*(self: RendezVousInterface) {.async.} =
 
     self.advertisementUpdated.clear()
     for toAdv in toAdvertise:
-      await self.rdv.advertise(toAdv, self.timeToAdvertise)
+      try:
+        await self.rdv.advertise(toAdv, self.ttl)
+      except CatchableError as error:
+        debug "RendezVous advertise error: ", msg = error.msg
 
     await sleepAsync(self.timeToAdvertise) or self.advertisementUpdated.wait()
 
 proc new*(T: typedesc[RendezVousInterface],
           rdv: RendezVous,
           ttr: Duration = 1.minutes,
-          tta: Duration = MinimumDuration): RendezVousInterface =
-  T(rdv: rdv, timeToRequest: ttr, timeToAdvertise: tta)
+          tta: Duration = 1.minutes,
+          ttl: Duration = MinimumDuration): RendezVousInterface =
+  T(rdv: rdv, timeToRequest: ttr, timeToAdvertise: tta, ttl: ttl)

libp2p/protocols/rendezvous.nim

@@ -469,16 +469,18 @@ proc advertisePeer(rdv: RendezVous,
         trace "Unexpected register response", peer, msgType = msgRecv.msgType
       elif msgRecv.registerResponse.tryGet().status != ResponseStatus.Ok:
         trace "Refuse to register", peer, response = msgRecv.registerResponse
+      else:
+        trace "Successfully registered", peer, response = msgRecv.registerResponse
     except CatchableError as exc:
       trace "exception in the advertise", error = exc.msg
     finally:
       rdv.sema.release()
   await rdv.sema.acquire()
   discard await advertiseWrap().withTimeout(5.seconds)
 
-proc advertise*(rdv: RendezVous,
+method advertise*(rdv: RendezVous,
                 ns: string,
-                ttl: Duration = MinimumDuration) {.async.} =
+                ttl: Duration = MinimumDuration) {.async, base.} =
   let sprBuff = rdv.switch.peerInfo.signedPeerRecord.encode().valueOr:
     raise newException(RendezVousError, "Wrong Signed Peer Record")
   if ns.len notin 1..255:

tests/testrendezvous.nim

@@ -14,6 +14,7 @@ import chronos
 import ../libp2p/[protocols/rendezvous,
                   switch,
                   builders,]
+import ../libp2p/discovery/[rendezvousinterface, discoverymngr]
 import ./helpers
 
 proc createSwitch(rdv: RendezVous = RendezVous.new()): Switch =

tests/testrendezvousinterface.nim

@@ -0,0 +1,73 @@
+{.used.}
+
+# Nim-Libp2p
+# Copyright (c) 2023 Status Research & Development GmbH
+# Licensed under either of
+#  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
+#  * MIT license ([LICENSE-MIT](LICENSE-MIT))
+# at your option.
+# This file may not be copied, modified, or distributed except according to
+# those terms.
+
+import sequtils, strutils
+import chronos
+import ../libp2p/[protocols/rendezvous,
+                  switch,
+                  builders,]
+import ../libp2p/discovery/[rendezvousinterface, discoverymngr]
+import ./helpers
+
+proc createSwitch(rdv: RendezVous = RendezVous.new()): Switch =
+  SwitchBuilder.new()
+    .withRng(newRng())
+    .withAddresses(@[ MultiAddress.init("/ip4/0.0.0.0/tcp/0").tryGet() ])
+    .withTcpTransport()
+    .withMplex()
+    .withNoise()
+    .withRendezVous(rdv)
+    .build()
+
+type
+  MockRendezVous = ref object of RendezVous
+    numAdvertiseNs1: int
+    numAdvertiseNs2: int
+
+  MockErrorRendezVous = ref object of MockRendezVous
+
+method advertise*(self: MockRendezVous, namespace: string, ttl: Duration) {.async.} =
+  if namespace == "ns1":
+    self.numAdvertiseNs1 += 1
+  elif namespace == "ns2":
+    self.numAdvertiseNs2 += 1
+  # Forward the call to the actual implementation
+  await procCall RendezVous(self).advertise(namespace, ttl)
+
+method advertise*(self: MockErrorRendezVous, namespace: string, ttl: Duration) {.async.} =
+  await procCall MockRendezVous(self).advertise(namespace, ttl)
+  raise newException(CatchableError, "MockErrorRendezVous.advertise")
+
+suite "RendezVous Interface":
+  teardown:
+    checkTrackers()
+
+  proc baseTimeToAdvertiseTest(rdv: MockRendezVous) {.async.} =
+    let
+      tta = 100.milliseconds
+      ttl = 2.hours
+      client = createSwitch(rdv)
+      dm = DiscoveryManager()
+
+    await client.start()
+    dm.add(RendezVousInterface.new(rdv = rdv, tta = tta, ttl = ttl))
+    dm.advertise(RdvNamespace("ns1"))
+    dm.advertise(RdvNamespace("ns2"))
+
+    checkExpiring: rdv.numAdvertiseNs1 >= 5
+    checkExpiring: rdv.numAdvertiseNs2 >= 5
+    await client.stop()
+
+  asyncTest "Check timeToAdvertise interval":
+    await baseTimeToAdvertiseTest(MockRendezVous.new(newRng()))
+
+  asyncTest "Check timeToAdvertise interval when there is an error":
+    await baseTimeToAdvertiseTest(MockErrorRendezVous.new(newRng()))