feat(service-disco): service table bootstrapping (#2718)

Co-authored-by: vladopajic <vladopajic@users.noreply.github.com>

Author: SionoiS

libp2p/protocols/service_discovery.nim

@@ -2,7 +2,7 @@
 # Copyright (c) Status Research & Development GmbH
 
 import chronos, chronicles, results, sets
-import ../utils/heartbeat
+import ../utils/[heartbeat, future]
 import ../[peerid, switch, multihash, peerinfo, extended_peer_record]
 import ./kademlia
 import
@@ -50,6 +50,15 @@ proc maintainServiceTables(
     disco.config.bucketRefreshTime, sleepFirst = true:
     await disco.rtManager.refreshAllTables(disco)
 
+proc bootstrapServiceTable*(
+    disco: ServiceDiscovery, serviceId: ServiceId
+) {.async: (raises: [CancelledError]).} =
+  let rtable = disco.rtManager.getTable(serviceId).valueOr:
+    return
+
+  await disco.refreshTable(rtable, forceRefresh = true)
+  debug "Service table bootstrap complete", serviceId
+
 proc new*(
     T: typedesc[ServiceDiscovery],
     switch: Switch,
@@ -87,6 +96,12 @@ proc new*(
   # Fill up buckets with initial bootstrap nodes
   disco.updatePeers(bootstrapNodes)
 
+  disco.rtManager.onServiceTableCreated = proc(serviceId: ServiceId) =
+    if disco.config.disableBootstrapping:
+      return
+
+    disco.serviceBootstrapFuts.trackFut(disco.bootstrapServiceTable(serviceId))
+
   disco.codec = codec
   if client:
     return disco
@@ -158,6 +173,9 @@ method stop*(disco: ServiceDiscovery) {.async: (raises: []).} =
 
   disco.advertiser.clear()
 
+  disco.serviceBootstrapFuts.cancelSoon()
+  disco.serviceBootstrapFuts = @[]
+
   if not disco.selfSignedPeerRecordLoop.isNil():
     disco.selfSignedPeerRecordLoop.cancelSoon()
     disco.selfSignedPeerRecordLoop = nil

libp2p/protocols/service_discovery/routing_table_manager.nim

@@ -61,6 +61,10 @@ proc addService*(
   manager.serviceStatus[serviceId] = status
 
   manager.updateServiceTablesMetrics()
+
+  if not manager.onServiceTableCreated.isNil:
+    manager.onServiceTableCreated(serviceId)
+
   return true
 
 proc removeService*(

libp2p/protocols/service_discovery/types.nim

@@ -43,6 +43,7 @@ type
   ServiceRoutingTableManager* = ref object
     tables*: Table[ServiceId, RoutingTable]
     serviceStatus*: Table[ServiceId, ServiceStatus]
+    onServiceTableCreated*: proc(serviceId: ServiceId) {.gcsafe, closure, raises: [].}
 
   AdvertisementKey* = tuple[peerId: PeerId, seqNo: uint64]
 
@@ -94,6 +95,7 @@ type
     refreshServiceTablesLoop*: Future[void]
     advertiserMaintenanceLoop*: Future[void]
     localRegistrationLoop*: Future[void]
+    serviceBootstrapFuts*: seq[Future[void]]
     clientMode*: bool
 
 proc new*(

tests/libp2p/service_discovery/test_routing_table_manager.nim

@@ -18,6 +18,11 @@ proc makeKey(x: byte): Key =
 proc makeServiceId(x: byte): ServiceId =
   return makeKey(x)
 
+# Heap-allocated recorder so a closure capturing it stays GC-safe (mirrors how
+# the production callback captures the disco ref rather than a stack local).
+type HitRecorder = ref object
+  ids: seq[ServiceId]
+
 proc makeMainTable(selfId: Key, peers: seq[Key]): RoutingTable =
   var rt = RoutingTable.new(selfId)
   for p in peers:
@@ -75,6 +80,53 @@ suite "ServiceRoutingTableManager":
       upgraded == true
       manager.serviceStatus[serviceId] == Both
 
+  test "addService fires onServiceTableCreated only for brand-new tables":
+    let manager = ServiceRoutingTableManager.new()
+    let mainRt = RoutingTable.new(makeKey(0))
+
+    let hits = HitRecorder()
+    manager.onServiceTableCreated = proc(sid: ServiceId) =
+      hits.ids.add(sid)
+
+    let serviceId = makeServiceId(1)
+
+    check manager.addService(
+      serviceId, mainRt, DefaultReplication, DefaultMaxBuckets, Interest
+    )
+    check:
+      hits.ids.len == 1
+      hits.ids[0] == serviceId
+
+    discard manager.addService(
+      serviceId, mainRt, DefaultReplication, DefaultMaxBuckets, Interest
+    )
+    check hits.ids.len == 1
+
+    discard manager.addService(
+      serviceId, mainRt, DefaultReplication, DefaultMaxBuckets, Provided
+    )
+    check:
+      hits.ids.len == 1
+      manager.serviceStatus[serviceId] == Both
+
+    let otherId = makeServiceId(2)
+    check manager.addService(
+      otherId, mainRt, DefaultReplication, DefaultMaxBuckets, Provided
+    )
+    check:
+      hits.ids.len == 2
+      otherId in hits.ids
+
+  test "addService with no callback set does not crash":
+    let manager = ServiceRoutingTableManager.new()
+    let serviceId = makeServiceId(1)
+    let mainRt = RoutingTable.new(makeKey(0))
+
+    check manager.addService(
+      serviceId, mainRt, DefaultReplication, DefaultMaxBuckets, Interest
+    )
+    check manager.hasService(serviceId)
+
   test "addService pre-populates table from main routing table":
     let selfId = makeKey(0)
     let peer1 = makeKey(1)