fix: when passed a multiaddr, only dial that multiaddr (#1498)

* fix: when passed a multiaddr, only dial that multiaddr

When a user passes a multiaddr to `libp2p.dial`, only dial that
multiaddr instead of extracting the peer id from the multiaddr,
using it to look up every known address and dialing them all.

When passing a peer id to `libp2p.dial`, all addresses will be
dialed as before.

Fixes #1497

* chore: review comments

* chore: linting

Author: achingbrain

src/connection-manager/dialer/index.ts

@@ -1,12 +1,7 @@
 import { logger } from '@libp2p/logger'
-import all from 'it-all'
-import filter from 'it-filter'
-import { pipe } from 'it-pipe'
 import errCode from 'err-code'
-import type { Multiaddr, Resolver } from '@multiformats/multiaddr'
-import { multiaddr, resolvers } from '@multiformats/multiaddr'
+import { isMultiaddr, Multiaddr, Resolver, multiaddr, resolvers } from '@multiformats/multiaddr'
 import { TimeoutController } from 'timeout-abort-controller'
-import { AbortError } from '@libp2p/interfaces/errors'
 import { anySignal } from 'any-signal'
 import { setMaxListeners } from 'events'
 import { DialAction, DialRequest } from './dial-request.js'
@@ -22,10 +17,8 @@ import {
 import type { Connection, ConnectionGater } from '@libp2p/interface-connection'
 import type { AbortOptions } from '@libp2p/interfaces'
 import type { Startable } from '@libp2p/interfaces/startable'
-import type { PeerId } from '@libp2p/interface-peer-id'
+import { isPeerId, PeerId } from '@libp2p/interface-peer-id'
 import { getPeer } from '../../get-peer.js'
-import sort from 'it-sort'
-import map from 'it-map'
 import type { AddressSorter, PeerStore } from '@libp2p/interface-peer-store'
 import type { Metrics } from '@libp2p/interface-metrics'
 import type { Dialer } from '@libp2p/interface-connection-manager'
@@ -98,7 +91,7 @@ export class DefaultDialer implements Startable, Dialer {
   private readonly maxDialsPerPeer: number
   public tokens: number[]
   public pendingDials: Map<string, PendingDial>
-  public pendingDialTargets: Map<string, PendingDialTarget>
+  public pendingDialTargets: Map<string, AbortController>
   private started: boolean
 
   constructor (components: DefaultDialerComponents, init: DialerInit = {}) {
@@ -147,7 +140,7 @@ export class DefaultDialer implements Startable, Dialer {
     this.pendingDials.clear()
 
     for (const pendingTarget of this.pendingDialTargets.values()) {
-      pendingTarget.reject(new AbortError('Dialer was destroyed'))
+      pendingTarget.abort()
     }
     this.pendingDialTargets.clear()
   }
@@ -157,8 +150,8 @@ export class DefaultDialer implements Startable, Dialer {
    * The dial to the first address that is successfully able to upgrade a connection
    * will be used.
    */
-  async dial (peer: PeerId | Multiaddr, options: AbortOptions = {}): Promise<Connection> {
-    const { id, multiaddrs } = getPeer(peer)
+  async dial (peerIdOrMultiaddr: PeerId | Multiaddr, options: AbortOptions = {}): Promise<Connection> {
+    const { id, multiaddrs } = getPeer(peerIdOrMultiaddr)
 
     if (this.components.peerId.equals(id)) {
       throw errCode(new Error('Tried to dial self'), codes.ERR_DIALED_SELF)
@@ -177,13 +170,35 @@ export class DefaultDialer implements Startable, Dialer {
 
     log('creating dial target for %p', id)
 
-    const dialTarget = await this._createCancellableDialTarget(id, options)
+    // resolving multiaddrs can involve dns lookups so allow them to be aborted
+    const controller = new AbortController()
+    const controllerId = randomId()
+    this.pendingDialTargets.set(controllerId, controller)
+    let signal = controller.signal
+
+    // merge with the passed signal, if any
+    if (options.signal != null) {
+      signal = anySignal([signal, options.signal])
+    }
+
+    let dialTarget: DialTarget
+
+    try {
+      dialTarget = await this._createDialTarget(peerIdOrMultiaddr, {
+        ...options,
+        signal
+      })
+    } finally {
+      // done resolving the multiaddrs so remove the abort controller
+      this.pendingDialTargets.delete(controllerId)
+    }
 
     if (dialTarget.addrs.length === 0) {
       throw errCode(new Error('The dial request has no valid addresses'), codes.ERR_NO_VALID_ADDRESSES)
     }
 
-    const pendingDial = this.pendingDials.get(dialTarget.id) ?? this._createPendingDial(dialTarget, options)
+    // try to join an in-flight dial for this peer if one is available
+    const pendingDial = this.pendingDials.get(dialTarget.id.toString()) ?? this._createPendingDial(dialTarget, options)
 
     try {
       const connection = await pendingDial.promise
@@ -202,76 +217,77 @@ export class DefaultDialer implements Startable, Dialer {
     }
   }
 
-  /**
-   * Connects to a given `peer` by dialing all of its known addresses.
-   * The dial to the first address that is successfully able to upgrade a connection
-   * will be used.
-   */
-  async _createCancellableDialTarget (peer: PeerId, options: AbortOptions): Promise<DialTarget> {
-    // Make dial target promise cancellable
-    const id = `${(parseInt(String(Math.random() * 1e9), 10)).toString()}${Date.now()}`
-    const cancellablePromise = new Promise<DialTarget>((resolve, reject) => {
-      this.pendingDialTargets.set(id, { resolve, reject })
-    })
-
-    try {
-      const dialTarget = await Promise.race([
-        this._createDialTarget(peer, options),
-        cancellablePromise
-      ])
-
-      return dialTarget
-    } finally {
-      this.pendingDialTargets.delete(id)
-    }
-  }
-
   /**
    * Creates a DialTarget. The DialTarget is used to create and track
    * the DialRequest to a given peer.
-   * If a multiaddr is received it should be the first address attempted.
+   *
+   * If a multiaddr is received it should be the only address attempted.
+   *
    * Multiaddrs not supported by the available transports will be filtered out.
    */
-  async _createDialTarget (peer: PeerId, options: AbortOptions): Promise<DialTarget> {
+  async _createDialTarget (peerIdOrMultiaddr: PeerId | Multiaddr, options: AbortOptions): Promise<DialTarget> {
     const _resolve = this._resolve.bind(this)
 
-    let addrs = await pipe(
-      await this.components.peerStore.addressBook.get(peer),
-      (source) => filter(source, async (address) => {
-        return !(await this.components.connectionGater.denyDialMultiaddr(peer, address.multiaddr))
-      }),
-      // Sort addresses so, for example, we try certified public address first
-      (source) => sort(source, this.addressSorter),
-      async function * resolve (source) {
-        for await (const a of source) {
-          yield * await _resolve(a.multiaddr, options)
-        }
-      },
-      // Multiaddrs not supported by the available transports will be filtered out.
-      (source) => filter(source, (ma) => Boolean(this.components.transportManager.transportForMultiaddr(ma))),
-      (source) => map(source, (ma) => {
-        if (peer.toString() === ma.getPeerId()) {
-          return ma
-        }
+    let addrs = isMultiaddr(peerIdOrMultiaddr) ? [peerIdOrMultiaddr] : await this._loadAddresses(peerIdOrMultiaddr)
 
-        return ma.encapsulate(`/p2p/${peer.toString()}`)
-      }),
-      async (source) => await all(source)
-    )
+    addrs = (await Promise.all(
+      addrs.map(async (ma) => await _resolve(ma, options))
+    ))
+      .flat()
+      // Multiaddrs not supported by the available transports will be filtered out.
+      .filter(ma => Boolean(this.components.transportManager.transportForMultiaddr(ma)))
 
+    // deduplicate addresses
     addrs = [...new Set(addrs)]
 
     if (addrs.length > this.maxAddrsToDial) {
-      await this.components.peerStore.delete(peer)
       throw errCode(new Error('dial with more addresses than allowed'), codes.ERR_TOO_MANY_ADDRESSES)
     }
 
+    const peerId = isPeerId(peerIdOrMultiaddr) ? peerIdOrMultiaddr : undefined
+
+    if (peerId != null) {
+      const peerIdMultiaddr = `/p2p/${peerId.toString()}`
+      addrs = addrs.map(addr => {
+        const addressPeerId = addr.getPeerId()
+
+        if (addressPeerId == null || !peerId.equals(addressPeerId)) {
+          return addr.encapsulate(peerIdMultiaddr)
+        }
+
+        return addr
+      })
+    }
+
     return {
-      id: peer.toString(),
+      id: peerId == null ? randomId() : peerId.toString(),
       addrs
     }
   }
 
+  /**
+   * Loads a list of addresses from the peer store for the passed peer id
+   */
+  async _loadAddresses (peer: PeerId): Promise<Multiaddr[]> {
+    const addresses = await this.components.peerStore.addressBook.get(peer)
+
+    return (await Promise.all(
+      addresses.map(async address => {
+        const deny = await this.components.connectionGater.denyDialMultiaddr(peer, address.multiaddr)
+
+        if (deny) {
+          return false
+        }
+
+        return address
+      })
+    ))
+      .filter(isTruthy)
+      // Sort addresses so, for example, we try certified public address first
+      .sort(this.addressSorter)
+      .map(address => address.multiaddr)
+  }
+
   /**
    * Creates a PendingDial that wraps the underlying DialRequest
    */
@@ -383,3 +399,17 @@ export class DefaultDialer implements Startable, Dialer {
     }
   }
 }
+
+/**
+ * Type safe version of `list.filter(Boolean)`
+ */
+function isTruthy <T> (e: T | false | null | undefined): e is T {
+  return Boolean(e)
+}
+
+/**
+ * Returns a random string
+ */
+function randomId (): string {
+  return `${(parseInt(String(Math.random() * 1e9), 10)).toString()}${Date.now()}`
+}

src/connection-manager/index.ts

@@ -12,13 +12,14 @@ import type { Connection, MultiaddrConnection } from '@libp2p/interface-connecti
 import type { ConnectionManager, Dialer } from '@libp2p/interface-connection-manager'
 import * as STATUS from '@libp2p/interface-connection/status'
 import type { AddressSorter, PeerStore } from '@libp2p/interface-peer-store'
-import { multiaddr, Multiaddr, Resolver } from '@multiformats/multiaddr'
+import { isMultiaddr, multiaddr, Multiaddr, Resolver } from '@multiformats/multiaddr'
 import { PeerMap } from '@libp2p/peer-collections'
 import { TimeoutController } from 'timeout-abort-controller'
 import { KEEP_ALIVE } from '@libp2p/interface-peer-store/tags'
 import { RateLimiterMemory } from 'rate-limiter-flexible'
 import type { Metrics } from '@libp2p/interface-metrics'
 import type { Upgrader } from '@libp2p/interface-transport'
+import { getPeer } from '../get-peer.js'
 
 const log = logger('libp2p:connection-manager')
 
@@ -494,7 +495,18 @@ export class DefaultConnectionManager extends EventEmitter<ConnectionManagerEven
     return conns
   }
 
-  async openConnection (peerId: PeerId, options: AbortOptions = {}): Promise<Connection> {
+  async openConnection (peerIdOrMultiaddr: PeerId | Multiaddr, options: AbortOptions = {}): Promise<Connection> {
+    let peerId: PeerId
+
+    if (isPeerId(peerIdOrMultiaddr)) {
+      peerId = peerIdOrMultiaddr
+    } else if (isMultiaddr(peerIdOrMultiaddr)) {
+      const info = getPeer(peerIdOrMultiaddr)
+      peerId = info.id
+    } else {
+      throw errCode(new TypeError('Can only open connections to PeerIds or Multiaddrs'), codes.ERR_INVALID_PARAMETERS)
+    }
+
     log('dial to %p', peerId)
     const existingConnections = this.getConnections(peerId)
 
@@ -517,7 +529,7 @@ export class DefaultConnectionManager extends EventEmitter<ConnectionManagerEven
     }
 
     try {
-      const connection = await this.components.dialer.dial(peerId, options)
+      const connection = await this.components.dialer.dial(peerIdOrMultiaddr, options)
       let peerConnections = this.connections.get(peerId.toString())
 
       if (peerConnections == null) {

src/keychain/index.ts

@@ -118,7 +118,7 @@ export interface KeyChainComponents {
  */
 export class KeyChain implements Startable {
   private readonly components: KeyChainComponents
-  private init: KeyChainInit
+  private readonly init: KeyChainInit
   private started: boolean
 
   /**