diff --git a/logos_delivery/waku/node/peer_manager/peer_manager.nim b/logos_delivery/waku/node/peer_manager/peer_manager.nim index 31c6133ab..182674068 100644 --- a/logos_delivery/waku/node/peer_manager/peer_manager.nim +++ b/logos_delivery/waku/node/peer_manager/peer_manager.nim @@ -687,30 +687,6 @@ proc connectToRelayPeers*(pm: PeerManager) {.async.} = index += numPeersToConnect numPendingConnReqs -= numPeersToConnect -proc reconnectPeers*( - pm: PeerManager, proto: string, backoffTime: chronos.Duration = chronos.seconds(0) -) {.async.} = - ## Reconnect to peers registered for this protocol. This will update connectedness. - ## Especially useful to resume connections from persistent storage after a restart. - - info "Reconnecting peers", proto = proto - - # Proto is not persisted, we need to iterate over all peers. - for peerInfo in pm.switch.peerStore.peers(protocolMatcher(proto)): - # Check that the peer can be connected - if peerInfo.connectedness == CannotConnect: - error "Not reconnecting to unreachable or non-existing peer", - peerId = peerInfo.peerId - continue - - if backoffTime > ZeroDuration: - info "Backing off before reconnect", - peerId = peerInfo.peerId, backoffTime = backoffTime - # We disconnected recently and still need to wait for a backoff period before connecting - await sleepAsync(backoffTime) - - await pm.connectToNodes(@[peerInfo]) - proc getNumStreams*(pm: PeerManager, protocol: string): (int, int) = var numStreamsIn = 0 diff --git a/logos_delivery/waku/node/waku_node.nim b/logos_delivery/waku/node/waku_node.nim index db90ab47c..459105ac4 100644 --- a/logos_delivery/waku/node/waku_node.nim +++ b/logos_delivery/waku/node/waku_node.nim @@ -136,7 +136,6 @@ type wakuMix*: WakuMix wakuKademlia*: WakuKademlia ports*: BoundPorts - relayReconnectFut*: Future[void] SubscriptionManager* = ref object of RootObj node*: WakuNode @@ -405,17 +404,6 @@ proc mountStoreSync*( return ok() -proc reconnectRelayPeers*(node: WakuNode) {.async.} = - ## Reconnect to previously-seen WakuRelay peers. - if node.wakuRelay.isNil(): - return - if not node.peerManager.switch.peerStore.hasPeers(protocolMatcher(WakuRelayCodec)): - return - info "Found previous WakuRelay peers. Reconnecting." - let backoffPeriod = - node.wakuRelay.parameters.pruneBackoff + chronos.seconds(BackoffSlackTime) - await node.peerManager.reconnectPeers(WakuRelayCodec, backoffPeriod) - proc selectRandomPeers*(peers: seq[PeerId], numRandomPeers: int): seq[PeerId] = var randomPeers = peers shuffle(randomPeers) @@ -621,10 +609,6 @@ proc start*(node: WakuNode) {.async.} = ## NOTE: This will dispatch gossipsub start to the WakuRelay.start method override await node.switch.start() - # Reconnect to known relay peers in the background; it waits a prune backoff - # and must not block startup. - node.relayReconnectFut = node.reconnectRelayPeers() - node.started = true if not node.wakuKademlia.isNil(): @@ -652,10 +636,6 @@ proc start*(node: WakuNode) {.async.} = proc stop*(node: WakuNode) {.async.} = ## By stopping the switch we are stopping all the underlying mounted protocols - # Cancel the background relay reconnection (may still be in its backoff wait). - if not node.relayReconnectFut.isNil(): - await node.relayReconnectFut.cancelAndWait() - await node.subscriptionManager.stop() node.stopProvidersAndListeners() diff --git a/logos_delivery/waku/node/waku_node/relay.nim b/logos_delivery/waku/node/waku_node/relay.nim index 18a3f0f59..686cf47f6 100644 --- a/logos_delivery/waku/node/waku_node/relay.nim +++ b/logos_delivery/waku/node/waku_node/relay.nim @@ -176,7 +176,6 @@ proc mountRelay*( if node.started: await node.wakuRelay.start() - await node.reconnectRelayPeers() node.switch.mount(node.wakuRelay, protocolMatcher(WakuRelayCodec)) diff --git a/tests/node/test_all.nim b/tests/node/test_all.nim index a53a83086..44a212fdc 100644 --- a/tests/node/test_all.nim +++ b/tests/node/test_all.nim @@ -8,4 +8,5 @@ import ./test_wakunode_store, ./test_wakunode_peer_manager, ./test_wakunode_health_monitor, - ./test_wakunode_restart + ./test_wakunode_restart, + ./test_wakunode_startup_reconnect diff --git a/tests/node/test_wakunode_peer_manager.nim b/tests/node/test_wakunode_peer_manager.nim index 0f85d61fe..9c2572a4f 100644 --- a/tests/node/test_wakunode_peer_manager.nim +++ b/tests/node/test_wakunode_peer_manager.nim @@ -591,62 +591,12 @@ suite "Peer Manager": serverPeerStore.getPeer(clientPeerId).connectedness == Connectedness.CannotConnect - suite "Automatic Reconnection": - asyncTest "Automatic Reconnection Implementation": - # Given two correctly initialised nodes, that are available for reconnection - (await server.mountRelay()).isOkOr: - assert false, "Failed to mount relay" - (await client.mountRelay()).isOkOr: - assert false, "Failed to mount relay" - await client.connectToNodes(@[serverRemotePeerInfo]) - - waitActive: - clientPeerStore.getPeer(serverPeerId).connectedness == - Connectedness.Connected and - serverPeerStore.getPeer(clientPeerId).connectedness == - Connectedness.Connected - - await client.disconnectNode(serverRemotePeerInfo) - - waitActive: - clientPeerStore.getPeer(serverPeerId).connectedness == - Connectedness.CanConnect and - serverPeerStore.getPeer(clientPeerId).connectedness == - Connectedness.CanConnect - - # When triggering the reconnection - await client.peerManager.reconnectPeers(WakuRelayCodec) - - # Then both peers should be marked as Connected - waitActive: - clientPeerStore.getPeer(serverPeerId).connectedness == - Connectedness.Connected and - serverPeerStore.getPeer(clientPeerId).connectedness == - Connectedness.Connected - - ## Now let's do the same but with backoff period - await client.disconnectNode(serverRemotePeerInfo) - - waitActive: - clientPeerStore.getPeer(serverPeerId).connectedness == - Connectedness.CanConnect and - serverPeerStore.getPeer(clientPeerId).connectedness == - Connectedness.CanConnect - - # When triggering a reconnection with a backoff period - let backoffPeriod = chronos.seconds(1) - let beforeReconnect = getTime().toUnixFloat() - await client.peerManager.reconnectPeers(WakuRelayCodec, backoffPeriod) - let reconnectDurationWithBackoffPeriod = - getTime().toUnixFloat() - beforeReconnect - - # Then both peers should be marked as Connected - check: - clientPeerStore.getPeer(serverPeerId).connectedness == - Connectedness.Connected - serverPeerStore.getPeer(clientPeerId).connectedness == - Connectedness.Connected - reconnectDurationWithBackoffPeriod > backoffPeriod.seconds.float + # NOTE: the "Automatic Reconnection" sub-suite that lived here tested + # peerManager.reconnectPeers, which was removed together with the + # startup relay reconnection (it misfired on discovery-populated peer + # stores). On-demand reconnection now goes through the connectivity + # maintenance path and is covered by + # tests/node/test_wakunode_startup_reconnect.nim. suite "Handling Connections on Different Networks": # TODO: Implement after discv5 and peer manager's interaction is understood diff --git a/tests/node/test_wakunode_restart.nim b/tests/node/test_wakunode_restart.nim index 8d05ce131..ffb07af5e 100644 --- a/tests/node/test_wakunode_restart.nim +++ b/tests/node/test_wakunode_restart.nim @@ -9,7 +9,8 @@ import ../testlib/[wakucore, wakunode, testasync] suite "WakuNode - restart (#3979)": asyncTest "start -> stop -> start re-opens the listener promptly": - ## A restart must not block on the relay-reconnect backoff. + ## A restart must complete promptly; historically start() could block on + ## a relay-reconnect backoff sweep of the peer store (since removed). let node1 = newTestWakuNode(generateSecp256k1Key()) node2 = newTestWakuNode(generateSecp256k1Key()) @@ -21,7 +22,7 @@ suite "WakuNode - restart (#3979)": await allFutures(node1.start(), node2.start()) - # node1 learns node2 as a relay peer, so a restart triggers reconnectRelayPeers. + # node1 learns node2 as a relay peer, so its peer store is non-empty on restart. await node1.connectToNodes(@[node2.peerInfo.toRemotePeerInfo()]) await node1.stop() diff --git a/tests/node/test_wakunode_startup_reconnect.nim b/tests/node/test_wakunode_startup_reconnect.nim new file mode 100644 index 000000000..0a34fc037 --- /dev/null +++ b/tests/node/test_wakunode_startup_reconnect.nim @@ -0,0 +1,300 @@ +{.used.} + +## Regression suite: relay peers present in the peer store at startup must not +## trigger reconnect-with-backoff behavior. +## +## Fast peer discovery (e.g. Kademlia service discovery bootstrapping) can +## populate the peer store before or during node startup. The peer store alone +## cannot tell such freshly discovered peers apart from peers restored from +## persistent storage, so any startup logic keyed on "the store has relay +## peers" misfires: discovered peers were being swept into a +## reconnect-after-prune-backoff wait (~62s), applied serially per peer, which +## stalled connectivity for minutes and blocked late relay mounts. +## +## The contract these tests pin down: +## - Adding a peer to the peer store is bookkeeping, not a dial instruction; +## neither node.start() nor a late mountRelay() may dial or sleep on store +## contents. +## - Establishing and re-establishing relay connectivity is the connectivity +## maintenance path's job (connectToRelayPeers / relayConnectivityLoop), +## which is prompt and respects per-peer failure backoff. + +import testutils/unittests, chronos, libp2p/[peerid, multiaddress] + +import + logos_delivery/waku/[waku_core, waku_node, node/peer_manager, waku_relay/protocol], + ../testlib/[wakucore, wakunode, testasync, testutils] + +# Valid peerId missing the last digit; append a digit to mint distinct peers. +const BasePeerId = "QmeuZJbXrszW2jdT7GdduSjQskPU3S7vvGWKtKgDfkDvW" + +proc unreachableRelayPeer(n: int, origin = PeerOrigin.Kademlia): RemotePeerInfo = + ## A relay-capable peer record the way discovery would hand it over: + ## peer id, dialable-looking address, advertised relay protocol. The port + ## has no listener, so any dial attempt fails fast and leaves a trace in + ## numberFailedConn. + var peerId: PeerId + doAssert peerId.init(BasePeerId & $n) + RemotePeerInfo.init( + peerId = peerId, + addrs = @[MultiAddress.init("/ip4/127.0.0.1/tcp/" & $n).tryGet()], + protocols = @[WakuRelayCodec], + origin = origin, + ) + +suite "Startup relay reconnect regression": + asyncTest "mounting relay on a running node is not stalled by known relay peers": + ## Given a started node without relay, whose peer store discovery has + ## already populated with relay-capable peers + let node = newTestWakuNode(generateSecp256k1Key()) + await node.start() + defer: + await node.stop() + + for n in 1 .. 3: + node.peerManager.addPeer(unreachableRelayPeer(n), PeerOrigin.Kademlia) + + ## When relay is mounted late (e.g. edge node promoted to relay) + let mountFut = node.mountRelay() + let mountedInTime = await mountFut.withTimeout(10.seconds) + if not mountedInTime: + await mountFut.cancelAndWait() + + ## Then the mount completes promptly instead of sleeping a prune backoff + ## per known peer + check mountedInTime + if mountedInTime: + check mountFut.read().isOk() + + asyncTest "already-connected relay peers do not stall a late relay mount": + ## Given a relay server and a started relay-less client connected to it + let + server = newTestWakuNode(generateSecp256k1Key()) + client = newTestWakuNode(generateSecp256k1Key()) + (await server.mountRelay()).isOkOr: + raiseAssert "mountRelay server: " & error + await allFutures(server.start(), client.start()) + defer: + await allFutures(client.stop(), server.stop()) + + let + serverPeerInfo = server.switch.peerInfo.toRemotePeerInfo() + clientPeerStore = client.peerManager.switch.peerStore + await client.connectToNodes(@[serverPeerInfo]) + + # identify has advertised the server's relay support to the client + waitActive: + clientPeerStore.getPeer(serverPeerInfo.peerId).protocols.contains(WakuRelayCodec) + + ## When the client mounts relay while that relay-capable peer is connected + let mountFut = client.mountRelay() + let mountedInTime = await mountFut.withTimeout(10.seconds) + if not mountedInTime: + await mountFut.cancelAndWait() + + ## Then the mount is not delayed on account of the connected peer + check mountedInTime + if mountedInTime: + check mountFut.read().isOk() + check clientPeerStore.getPeer(serverPeerInfo.peerId).connectedness == Connected + + asyncTest "startup does not auto-dial peers that discovery added to the peer store": + ## Given a relay server and a relay client that learned about the server + ## (plus some unreachable peers) via discovery before starting + let + server = newTestWakuNode(generateSecp256k1Key()) + client = newTestWakuNode(generateSecp256k1Key()) + (await server.mountRelay()).isOkOr: + raiseAssert "mountRelay server: " & error + (await client.mountRelay()).isOkOr: + raiseAssert "mountRelay client: " & error + await server.start() + + # keep any legacy reconnect wait short enough to observe inside the window + client.wakuRelay.parameters.pruneBackoff = chronos.seconds(1) + + let + serverPeerInfo = server.switch.peerInfo.toRemotePeerInfo() + clientPeerStore = client.peerManager.switch.peerStore + client.peerManager.addPeer(serverPeerInfo, PeerOrigin.Kademlia) + for n in 1 .. 2: + client.peerManager.addPeer(unreachableRelayPeer(n), PeerOrigin.Kademlia) + + ## When the client starts + await client.start() + defer: + await allFutures(client.stop(), server.stop()) + + ## Then no store peer is dialed behind the connectivity policy's back + var autoDialed = false + for _ in 0 ..< 16: + if clientPeerStore.getPeer(serverPeerInfo.peerId).connectedness == Connected: + autoDialed = true + break + await sleepAsync(500.milliseconds) + + check not autoDialed + for n in 1 .. 2: + check clientPeerStore.getPeer(unreachableRelayPeer(n).peerId).numberFailedConn == 0 + + asyncTest "startup ignores peer-store relay peers regardless of peer origin": + ## Given a relay node whose peer store holds relay-capable peers of + ## assorted origins (none of them from persistent storage — persistence + ## is not even enabled) + let node = newTestWakuNode(generateSecp256k1Key()) + (await node.mountRelay()).isOkOr: + raiseAssert "mountRelay: " & error + node.wakuRelay.parameters.pruneBackoff = chronos.seconds(1) + + let peerStore = node.peerManager.switch.peerStore + let origins = + [PeerOrigin.Kademlia, PeerOrigin.Discv5, PeerOrigin.Static, PeerOrigin.Dns] + for n, origin in origins: + node.peerManager.addPeer(unreachableRelayPeer(n + 1, origin), origin) + + ## When the node starts and runs for a while + await node.start() + defer: + await node.stop() + await sleepAsync(6.seconds) + + ## Then none of them was treated as a reconnect candidate + for n, origin in origins: + let peer = peerStore.getPeer(unreachableRelayPeer(n + 1, origin).peerId) + check: + peer.numberFailedConn == 0 + peer.connectedness == NotConnected + + asyncTest "the connectivity loop connects discovered relay peers promptly": + ## Given a relay server known to the client only through its peer store + ## (as after discovery), with the connectivity maintenance loop running + ## as node_factory.startNode wires it in production + let + server = newTestWakuNode(generateSecp256k1Key()) + client = newTestWakuNode(generateSecp256k1Key()) + (await server.mountRelay()).isOkOr: + raiseAssert "mountRelay server: " & error + (await client.mountRelay()).isOkOr: + raiseAssert "mountRelay client: " & error + await allFutures(server.start(), client.start()) + defer: + await allFutures(client.stop(), server.stop()) + + let + serverPeerInfo = server.switch.peerInfo.toRemotePeerInfo() + clientPeerStore = client.peerManager.switch.peerStore + client.peerManager.addPeer(serverPeerInfo, PeerOrigin.Kademlia) + + ## When the connectivity loop starts + client.peerManager.start() + defer: + client.peerManager.stop() + + ## Then the discovered peer is connected promptly, no backoff involved + waitActive: + clientPeerStore.getPeer(serverPeerInfo.peerId).connectedness == Connected + + asyncTest "relay connectivity is re-established on demand via the maintenance path": + ## Given two connected relay nodes that then disconnect + let + server = newTestWakuNode(generateSecp256k1Key()) + client = newTestWakuNode(generateSecp256k1Key()) + (await server.mountRelay()).isOkOr: + raiseAssert "mountRelay server: " & error + (await client.mountRelay()).isOkOr: + raiseAssert "mountRelay client: " & error + await allFutures(server.start(), client.start()) + defer: + await allFutures(client.stop(), server.stop()) + + let + serverPeerInfo = server.switch.peerInfo.toRemotePeerInfo() + clientPeerStore = client.peerManager.switch.peerStore + await client.connectToNodes(@[serverPeerInfo]) + waitActive: + clientPeerStore.getPeer(serverPeerInfo.peerId).connectedness == Connected + + await client.disconnectNode(serverPeerInfo) + waitActive: + clientPeerStore.getPeer(serverPeerInfo.peerId).connectedness == CanConnect + + ## When the connectivity maintenance pass runs + await client.peerManager.connectToRelayPeers() + + ## Then the relay peer is connected again, with no built-in delay + waitActive: + clientPeerStore.getPeer(serverPeerInfo.peerId).connectedness == Connected + + asyncTest "restart re-establishes relay connectivity via the maintenance path": + ## Given a relay node that was connected to a relay peer and restarted + let + node1 = newTestWakuNode(generateSecp256k1Key()) + node2 = newTestWakuNode(generateSecp256k1Key()) + (await node1.mountRelay()).isOkOr: + raiseAssert "mountRelay node1: " & error + (await node2.mountRelay()).isOkOr: + raiseAssert "mountRelay node2: " & error + await allFutures(node1.start(), node2.start()) + + let + node2PeerInfo = node2.switch.peerInfo.toRemotePeerInfo() + node1PeerStore = node1.peerManager.switch.peerStore + await node1.connectToNodes(@[node2PeerInfo]) + waitActive: + node1PeerStore.getPeer(node2PeerInfo.peerId).connectedness == Connected + + await node1.stop() + + ## When it starts again and the connectivity maintenance pass runs + let restarted = await node1.start().withTimeout(10.seconds) + check restarted + defer: + await allFutures(node1.stop(), node2.stop()) + + await node1.peerManager.connectToRelayPeers() + + ## Then the previously-known relay peer is connected again promptly + waitActive: + node1PeerStore.getPeer(node2PeerInfo.peerId).connectedness == Connected + + asyncTest "start and stop with a populated peer store are prompt and clean": + ## Given a relay node whose peer store holds unreachable relay peers + let node = newTestWakuNode(generateSecp256k1Key()) + (await node.mountRelay()).isOkOr: + raiseAssert "mountRelay: " & error + for n in 1 .. 3: + node.peerManager.addPeer(unreachableRelayPeer(n), PeerOrigin.Kademlia) + + ## When it starts and immediately stops + let startedInTime = await node.start().withTimeout(10.seconds) + check startedInTime + + let stoppedInTime = await node.stop().withTimeout(10.seconds) + + ## Then neither direction blocks on peer-store contents + check stoppedInTime + + asyncTest "relay streams form normally after startup with a pre-populated peer store": + ## Given a relay client that starts with discovered peers (reachable and + ## not) already in its peer store + let + server = newTestWakuNode(generateSecp256k1Key()) + client = newTestWakuNode(generateSecp256k1Key()) + (await server.mountRelay()).isOkOr: + raiseAssert "mountRelay server: " & error + (await client.mountRelay()).isOkOr: + raiseAssert "mountRelay client: " & error + await server.start() + + for n in 1 .. 2: + client.peerManager.addPeer(unreachableRelayPeer(n), PeerOrigin.Kademlia) + await client.start() + defer: + await allFutures(client.stop(), server.stop()) + + ## When it connects to a reachable relay peer + await client.connectToNodes(@[server.switch.peerInfo.toRemotePeerInfo()]) + + ## Then relay protocol streams are established as usual + waitActive: + client.peerManager.connectedPeers(WakuRelayCodec)[1].len == 1 diff --git a/tests/test_peer_manager.nim b/tests/test_peer_manager.nim index b364fc8c3..7f6d8aa64 100644 --- a/tests/test_peer_manager.nim +++ b/tests/test_peer_manager.nim @@ -396,8 +396,18 @@ procSuite "Peer Manager": quicEnabled = false, ) node2Key = generateSecp256k1Key() - node2 = - newTestWakuNode(node2Key, getPrimaryIPAddr(), Port(0), quicEnabled = false) + # The sharded peer manager only dials peers whose ENR advertises the + # Relay capability and a matching shard, so node2 must announce them. + # (Historically this test passed without the flags because mountRelay + # on a started node ran the since-removed startup relay reconnect, + # which dialed store peers indiscriminately.) + node2 = newTestWakuNode( + node2Key, + getPrimaryIPAddr(), + Port(0), + quicEnabled = false, + wakuFlags = some(CapabilitiesBitfield.init(@[Relay])), + ) node1.mountMetadata(0, @[0'u16]).expect("Mounted Waku Metadata") node2.mountMetadata(0, @[0'u16]).expect("Mounted Waku Metadata") @@ -454,6 +464,15 @@ procSuite "Peer Manager": (await node3.mountRelay()).isOkOr: assert false, "Failed to mount relay" + # The sharded peer manager works off the shards the node participates in + # (relay subscriptions); without one it manages nothing. + proc simpleHandler( + topic: PubsubTopic, msg: WakuMessage + ): Future[void] {.async, gcsafe.} = + await sleepAsync(0.millis) + + node3.wakuRelay.subscribe("/waku/2/rs/0/0", simpleHandler) + await node3.peerManager.manageRelayPeers() await sleepAsync(chronos.milliseconds(500))