fix: check for new addresses during dialing (#3003)

Fixes a race condition where two independent dials occur for the same peer id with different addresses (e.g. one IPv4 and one IPv6) - if the IPv6 is the current dial, the IPv4 dial joins the IPv6 one which can then fail if the current network doesn't support IPv6.

Instead check the list of addresses to dial after each dial attempt to see if new addresses have been added.

Author: achingbrain

packages/libp2p/src/connection-manager/dial-queue.ts

@@ -203,7 +203,7 @@ export class DialQueue {
 
     options.onProgress?.(new CustomProgressEvent('dial-queue:add-to-dial-queue'))
     return this.queue.add(async (options) => {
-      options?.onProgress?.(new CustomProgressEvent('dial-queue:start-dial'))
+      options.onProgress?.(new CustomProgressEvent('dial-queue:start-dial'))
       // create abort conditions - need to do this before `calculateMultiaddrs` as
       // we may be about to resolve a dns addr which can time out
       const signal = anySignal([
@@ -212,103 +212,145 @@ export class DialQueue {
       ])
       setMaxListeners(Infinity, signal)
 
-      let addrsToDial: Address[]
-
       try {
-        // load addresses from address book, resolve and dnsaddrs, filter
-        // undialables, add peer IDs, etc
-        addrsToDial = await this.calculateMultiaddrs(peerId, options?.multiaddrs, {
-          ...options,
-          signal
-        })
+        return await this.dialPeer(options, signal)
+      } finally {
+        // clean up abort signals/controllers
+        signal.clear()
+      }
+    }, {
+      peerId,
+      priority: options.priority ?? DEFAULT_DIAL_PRIORITY,
+      multiaddrs: new Set(multiaddrs.map(ma => ma.toString())),
+      signal: options.signal ?? AbortSignal.timeout(this.dialTimeout),
+      onProgress: options.onProgress
+    })
+  }
 
-        options?.onProgress?.(new CustomProgressEvent<Address[]>('dial-queue:calculated-addresses', addrsToDial))
+  private async dialPeer (options: DialQueueJobOptions, signal: AbortSignal): Promise<Connection> {
+    const peerId = options.peerId
+    const multiaddrs = options.multiaddrs
+    const failedMultiaddrs = new Set<string>()
 
-        addrsToDial.map(({ multiaddr }) => multiaddr.toString()).forEach(addr => {
-          options?.multiaddrs.add(addr)
-        })
-      } catch (err) {
-        signal.clear()
-        throw err
+    // if we have no multiaddrs, only a peer id, set a flag so we will look the
+    // peer up in the peer routing to obtain multiaddrs
+    let forcePeerLookup = options.multiaddrs.size === 0
+
+    let dialed = 0
+    let dialIteration = 0
+    const errors: Error[] = []
+
+    this.log('starting dial to %p', peerId)
+
+    // repeat this operation in case addresses are added to the dial while we
+    // resolve multiaddrs, etc
+    while (forcePeerLookup || multiaddrs.size > 0) {
+      dialIteration++
+
+      // only perform peer lookup once
+      forcePeerLookup = false
+
+      // the addresses we will dial
+      const addrsToDial: Address[] = []
+
+      // copy the addresses into a new set
+      const addrs = new Set(options.multiaddrs)
+
+      // empty the old set - subsequent dial attempts for the same peer id may
+      // add more addresses to try
+      multiaddrs.clear()
+
+      this.log('calculating addrs to dial %p from %s', peerId, [...addrs])
+
+      // load addresses from address book, resolve and dnsaddrs, filter
+      // undialables, add peer IDs, etc
+      const calculatedAddrs = await this.calculateMultiaddrs(peerId, addrs, {
+        ...options,
+        signal
+      })
+
+      for (const addr of calculatedAddrs) {
+        // skip any addresses we have previously failed to dial
+        if (failedMultiaddrs.has(addr.multiaddr.toString())) {
+          this.log.trace('skipping previously failed multiaddr %a while dialing %p', addr.multiaddr, peerId)
+          continue
+        }
+
+        addrsToDial.push(addr)
       }
 
-      try {
-        let dialed = 0
-        const errors: Error[] = []
+      this.log('%s dial to %p with %s', dialIteration === 1 ? 'starting' : 'continuing', peerId, addrsToDial.map(ma => ma.multiaddr.toString()))
 
-        for (const address of addrsToDial) {
-          if (dialed === this.maxPeerAddrsToDial) {
-            this.log('dialed maxPeerAddrsToDial (%d) addresses for %p, not trying any others', dialed, peerId)
+      options?.onProgress?.(new CustomProgressEvent<Address[]>('dial-queue:calculated-addresses', addrsToDial))
 
-            throw new DialError('Peer had more than maxPeerAddrsToDial')
-          }
+      for (const address of addrsToDial) {
+        if (dialed === this.maxPeerAddrsToDial) {
+          this.log('dialed maxPeerAddrsToDial (%d) addresses for %p, not trying any others', dialed, options.peerId)
 
-          dialed++
+          throw new DialError('Peer had more than maxPeerAddrsToDial')
+        }
 
+        dialed++
+
+        try {
+          // try to dial the address
+          const conn = await this.components.transportManager.dial(address.multiaddr, {
+            ...options,
+            signal
+          })
+
+          this.log('dial to %a succeeded', address.multiaddr)
+
+          // record the successful dial and the address
           try {
-            const conn = await this.components.transportManager.dial(address.multiaddr, {
-              ...options,
-              signal
+            await this.components.peerStore.merge(conn.remotePeer, {
+              multiaddrs: [
+                conn.remoteAddr
+              ],
+              metadata: {
+                [LAST_DIAL_SUCCESS_KEY]: uint8ArrayFromString(Date.now().toString())
+              }
             })
+          } catch (err: any) {
+            this.log.error('could not update last dial failure key for %p', peerId, err)
+          }
+
+          // dial successful, return the connection
+          return conn
+        } catch (err: any) {
+          this.log.error('dial failed to %a', address.multiaddr, err)
 
-            this.log('dial to %a succeeded', address.multiaddr)
+          // ensure we don't dial it again in this attempt
+          failedMultiaddrs.add(address.multiaddr.toString())
 
-            // record the successful dial and the address
+          if (peerId != null) {
+            // record the failed dial
             try {
-              await this.components.peerStore.merge(conn.remotePeer, {
-                multiaddrs: [
-                  conn.remoteAddr
-                ],
+              await this.components.peerStore.merge(peerId, {
                 metadata: {
-                  [LAST_DIAL_SUCCESS_KEY]: uint8ArrayFromString(Date.now().toString())
+                  [LAST_DIAL_FAILURE_KEY]: uint8ArrayFromString(Date.now().toString())
                 }
               })
             } catch (err: any) {
               this.log.error('could not update last dial failure key for %p', peerId, err)
             }
+          }
 
-            return conn
-          } catch (err: any) {
-            this.log.error('dial failed to %a', address.multiaddr, err)
-
-            if (peerId != null) {
-              // record the failed dial
-              try {
-                await this.components.peerStore.merge(peerId, {
-                  metadata: {
-                    [LAST_DIAL_FAILURE_KEY]: uint8ArrayFromString(Date.now().toString())
-                  }
-                })
-              } catch (err: any) {
-                this.log.error('could not update last dial failure key for %p', peerId, err)
-              }
-            }
-
-            // the user/dial timeout/shutdown controller signal aborted
-            if (signal.aborted) {
-              throw new TimeoutError(err.message)
-            }
-
-            errors.push(err)
+          // the user/dial timeout/shutdown controller signal aborted
+          if (signal.aborted) {
+            throw new TimeoutError(err.message)
           }
-        }
 
-        if (errors.length === 1) {
-          throw errors[0]
+          errors.push(err)
         }
-
-        throw new AggregateError(errors, 'All multiaddr dials failed')
-      } finally {
-        // clean up abort signals/controllers
-        signal.clear()
       }
-    }, {
-      peerId,
-      priority: options.priority ?? DEFAULT_DIAL_PRIORITY,
-      multiaddrs: new Set(multiaddrs.map(ma => ma.toString())),
-      signal: options.signal ?? AbortSignal.timeout(this.dialTimeout),
-      onProgress: options.onProgress
-    })
+    }
+
+    if (errors.length === 1) {
+      throw errors[0]
+    }
+
+    throw new AggregateError(errors, 'All multiaddr dials failed')
   }
 
   // eslint-disable-next-line complexity
@@ -358,8 +400,10 @@ export class DialQueue {
             isCertified: false
           })))
         } catch (err: any) {
-          if (err.name !== 'NoPeerRoutersError') {
-            this.log.error('looking up multiaddrs for %p in the peer routing failed', peerId, err)
+          if (err.name === 'NoPeerRoutersError') {
+            this.log('no peer routers configured', peerId)
+          } else {
+            this.log.error('looking up multiaddrs for %p in the peer routing failed - %e', peerId, err)
           }
         }
       }

packages/libp2p/test/connection-manager/dial-queue.spec.ts

@@ -5,7 +5,7 @@ import { NotFoundError } from '@libp2p/interface'
 import { peerLogger } from '@libp2p/logger'
 import { peerIdFromPrivateKey } from '@libp2p/peer-id'
 import { multiaddr, resolvers } from '@multiformats/multiaddr'
-import { WebRTC } from '@multiformats/multiaddr-matcher'
+import { TCP, WebRTC } from '@multiformats/multiaddr-matcher'
 import { expect } from 'aegir/chai'
 import delay from 'delay'
 import pDefer from 'p-defer'
@@ -377,4 +377,47 @@ describe('dial queue', () => {
     expect(all[1].status).to.equal('fulfilled', 'did not respect user dial timeout')
     expect(components.transportManager.dial.callCount).to.equal(1, 'should have coalesced multiple dials to same dial')
   })
+
+  it('should continue dial when new addresses are discovered', async () => {
+    const remotePeer = peerIdFromPrivateKey(await generateKeyPair('Ed25519'))
+    const ma1 = multiaddr(`/ip6/2001:db8:1:2:3:4:5:6/tcp/123/p2p/${remotePeer}`)
+    const ma2 = multiaddr(`/ip4/123.123.123.123/tcp/123/p2p/${remotePeer}`)
+
+    components.transportManager.dialTransportForMultiaddr.callsFake(ma => {
+      if (TCP.exactMatch(ma)) {
+        return stubInterface<Transport>()
+      }
+    })
+
+    const connection = stubInterface<Connection>({
+      remotePeer
+    })
+
+    components.transportManager.dial.callsFake(async (ma, opts = {}) => {
+      if (ma.equals(ma2)) {
+        await delay(100)
+        return connection
+      }
+
+      // second dial should take place while this dial is in progress but has
+      // not yet failed
+      await delay(500)
+      throw new Error('Could not dial address')
+    })
+
+    dialer = new DialQueue(components)
+
+    // dial peer with address that fails
+    const dial1 = dialer.dial(ma1)
+
+    // let dial begin
+    await delay(50)
+
+    // dial same peer again with address that succeeds
+    const dial2 = dialer.dial(ma2)
+
+    // both dials should coalesce to the same connection
+    await expect(dial1).to.eventually.equal(connection)
+    await expect(dial2).to.eventually.equal(connection)
+  })
 })