From 296460a407cecaa743787173ddd2abc2d7b8ad9e Mon Sep 17 00:00:00 2001 From: Fabiana Cecin Date: Fri, 24 Jul 2026 09:41:35 -0300 Subject: [PATCH] fix: dial QUIC before TCP regardless of address list order (#4061) --- logos_delivery/waku/node/delivery_dialer.nim | 63 ++++++++++ logos_delivery/waku/node/waku_switch.nim | 5 +- logos_delivery/waku/waku_core/peers.nim | 10 +- tests/test_peer_manager.nim | 114 ++++++++++++++++++- tests/test_wakunode.nim | 18 --- 5 files changed, 182 insertions(+), 28 deletions(-) create mode 100644 logos_delivery/waku/node/delivery_dialer.nim diff --git a/logos_delivery/waku/node/delivery_dialer.nim b/logos_delivery/waku/node/delivery_dialer.nim new file mode 100644 index 000000000..5d34a33c5 --- /dev/null +++ b/logos_delivery/waku/node/delivery_dialer.nim @@ -0,0 +1,63 @@ +{.push raises: [].} + +import std/[sequtils, strutils] +import chronos, results +import + libp2p/dial, + libp2p/dialer, + libp2p/switch, + libp2p/peerid, + libp2p/multiaddress, + libp2p/stream/connection, + libp2p/muxers/muxer + +export dialer + +proc sortQuicFirst(addrs: seq[MultiAddress]): seq[MultiAddress] = + addrs.filterIt("/quic-v1" in $it) & addrs.filterIt("/quic-v1" notin $it) + +type DeliveryDialer* = ref object of Dialer + ## Logos Delivery dial policy layer. Replaces the switch dialer; every + ## dial in the process goes through here. Dials quic addresses before tcp. + +proc install*(T: typedesc[DeliveryDialer], switch: Switch) = + switch.dialer = DeliveryDialer.new( + switch.peerInfo.peerId, switch.connManager, switch.peerStore, switch.transports, + switch.ms, switch.nameResolver, + ) + +method connect*( + self: DeliveryDialer, + peerId: PeerId, + addrs: seq[MultiAddress], + forceDial = false, + reuseConnection = true, + dir = Direction.Out, +) {.async: (raises: [DialFailedError, CancelledError]).} = + await procCall Dialer(self).connect( + peerId, sortQuicFirst(addrs), forceDial, reuseConnection, dir + ) + +method dial*( + self: DeliveryDialer, + peerId: PeerId, + addrs: seq[MultiAddress], + protos: seq[string], + forceDial = false, +): Future[Stream] {.async: (raises: [DialFailedError, CancelledError]).} = + await procCall Dialer(self).dial(peerId, sortQuicFirst(addrs), protos, forceDial) + +method dialAndUpgrade*( + self: DeliveryDialer, + peerId: Opt[PeerId], + addrs: seq[MultiAddress], + dir = Direction.Out, +): Future[Muxer] {. + async: (raises: [CancelledError, MaError, TransportAddressError, LPError]) +.} = + await procCall Dialer(self).dialAndUpgrade(peerId, sortQuicFirst(addrs), dir) + +method tryDial*( + self: DeliveryDialer, peerId: PeerId, addrs: seq[MultiAddress] +): Future[Opt[MultiAddress]] {.async: (raises: [DialFailedError, CancelledError]).} = + await procCall Dialer(self).tryDial(peerId, sortQuicFirst(addrs)) diff --git a/logos_delivery/waku/node/waku_switch.nim b/logos_delivery/waku/node/waku_switch.nim index e0983cbee..37e2b99af 100644 --- a/logos_delivery/waku/node/waku_switch.nim +++ b/logos_delivery/waku/node/waku_switch.nim @@ -14,6 +14,7 @@ import libp2p/builders, libp2p/switch, libp2p/transports/[transport, tcptransport, wstransport] +import ./delivery_dialer # override nim-libp2p default value (which is also 1) const MaxConnectionsPerPeer* = 1 @@ -132,4 +133,6 @@ proc newWakuSwitch*( if not rendezvous.isNil(): b = b.withRendezVous() - b.build() + let switch = b.build() + DeliveryDialer.install(switch) + switch diff --git a/logos_delivery/waku/waku_core/peers.nim b/logos_delivery/waku/waku_core/peers.nim index 5fca2a649..f88840afa 100644 --- a/logos_delivery/waku/waku_core/peers.nim +++ b/logos_delivery/waku/waku_core/peers.nim @@ -251,13 +251,9 @@ proc parseUrlPeerAddr*(peerAddr: Opt[string]): Result[Opt[RemotePeerInfo], strin return ok(Opt.some(parsedPeerInfo)) -proc sortQuicFirst(addrs: seq[MultiAddress]): seq[MultiAddress] = - ## QUIC addresses first, so they are dialed ahead of TCP. - addrs.filterIt("/quic-v1" in $it) & addrs.filterIt("/quic-v1" notin $it) - proc toRemotePeerInfo*(enrRec: enr.Record): Result[RemotePeerInfo, cstring] = ## enr to dialable RemotePeerInfo. tcp from tcp/tcp6 fields, quic from the - ## multiaddrs ext (udp field is discv5, not quic). quic sorted first. + ## multiaddrs ext (udp field is discv5, not quic). let typedR = enrRec.toTyped().valueOr: return err(cstring("enr: failed to construct typed record: " & $error)) if not typedR.secp256k1.isSome(): @@ -296,8 +292,6 @@ proc toRemotePeerInfo*(enrRec: enr.Record): Result[RemotePeerInfo, cstring] = if addrs.len == 0: return err("enr: no dialable addresses in record") - addrs = sortQuicFirst(addrs) - let protocolsRes = catch: enrRec.getCapabilitiesCodecs() @@ -320,7 +314,7 @@ converter toRemotePeerInfo*(peerInfo: PeerInfo): RemotePeerInfo = ## Useful for testing or internal connections RemotePeerInfo( peerId: peerInfo.peerId, - addrs: sortQuicFirst(peerInfo.listenAddrs), + addrs: peerInfo.listenAddrs, enr: Opt.none(enr.Record), protocols: peerInfo.protocols, shards: @[], diff --git a/tests/test_peer_manager.nim b/tests/test_peer_manager.nim index e2f5a5a26..bd747299a 100644 --- a/tests/test_peer_manager.nim +++ b/tests/test_peer_manager.nim @@ -1,7 +1,7 @@ {.used.} import - std/[sequtils, times, sugar, net], + std/[sequtils, strutils, times, sugar, net], results, testutils/unittests, chronos, @@ -61,6 +61,118 @@ procSuite "Peer Manager": nodes[0].peerManager.switch.peerStore.connectedness(nodes[1].peerInfo.peerId) == Connectedness.Connected + asyncTest "connectPeer() prefers quic even when tcp is listed first": + ## Announced address lists are tcp-first and identify rewrites the + ## address book with that order after every connection, so dial paths + ## must reorder quic ahead of tcp themselves. + let nodes = toSeq(0 ..< 2).mapIt(newTestWakuNode(generateSecp256k1Key())) + await allFutures(nodes.mapIt(it.start())) + + let announced = nodes[1].peerInfo.toRemotePeerInfo().addrs + let tcpFirst = + announced.filterIt("/quic-v1" notin $it) & announced.filterIt("/quic-v1" in $it) + require tcpFirst.anyIt("/quic-v1" in $it) + require "/quic-v1" notin $tcpFirst[0] + + let peer = RemotePeerInfo.init(nodes[1].peerInfo.peerId, tcpFirst) + require await nodes[0].peerManager.connectPeer(peer) + await sleepAsync(chronos.milliseconds(200)) + + let conns = nodes[0].peerManager.switch.connManager.getConnections().getOrDefault( + nodes[1].peerInfo.peerId + ) + require conns.len >= 1 + let obsAddr = conns[0].connection.observedAddr + check: + obsAddr.isSome() + "/quic-v1" in $obsAddr.get() + + await allFutures(nodes.mapIt(it.stop())) + + asyncTest "reconnect after identify address book refresh still dials quic": + ## The original quic-to-tcp drift: identify rewrites the address book + ## with the peer's announced tcp-first order after every connection, so + ## reconnects from the book must still come out quic. + let nodes = toSeq(0 ..< 2).mapIt(newTestWakuNode(generateSecp256k1Key())) + await allFutures(nodes.mapIt(it.start())) + let peerId = nodes[1].peerInfo.peerId + + require await nodes[0].peerManager.connectPeer(nodes[1].peerInfo.toRemotePeerInfo()) + await sleepAsync(chronos.milliseconds(500)) + + # identify must have populated the book with a dual-stack address set + let bookAddrs = nodes[0].peerManager.switch.peerStore[AddressBook][peerId] + require bookAddrs.anyIt("/quic-v1" in $it) + require bookAddrs.anyIt("/quic-v1" notin $it) + + # pin the book to tcp-first, as identify writes it today + let tcpFirst = + bookAddrs.filterIt("/quic-v1" notin $it) & bookAddrs.filterIt("/quic-v1" in $it) + nodes[0].peerManager.switch.peerStore[AddressBook][peerId] = tcpFirst + + await nodes[0].peerManager.disconnectNode(peerId) + await sleepAsync(chronos.milliseconds(500)) + + require await nodes[0].peerManager.connectPeer( + RemotePeerInfo.init(peerId, tcpFirst) + ) + let conns = + nodes[0].peerManager.switch.connManager.getConnections().getOrDefault(peerId) + require conns.len >= 1 + let obsAddr = conns[0].connection.observedAddr + check: + obsAddr.isSome() + "/quic-v1" in $obsAddr.get() + + await allFutures(nodes.mapIt(it.stop())) + + asyncTest "protocol stream dial from tcp-first address book uses quic": + ## dialPeer(peerId, proto) resolves addresses from the address book and + ## must also come out quic when the book is tcp-first. + let nodes = toSeq(0 ..< 2).mapIt(newTestWakuNode(generateSecp256k1Key())) + await allFutures(nodes.mapIt(it.start())) + let peerId = nodes[1].peerInfo.peerId + + let announced = nodes[1].peerInfo.toRemotePeerInfo().addrs + let tcpFirst = + announced.filterIt("/quic-v1" notin $it) & announced.filterIt("/quic-v1" in $it) + require "/quic-v1" notin $tcpFirst[0] + nodes[0].peerManager.addPeer(RemotePeerInfo.init(peerId, tcpFirst)) + + let connOpt = await nodes[0].peerManager.dialPeer(peerId, "/ipfs/id/1.0.0") + require connOpt.isSome() + let obsAddr = connOpt.get().observedAddr + check: + obsAddr.isSome() + "/quic-v1" in $obsAddr.get() + + await allFutures(nodes.mapIt(it.stop())) + + asyncTest "tcp-only dialer connects to a quic-first address list": + ## A node without quic support must skip quic addresses at no cost and + ## connect over tcp. + let dialer = newTestWakuNode(generateSecp256k1Key(), quicEnabled = false) + let server = newTestWakuNode(generateSecp256k1Key()) + await allFutures(dialer.start(), server.start()) + + let announced = server.peerInfo.toRemotePeerInfo().addrs + let quicFirst = + announced.filterIt("/quic-v1" in $it) & announced.filterIt("/quic-v1" notin $it) + require "/quic-v1" in $quicFirst[0] + + let peer = RemotePeerInfo.init(server.peerInfo.peerId, quicFirst) + require await dialer.peerManager.connectPeer(peer) + let conns = dialer.peerManager.switch.connManager.getConnections().getOrDefault( + server.peerInfo.peerId + ) + require conns.len >= 1 + let obsAddr = conns[0].connection.observedAddr + check: + obsAddr.isSome() + "/quic-v1" notin $obsAddr.get() + + await allFutures(dialer.stop(), server.stop()) + asyncTest "Peer manager tracks active store request state": let nodes = toSeq(0 ..< 2).mapIt(newTestWakuNode(generateSecp256k1Key())) diff --git a/tests/test_wakunode.nim b/tests/test_wakunode.nim index a607c9f2c..d6c6f8cd4 100644 --- a/tests/test_wakunode.nim +++ b/tests/test_wakunode.nim @@ -242,24 +242,6 @@ suite "WakuNode": await node.stop() - test "toRemotePeerInfo sorts quic-v1 addresses first": - let - nodeKey = generateSecp256k1Key() - node = newTestWakuNode(nodeKey, quicEnabled = true) - - # peerinfo path - let fromPeerInfo = node.switch.peerInfo.toRemotePeerInfo() - check: - fromPeerInfo.addrs.anyIt("/quic-v1" in $it) - "/quic-v1" in $fromPeerInfo.addrs[0] - - # enr path - let fromEnr = toRemotePeerInfo(node.enr).valueOr: - raiseAssert "toRemotePeerInfo(enr) failed: " & $error - check: - fromEnr.addrs.anyIt("/quic-v1" in $it) - "/quic-v1" in $fromEnr.addrs[0] - asyncTest "Dual-stack nodes connect over QUIC": let nodeKey1 = generateSecp256k1Key()