fix: remove peer refresh

js-libp2p occasionally gets all peer routers to run a query for it's
own node id.

The only module that actually requires this is kad-dht in order to
insure it has peers when it runs a query and it does this anyway since
it's part of the spec.

libp2p/js-libp2p-kad-dht#457 also updates kad-dht
to pause all executed queries until the self-query has run.

Consequently this is redundant and just creates unecessary work.

Author: achingbrain

src/config.ts

@@ -24,62 +24,7 @@ const DefaultConfig: Partial<Libp2pInit> = {
   },
   transportManager: {
     faultTolerance: FaultTolerance.FATAL_ALL
-  },
-  peerRouting: {
-    refreshManager: {
-      enabled: true,
-      interval: 6e5,
-      bootDelay: 10e3
-    }
   }
-/*
-
-  nat: {
-    enabled: true,
-    ttl: 7200,
-    keepAlive: true
-  },
-  identify: {
-    protocolPrefix: 'ipfs',
-    host: {
-      agentVersion: AGENT_VERSION
-    },
-    // https://github.com/libp2p/go-libp2p/blob/8d2e54e1637041d5cf4fac1e531287560bd1f4ac/p2p/protocol/identify/id.go#L48
-    timeout: 60000,
-    maxInboundStreams: 1,
-    maxOutboundStreams: 1,
-    maxPushIncomingStreams: 1,
-    maxPushOutgoingStreams: 1,
-    maxObservedAddresses: 10
-  },
-  ping: {
-    protocolPrefix: 'ipfs',
-    // See https://github.com/libp2p/specs/blob/d4b5fb0152a6bb86cfd9ea/ping/ping.md?plain=1#L38-L43
-    // The dialing peer MUST NOT keep more than one outbound stream for the ping protocol per peer.
-    // The listening peer SHOULD accept at most two streams per peer since cross-stream behavior is
-    // non-linear and stream writes occur asynchronously. The listening peer may perceive the
-    // dialing peer closing and opening the wrong streams (for instance, closing stream B and
-    // opening stream A even though the dialing peer is opening stream B and closing stream A).
-    maxInboundStreams: 2,
-    maxOutboundStreams: 1,
-    timeout: 10000
-  },
-  fetch: {
-    protocolPrefix: 'libp2p',
-    maxInboundStreams: 1,
-    maxOutboundStreams: 1,
-    timeout: 10000
-  },
-  autonat: {
-    protocolPrefix: 'libp2p',
-    maxInboundStreams: 1,
-    maxOutboundStreams: 1,
-    timeout: 30000,
-    startupDelay: 5000,
-    refreshInterval: 60000
-  }
-
-*/
 }
 
 export function validateConfig <T extends ServiceMap = {}> (opts: RecursivePartial<Libp2pInit<T>>): Libp2pInit<T> {

src/index.ts

@@ -32,7 +32,6 @@ import type { Components } from './components.js'
 import type { Libp2p, ServiceMap } from '@libp2p/interface-libp2p'
 import type { KeyChainInit } from '@libp2p/keychain'
 import type { AddressManagerInit } from './address-manager/index.js'
-import type { PeerRoutingInit } from './peer-routing.js'
 import type { ConnectionManagerInit } from './connection-manager/index.js'
 import type { PersistentPeerStoreInit } from '@libp2p/peer-store'
 
@@ -81,11 +80,6 @@ export interface Libp2pInit<T extends ServiceMap = {}> {
    */
   peerStore: PersistentPeerStoreInit
 
-  /**
-   * libp2p Peer routing service configuration
-   */
-  peerRouting: PeerRoutingInit
-
   /**
    * keychain configuration
    */

src/libp2p.ts

@@ -147,7 +147,6 @@ export class Libp2pNode<T extends ServiceMap = {}> extends EventEmitter<Libp2pEv
     // Peer routers
     const peerRouters: PeerRouting[] = (init.peerRouters ?? []).map((fn, index) => this.configureComponent(`peer-router-${index}`, fn(this.components)))
     this.peerRouting = this.components.peerRouting = this.configureComponent('peerRouting', new DefaultPeerRouting(this.components, {
-      ...init.peerRouting,
       routers: peerRouters
     }))
 

src/peer-routing.ts

@@ -9,133 +9,31 @@ import {
 import merge from 'it-merge'
 import { pipe } from 'it-pipe'
 import first from 'it-first'
-import drain from 'it-drain'
 import filter from 'it-filter'
-import {
-  setDelayedInterval,
-  clearDelayedInterval
-// @ts-expect-error module with no types
-} from 'set-delayed-interval'
-import { setMaxListeners } from 'events'
 import type { PeerId } from '@libp2p/interface-peer-id'
 import type { PeerRouting } from '@libp2p/interface-peer-routing'
 import type { AbortOptions } from '@libp2p/interfaces'
-import type { Startable } from '@libp2p/interfaces/startable'
 import type { PeerInfo } from '@libp2p/interface-peer-info'
 import type { PeerStore } from '@libp2p/interface-peer-store'
-import { anySignal } from 'any-signal'
 
 const log = logger('libp2p:peer-routing')
 
-export interface RefreshManagerInit {
-  /**
-   * Whether to enable the Refresh manager
-   */
-  enabled?: boolean
-
-  /**
-   * Boot delay to start the Refresh Manager (in ms)
-   */
-  bootDelay?: number
-
-  /**
-   * Interval between each Refresh Manager run (in ms)
-   */
-  interval?: number
-
-  /**
-   * How long to let each refresh run (in ms)
-   */
-  timeout?: number
-}
-
 export interface PeerRoutingInit {
   routers?: PeerRouting[]
-  refreshManager?: RefreshManagerInit
 }
 
 export interface DefaultPeerRoutingComponents {
   peerId: PeerId
   peerStore: PeerStore
 }
 
-export class DefaultPeerRouting implements PeerRouting, Startable {
+export class DefaultPeerRouting implements PeerRouting {
   private readonly components: DefaultPeerRoutingComponents
   private readonly routers: PeerRouting[]
-  private readonly refreshManagerInit: RefreshManagerInit
-  private timeoutId?: ReturnType<typeof setTimeout>
-  private started: boolean
-  private abortController?: AbortController
 
   constructor (components: DefaultPeerRoutingComponents, init: PeerRoutingInit) {
     this.components = components
     this.routers = init.routers ?? []
-    this.refreshManagerInit = init.refreshManager ?? {}
-    this.started = false
-
-    this._findClosestPeersTask = this._findClosestPeersTask.bind(this)
-  }
-
-  isStarted (): boolean {
-    return this.started
-  }
-
-  /**
-   * Start peer routing service.
-   */
-  async start (): Promise<void> {
-    if (this.started || this.routers.length === 0 || this.timeoutId != null || this.refreshManagerInit.enabled === false) {
-      return
-    }
-
-    this.timeoutId = setDelayedInterval(
-      this._findClosestPeersTask, this.refreshManagerInit.interval, this.refreshManagerInit.bootDelay
-    )
-
-    this.started = true
-  }
-
-  /**
-   * Recurrent task to find closest peers and add their addresses to the Address Book.
-   */
-  async _findClosestPeersTask (): Promise<void> {
-    if (this.abortController != null) {
-      // we are already running the query
-      return
-    }
-
-    this.abortController = new AbortController()
-
-    const signal = anySignal([this.abortController.signal, AbortSignal.timeout(this.refreshManagerInit.timeout ?? 10e3)])
-
-    try {
-      // this controller may be used while dialing lots of peers so prevent MaxListenersExceededWarning
-      // appearing in the console
-      try {
-        setMaxListeners?.(Infinity, signal)
-      } catch {}
-
-      // nb getClosestPeers adds the addresses to the address book
-      await drain(this.getClosestPeers(this.components.peerId.toBytes(), { signal }))
-    } catch (err: any) {
-      log.error(err)
-    } finally {
-      this.abortController?.abort()
-      this.abortController = undefined
-      signal.clear()
-    }
-  }
-
-  /**
-   * Stop peer routing service.
-   */
-  async stop (): Promise<void> {
-    clearDelayedInterval(this.timeoutId)
-
-    // abort query if it is in-flight
-    this.abortController?.abort()
-
-    this.started = false
   }
 
   /**

test/peer-routing/peer-routing.node.ts

@@ -4,7 +4,6 @@ import { expect } from 'aegir/chai'
 import sinon from 'sinon'
 import delay from 'delay'
 import pDefer from 'p-defer'
-import pWaitFor from 'p-wait-for'
 import drain from 'it-drain'
 import all from 'it-all'
 import { multiaddr } from '@multiformats/multiaddr'
@@ -550,154 +549,4 @@ describe('peer-routing', () => {
       expect(peers).to.be.an('array').with.a.lengthOf(1).that.deep.equals(results)
     })
   })
-
-  describe('peer routing refresh manager service', () => {
-    let node: Libp2p<{ dht: DHT }>
-    let peerIds: PeerId[]
-
-    before(async () => {
-      peerIds = await Promise.all([
-        createPeerId(),
-        createPeerId()
-      ])
-    })
-
-    afterEach(async () => {
-      sinon.restore()
-
-      if (node != null) {
-        await node.stop()
-      }
-    })
-
-    it('should be enabled and start by default', async () => {
-      const results: PeerInfo[] = [
-        { id: peerIds[0], multiaddrs: [multiaddr('/ip4/30.0.0.1/tcp/2000')], protocols: [] },
-        { id: peerIds[1], multiaddrs: [multiaddr('/ip4/32.0.0.1/tcp/2000')], protocols: [] }
-      ]
-
-      node = await createNode({
-        config: createRoutingOptions({
-          start: false,
-          peerRouting: {
-            refreshManager: {
-              enabled: true,
-              bootDelay: 100
-            }
-          }
-        }),
-        started: false
-      })
-
-      if (node.services.dht == null) {
-        throw new Error('DHT not configured')
-      }
-
-      const peerStoreMergeStub = sinon.spy(node.peerStore, 'merge')
-      const dhtGetClosestPeersStub = sinon.stub(node.services.dht, 'getClosestPeers').callsFake(async function * () {
-        yield {
-          name: 'FINAL_PEER',
-          type: EventTypes.FINAL_PEER,
-          messageName: 'FIND_NODE',
-          messageType: MessageType.FIND_NODE,
-          from: peerIds[0],
-          peer: results[0]
-        }
-        yield {
-          name: 'FINAL_PEER',
-          type: EventTypes.FINAL_PEER,
-          messageName: 'FIND_NODE',
-          messageType: MessageType.FIND_NODE,
-          from: peerIds[0],
-          peer: results[1]
-        }
-      })
-
-      await node.start()
-
-      await pWaitFor(() => dhtGetClosestPeersStub.callCount === 1)
-      await pWaitFor(() => peerStoreMergeStub.callCount >= results.length)
-
-      const peer0 = await node.peerStore.get(peerIds[0])
-      expect(peer0.addresses.map(({ multiaddr }) => multiaddr.toString()))
-        .to.include.members(results[0].multiaddrs.map(ma => ma.toString()))
-
-      const peer1 = await node.peerStore.get(peerIds[1])
-      expect(peer1.addresses.map(({ multiaddr }) => multiaddr.toString()))
-        .to.include.members(results[1].multiaddrs.map(ma => ma.toString()))
-    })
-
-    it('should support being disabled', async () => {
-      node = await createNode({
-        config: createRoutingOptions({
-          start: false,
-          peerRouting: {
-            refreshManager: {
-              bootDelay: 100,
-              enabled: false
-            }
-          }
-        }),
-        started: false
-      })
-
-      if (node.services.dht == null) {
-        throw new Error('DHT not configured')
-      }
-
-      const dhtGetClosestPeersStub = sinon.stub(node.services.dht, 'getClosestPeers').callsFake(async function * () {
-        yield {
-          name: 'SENDING_QUERY',
-          type: EventTypes.SENDING_QUERY,
-          to: peerIds[0],
-          messageName: 'FIND_NODE',
-          messageType: MessageType.FIND_NODE
-        }
-        throw new Error('should not be called')
-      })
-
-      await node.start()
-      await delay(100)
-
-      expect(dhtGetClosestPeersStub.callCount === 0)
-    })
-
-    it('should start and run on interval', async () => {
-      node = await createNode({
-        config: createRoutingOptions({
-          start: false,
-          peerRouting: {
-            refreshManager: {
-              interval: 500,
-              bootDelay: 200
-            }
-          }
-        }),
-        started: false
-      })
-
-      if (node.services.dht == null) {
-        throw new Error('DHT not configured')
-      }
-
-      const dhtGetClosestPeersStub = sinon.stub(node.services.dht, 'getClosestPeers').callsFake(async function * () {
-        yield {
-          name: 'PEER_RESPONSE',
-          type: EventTypes.PEER_RESPONSE,
-          messageName: 'FIND_NODE',
-          messageType: MessageType.FIND_NODE,
-          from: peerIds[0],
-          closer: [
-            { id: peerIds[0], multiaddrs: [multiaddr('/ip4/30.0.0.1/tcp/2000')], protocols: [] }
-          ],
-          providers: []
-        }
-      })
-
-      await node.start()
-
-      // should run more than once
-      await pWaitFor(() => dhtGetClosestPeersStub.callCount === 2)
-    })
-  })
 })