fix: re-advertise provider records when addresses change (#1493)

Signed-off-by: Chrysostomos Nanakos <chris@include.gr>

storage/blockexchange/engine/advertiser.nim

@@ -45,6 +45,7 @@ type Advertiser* = ref object of RootObj
 
   advertiseLocalStoreLoopSleep: Duration # Advertise loop sleep
   inFlightAdvReqs*: Table[Cid, Future[void]] # Inflight advertise requests
+  addrChanged: AsyncEvent # Fired when the announced addresses change
 
 proc addCidToQueue(b: Advertiser, cid: Cid) {.async: (raises: [CancelledError]).} =
   if cid notin b.advertiseQueue:
@@ -70,6 +71,8 @@ proc advertiseBlock(b: Advertiser, cid: Cid) {.async: (raises: [CancelledError])
 proc advertiseLocalStoreLoop(b: Advertiser) {.async: (raises: []).} =
   try:
     while b.advertiserRunning:
+      b.addrChanged.clear()
+
       if cidsIter =? await b.localStore.listBlocks(blockType = BlockType.Manifest):
         trace "Advertiser begins iterating blocks..."
         for c in cidsIter:
@@ -77,7 +80,7 @@ proc advertiseLocalStoreLoop(b: Advertiser) {.async: (raises: []).} =
             await b.advertiseBlock(cid)
         trace "Advertiser iterating blocks finished."
 
-      await sleepAsync(b.advertiseLocalStoreLoopSleep)
+      discard await b.addrChanged.wait().withTimeout(b.advertiseLocalStoreLoopSleep)
   except CancelledError:
     warn "Cancelled advertise local store loop"
 
@@ -125,6 +128,8 @@ proc start*(b: Advertiser) {.async: (raises: []).} =
   b.localStore.onBlockStored = onBlock.some
 
   b.advertiserRunning = true
+  b.discovery.onAddrChange = proc() {.gcsafe, raises: [].} =
+    b.addrChanged.fire()
   for i in 0 ..< b.concurrentAdvReqs:
     let fut = b.processQueueLoop()
     b.trackedFutures.track(fut)
@@ -144,6 +149,7 @@ proc stop*(b: Advertiser) {.async: (raises: []).} =
   b.advertiserRunning = false
   # Stop incoming tasks from callback and localStore loop
   b.localStore.onBlockStored = CidCallback.none
+  b.discovery.onAddrChange = nil
   trace "Stopping advertise loop and tasks"
   await b.trackedFutures.cancelTracked()
   trace "Advertiser loop and tasks stopped"
@@ -165,4 +171,5 @@ proc new*(
     trackedFutures: TrackedFutures.new(),
     inFlightAdvReqs: initTable[Cid, Future[void]](),
     advertiseLocalStoreLoopSleep: advertiseLocalStoreLoopSleep,
+    addrChanged: newAsyncEvent(),
   )

storage/discovery.nim

@@ -54,6 +54,7 @@ type Discovery* = ref object of RootObj
   mixProto*: MixProtocol
   dhtMixProxies*: seq[SignedPeerRecord]
   privateQueries: bool
+  onAddrChange*: proc() {.gcsafe, raises: [].}
 
 proc toNodeId*(cid: Cid): NodeId =
   ## Cid to discovery id
@@ -222,6 +223,7 @@ proc announceDirectAddrs*(
 ) =
   # UDP addresses are derived from TCP addresses by remapping protocol and port.
   let tcpAddrs = @providerAddrs
+  let addrsChanged = tcpAddrs.len > 0 and tcpAddrs.sorted() != d.providerAddrs.sorted()
   let udpAddrs =
     tcpAddrs.mapIt(it.remapAddr(protocol = some("udp"), port = some(udpPort)))
 
@@ -242,9 +244,13 @@ proc announceDirectAddrs*(
       .expect("Should construct signed record").some
     d.protocol.updateRecord(spr).expect("Should update SPR")
 
+  if addrsChanged and not d.onAddrChange.isNil:
+    d.onAddrChange()
+
 proc announceRelayAddrs*(d: Discovery, addrs: openArray[MultiAddress]) =
   ## Updates the announce addresses and the SPR with the relay circuit addresses.
   ## Unlike announceDirectAddrs, no UDP address is derived so discoveryAddrs is left untouched.
+  let addrsChanged = addrs.len > 0 and addrs.sorted() != d.providerAddrs.sorted()
   d.providerAddrs = @addrs
 
   info "Updating announce record", addrs = d.providerAddrs
@@ -256,6 +262,9 @@ proc announceRelayAddrs*(d: Discovery, addrs: openArray[MultiAddress]) =
   info "Provider record updated",
     addrs = d.providerAddrs, spr = d.providerRecord.get.toURI
 
+  if addrsChanged and not d.onAddrChange.isNil:
+    d.onAddrChange()
+
 proc start*(d: Discovery) {.async: (raises: []).} =
   try:
     d.protocol.open()