Ping every connection until traffic starts (#43)

quic closes a connection after 30s of silence and there is nothing to gossip about
before the first message, so an idle cold start dropped every connection outside the
mesh. Ping now covers every connection at 20s, under the timeout, and stops once 
messages arrive because gossip reaches each peer every few seconds by then.

nim-test-node/regression/main.nim

@@ -35,6 +35,8 @@ proc createMessageHandler(): proc(topic: string, data: seq[byte]) {.async, gcsaf
     # warm-up
     if timestampNs < 1000000: return
 
+    notePublishingStarted()
+
     # Log received message
     info "Received message",
       msgId = msgId,
@@ -202,6 +204,12 @@ proc main {.async.} =
 
   await switch.start()
 
+  # Before the staggered bootstrap dial below, not after. A node's mesh arrives on
+  # inbound dials from nodes that bootstrapped earlier, so it can publish long before
+  # its own dial returns; gating :8645 on that left half the fleet refusing publishes.
+  info "Starting listening endpoint for publish controller"
+  discard gossipSub.startHttpServer(myId)
+
   info "Starting metrics server"
   let metricsServer = startMetricsServer(parseIpAddress("0.0.0.0"), prometheusPort)
   if metricsServer.isErr:
@@ -233,11 +241,9 @@ proc main {.async.} =
   info "Mesh details ", meshSize = gossipSub.mesh.getOrDefault("test").len,
     peersConnected = gossipSub.gossipsub.getOrDefault("test").len
 
-  # Periodic mesh-only pings
-  asyncSpawn pingMeshLoop(switch, pingProtocol, gossipSub, "test")
+  # Hold connections open until gossipsub traffic takes over
+  asyncSpawn pingLoop(switch, pingProtocol)
 
-  info "Starting listening endpoint for publish controller"
-  discard gossipSub.startHttpServer(myId)
 
   await sleepAsync(2.days)
 

nim-test-node/regression/ping_utils.nim

@@ -1,26 +1,38 @@
 import chronos
-import std/[tables, sets]
+import std/[random, tables]
 
 import libp2p
 import libp2p/protocols/ping
-import libp2p/protocols/pubsub/gossipsub
 
 const
-  MeshPingInterval = 45.seconds
-  MeshPingTimeout  = 4.seconds
-  MeshDialTimeout  = 4.seconds
-  MeshCloseTimeout = 2.seconds
-  SlowDialLog      = 500.milliseconds
-  SlowCloseLog     = 500.milliseconds
+  PingInterval  = 12.seconds
+    ## A full sweep in batches takes a few seconds, so the gap between two pings of the
+    ## same peer is the interval plus a sweep. Keep that comfortably under quic's 30s
+    ## idle timeout.
+  PingBatch     = 16
+    ## Dials in flight at once. All ~250 together stalls the node's own publish
+    ## endpoint, which cost 34 injections in the first minute of a 1000-node run.
+  PingBatchGap  = 200.milliseconds
+  PingTimeout   = 4.seconds
+  DialTimeout   = 4.seconds
+  CloseTimeout  = 2.seconds
+  SlowDialLog   = 500.milliseconds
+  SlowCloseLog  = 500.milliseconds
 
-proc meshPeerIds*(gossipSub: GossipSub, topic: string): seq[PeerId] =
-  ## Only peers in the mesh for `topic` (convert HashSet[PubSubPeer] -> seq[PeerId])
-  let peers: HashSet[PubSubPeer] = gossipSub.mesh.getOrDefault(topic, initHashSet[PubSubPeer]())
-  result = newSeqOfCap[PeerId](peers.len)
-  for p in peers:
-    result.add(p.peerId)
+var messagesStarted = false
 
-proc pingMeshPeer*(switch: Switch, pingProtocol: Ping, peerId: PeerId) {.async.} =
+proc notePublishingStarted*() =
+  ## Called from the message handler; stops the ping loop.
+  messagesStarted = true
+
+proc connectedPeerIds*(switch: Switch): seq[PeerId] =
+  ## Every peer we hold a connection to, not just the gossipsub mesh.
+  let conns = switch.connManager.getConnections()
+  result = newSeqOfCap[PeerId](conns.len)
+  for peerId in conns.keys:
+    result.add(peerId)
+
+proc pingPeer*(switch: Switch, pingProtocol: Ping, peerId: PeerId) {.async.} =
   let book = switch.peerStore[AddressBook]
 
   if not book.book.hasKey(peerId):
@@ -30,20 +42,20 @@ proc pingMeshPeer*(switch: Switch, pingProtocol: Ping, peerId: PeerId) {.async.}
   if addrs.len == 0:
     return
 
-  var stream: Stream
+  var stream: Connection
   let dialStart = Moment.now()
   try:
-    stream = await switch.dial(peerId, addrs, PingCodec).wait(MeshDialTimeout)
+    stream = await switch.dial(peerId, addrs, PingCodec).wait(DialTimeout)
 
     let dialDur = Moment.now() - dialStart
     if dialDur >= SlowDialLog:
-      warn "mesh ping: slow dial", peerId = peerId, dialMs = dialDur.milliseconds
+      warn "keepalive ping: slow dial", peerId = peerId, dialMs = dialDur.milliseconds
 
     let pingStart = Moment.now()
-    let latency = await pingProtocol.ping(stream).wait(MeshPingTimeout)
+    let latency = await pingProtocol.ping(stream).wait(PingTimeout)
     let pingDur = Moment.now() - pingStart
 
-    info "mesh ping",
+    trace "keepalive ping",
       peerId = peerId,
       latency = latency,
       dialMs = dialDur.milliseconds,
@@ -52,36 +64,52 @@ proc pingMeshPeer*(switch: Switch, pingProtocol: Ping, peerId: PeerId) {.async.}
     raise exc
   except CatchableError as exc:
     let dialDur = Moment.now() - dialStart
-    warn "mesh ping failed", peerId = peerId, error = exc.msg, dialMs = dialDur.milliseconds
+    warn "keepalive ping failed", peerId = peerId, error = exc.msg, dialMs = dialDur.milliseconds
   finally:
     if not stream.isNil and not stream.closed:
       let closeStart = Moment.now()
       try:
         # Make close observable: if it never completes, you'll never see "slow close" today.
-        await stream.closeWithEOF().wait(MeshCloseTimeout)
+        await stream.closeWithEOF().wait(CloseTimeout)
       except CancelledError as exc:
         raise exc
       except CatchableError as exc:
-        warn "mesh ping: stream close failed", peerId = peerId, error = exc.msg
+        warn "keepalive ping: stream close failed", peerId = peerId, error = exc.msg
       finally:
         let closeDur = Moment.now() - closeStart
         if closeDur >= SlowCloseLog:
-          warn "mesh ping: slow close", peerId = peerId, closeMs = closeDur.milliseconds
+          warn "keepalive ping: slow close", peerId = peerId, closeMs = closeDur.milliseconds
 
-proc pingMeshOnce*(switch: Switch, pingProtocol: Ping, gossipSub: GossipSub, topic: string) {.async.} =
-  let peers = gossipSub.meshPeerIds(topic)
+proc pingAllOnce*(switch: Switch, pingProtocol: Ping) {.async.} =
+  var peers: seq[PeerId] = @[]
+  for pid in switch.connectedPeerIds():
+    if pid != switch.peerInfo.peerId:
+      peers.add(pid)
   if peers.len == 0:
     return
 
-  var futs: seq[Future[void]] = @[]
-  for pid in peers:
-    if pid == switch.peerInfo.peerId: continue
-    futs.add(switch.pingMeshPeer(pingProtocol, pid))
-
-  if futs.len > 0:
+  var i = 0
+  while i < peers.len:
+    if messagesStarted:
+      return
+    var futs: seq[Future[void]] = @[]
+    for pid in peers[i ..< min(i + PingBatch, peers.len)]:
+      futs.add(switch.pingPeer(pingProtocol, pid))
     await allFutures(futs)
+    i += PingBatch
+    if i < peers.len:
+      await sleepAsync(PingBatchGap)
 
-proc pingMeshLoop*(switch: Switch, pingProtocol: Ping, gossipSub: GossipSub, topic: string) {.async.} =
-  while true:
-    await switch.pingMeshOnce(pingProtocol, gossipSub, topic)
-    await sleepAsync(MeshPingInterval)
+proc pingLoop*(switch: Switch, pingProtocol: Ping) {.async.} =
+  ## Hold every connection open until traffic starts doing it for us.
+  ##
+  ## quic closes a connection after 30s of silence, and before the first message there
+  ## is nothing to gossip about, so an idle cold start drops every connection that is
+  ## not in the mesh. Once messages flow, gossip reaches each peer every few seconds and
+  ## the ping is redundant, so it stops.
+  # Stagger the start so a thousand nodes do not ping in lockstep.
+  await sleepAsync(rand(PingInterval.milliseconds.int).milliseconds)
+  while not messagesStarted:
+    await switch.pingAllOnce(pingProtocol)
+    await sleepAsync(PingInterval)
+  info "keepalive ping stopped, gossipsub traffic now keeps connections open"

nim-test-node/regression/test_node.nimble

@@ -11,6 +11,6 @@ skipDirs      = @[]
 
 requires "nim >= 2.2.0",
           "nimcrypto 0.6.4",
-          "https://github.com/vacp2p/nim-libp2p#26e181e4dd65188051ab4783bcc538ed9579644f", # 2.0.0
+          "https://github.com/vacp2p/nim-libp2p#d721f9cb13c36b844957556ac04fecade371b059", # release/v2.2.0
           # The shadow image's tick-floor patch targets lsquic 0.5.4's layout.
           "lsquic >= 0.5.4 & < 0.5.5"