fix: dial QUIC before TCP regardless of address list order (#4061)

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))

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

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: @[],

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()))
 

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()