mirror of
https://github.com/logos-messaging/logos-delivery.git
synced 2026-07-17 10:10:09 +00:00
fix: don't apply reconnect backoff to discovered relay peers (#4029)
This commit is contained in:
parent
77cb8a6c7a
commit
0a19473a3b
@ -223,7 +223,10 @@ proc loadFromStorage(pm: PeerManager) {.gcsafe.} =
|
||||
pm.switch.peerStore[ConnectionBook][peerId] = NotConnected
|
||||
# Reset connectedness state
|
||||
pm.switch.peerStore[DisconnectBook][peerId] = remotePeerInfo.disconnectTime
|
||||
pm.switch.peerStore[SourceBook][peerId] = remotePeerInfo.origin
|
||||
# Mark as Cache so it is distinguishable from live-discovered peers. If the
|
||||
# same peer is later rediscovered, addPeer overrides this with the live
|
||||
# origin, and reconnect backoff no longer applies to it.
|
||||
pm.switch.peerStore[SourceBook][peerId] = Cache
|
||||
|
||||
if remotePeerInfo.enr.isSome():
|
||||
pm.switch.peerStore[ENRBook][peerId] = remotePeerInfo.enr.get()
|
||||
@ -695,21 +698,28 @@ proc reconnectPeers*(
|
||||
|
||||
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
|
||||
# Only reconnect peers that come from persistent storage (Cache). Freshly
|
||||
# discovered peers must not be delayed by the reconnect backoff: they are
|
||||
# connected right away by the relay connectivity loop. Rediscovered peers get
|
||||
# their origin overridden by addPeer, so they naturally drop out of this set.
|
||||
let peersToReconnect = pm.switch.peerStore.peers(protocolMatcher(proto)).filterIt(
|
||||
it.origin == Cache and it.connectedness != CannotConnect
|
||||
)
|
||||
|
||||
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)
|
||||
if peersToReconnect.len == 0:
|
||||
return
|
||||
|
||||
await pm.connectToNodes(@[peerInfo])
|
||||
# We disconnected recently and still need to wait a backoff period before
|
||||
# reconnecting. The wait is shared across all peers rather than applied once
|
||||
# per peer, so reconnection can't stall behind a slow/unreachable peer and
|
||||
# keep the node from taking on freshly discovered ones.
|
||||
if backoffTime > ZeroDuration:
|
||||
info "Backing off before reconnect", backoffTime = backoffTime
|
||||
await sleepAsync(backoffTime)
|
||||
|
||||
# Dial all reconnectable peers in parallel; a single slow dial must not delay
|
||||
# the rest.
|
||||
await allFutures(peersToReconnect.mapIt(pm.connectToNodes(@[it])))
|
||||
|
||||
proc getNumStreams*(pm: PeerManager, protocol: string): (int, int) =
|
||||
var
|
||||
|
||||
@ -40,6 +40,7 @@ type
|
||||
PeerExchange
|
||||
Dns
|
||||
Kademlia
|
||||
Cache # Loaded from persistent peer storage (not a live discovery source)
|
||||
|
||||
PeerDirection* = enum
|
||||
UnknownDirection
|
||||
|
||||
@ -614,6 +614,11 @@ suite "Peer Manager":
|
||||
serverPeerStore.getPeer(clientPeerId).connectedness ==
|
||||
Connectedness.CanConnect
|
||||
|
||||
# reconnectPeers only acts on peers restored from persistent storage,
|
||||
# so tag the server as such. Peers coming from live discovery are
|
||||
# deliberately left out of the reconnect (and its backoff) path.
|
||||
client.peerManager.addPeer(serverRemotePeerInfo, origin = Cache)
|
||||
|
||||
# When triggering the reconnection
|
||||
await client.peerManager.reconnectPeers(WakuRelayCodec)
|
||||
|
||||
|
||||
@ -300,6 +300,54 @@ procSuite "Peer Manager":
|
||||
|
||||
await allFutures(nodes.mapIt(it.stop()))
|
||||
|
||||
asyncTest "reconnectPeers only reconnects persisted (Cache) peers":
|
||||
## Regression: freshly discovered relay peers were wrongly swept into the
|
||||
## reconnect backoff path. reconnectPeers must act only on peers loaded from
|
||||
## persistent storage (origin == Cache); discovered peers are left for the
|
||||
## connectivity loop to dial without any backoff.
|
||||
let node = newTestWakuNode(generateSecp256k1Key())
|
||||
await node.start()
|
||||
|
||||
let basePeerId = "QmeuZJbXrszW2jdT7GdduSjQskPU3S7vvGWKtKgDfkDvW"
|
||||
var cachePeerId, discoveredPeerId: PeerId
|
||||
require cachePeerId.init(basePeerId & "1")
|
||||
require discoveredPeerId.init(basePeerId & "2")
|
||||
|
||||
# Both peers advertise relay and point at an unreachable address, so a dial
|
||||
# attempt fails fast and is observable via the failed-connection counter.
|
||||
let unreachableAddr = MultiAddress.init("/ip4/127.0.0.1/tcp/1").tryGet()
|
||||
|
||||
node.peerManager.addPeer(
|
||||
RemotePeerInfo.init(
|
||||
peerId = cachePeerId, addrs = @[unreachableAddr], protocols = @[WakuRelayCodec]
|
||||
),
|
||||
origin = Cache,
|
||||
)
|
||||
node.peerManager.addPeer(
|
||||
RemotePeerInfo.init(
|
||||
peerId = discoveredPeerId,
|
||||
addrs = @[unreachableAddr],
|
||||
protocols = @[WakuRelayCodec],
|
||||
),
|
||||
origin = Discv5,
|
||||
)
|
||||
|
||||
# Default backoff is zero, so the test stays fast; the origin filter is what
|
||||
# we are validating here.
|
||||
await node.peerManager.reconnectPeers(WakuRelayCodec)
|
||||
|
||||
let peerStore = node.peerManager.switch.peerStore
|
||||
check:
|
||||
# The Cache peer was dialed (and failed against the unreachable address).
|
||||
peerStore[NumberFailedConnBook][cachePeerId] == 1
|
||||
peerStore.connectedness(cachePeerId) == CannotConnect
|
||||
|
||||
# The discovered peer was never touched by reconnectPeers.
|
||||
peerStore[NumberFailedConnBook][discoveredPeerId] == 0
|
||||
peerStore.connectedness(discoveredPeerId) == NotConnected
|
||||
|
||||
await node.stop()
|
||||
|
||||
asyncTest "Peer manager can use persistent storage and survive restarts":
|
||||
let
|
||||
database = SqliteDatabase.new(":memory:")[]
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user