fix: debounce sending identify push messages (#3193)

To handle the case where several protocol handlers are regstered and
deregistered rapidly, introduce a small delay before actually
sending the push messages.

Fixes #2389

Author: achingbrain

packages/protocol-identify/src/consts.ts

@@ -13,3 +13,5 @@ export const MAX_IDENTIFY_MESSAGE_SIZE = 1024 * 8
 
 // https://github.com/libp2p/go-libp2p/blob/0385ec924bad172f74a74db09939e97c079b1420/p2p/protocol/identify/id.go#L47C7-L47C25
 export const MAX_PUSH_CONCURRENCY = 32
+
+export const PUSH_DEBOUNCE_MS = 1_000

packages/protocol-identify/src/identify-push.ts

@@ -1,5 +1,6 @@
 import { serviceCapabilities } from '@libp2p/interface'
 import { RecordEnvelope, PeerRecord } from '@libp2p/peer-record'
+import { debounce } from '@libp2p/utils/debounce'
 import { protocols } from '@multiformats/multiaddr'
 import drain from 'it-drain'
 import parallel from 'it-parallel'
@@ -9,7 +10,8 @@ import { fromString as uint8ArrayFromString } from 'uint8arrays/from-string'
 import { toString as uint8ArrayToString } from 'uint8arrays/to-string'
 import {
   MULTICODEC_IDENTIFY_PUSH_PROTOCOL_NAME,
-  MULTICODEC_IDENTIFY_PUSH_PROTOCOL_VERSION
+  MULTICODEC_IDENTIFY_PUSH_PROTOCOL_VERSION,
+  PUSH_DEBOUNCE_MS
 } from './consts.js'
 import { Identify as IdentifyMessage } from './pb/message.js'
 import { AbstractIdentify, consumeIdentifyMessage, defaultValues } from './utils.js'
@@ -20,6 +22,7 @@ import type { ConnectionManager } from '@libp2p/interface-internal'
 export class IdentifyPush extends AbstractIdentify implements Startable, IdentifyPushInterface {
   private readonly connectionManager: ConnectionManager
   private readonly concurrency: number
+  private _push: () => void
 
   constructor (components: IdentifyPushComponents, init: IdentifyPushInit = {}) {
     super(components, {
@@ -31,10 +34,14 @@ export class IdentifyPush extends AbstractIdentify implements Startable, Identif
     this.connectionManager = components.connectionManager
     this.concurrency = init.concurrency ?? defaultValues.concurrency
 
+    this._push = debounce(this.sendPushMessage.bind(this), init.debounce ?? PUSH_DEBOUNCE_MS)
+
     if ((init.runOnSelfUpdate ?? defaultValues.runOnSelfUpdate)) {
       // When self peer record changes, trigger identify-push
       components.events.addEventListener('self:peer:update', (evt) => {
-        void this.push().catch(err => { this.log.error(err) })
+        this.push().catch(err => {
+          this.log.error('error pushing updates to peers - %e', err)
+        })
       })
     }
   }
@@ -47,72 +54,80 @@ export class IdentifyPush extends AbstractIdentify implements Startable, Identif
    * Calls `push` on all peer connections
    */
   async push (): Promise<void> {
+    this._push()
+  }
+
+  private async sendPushMessage (): Promise<void> {
     // Do not try to push if we are not running
     if (!this.isStarted()) {
       return
     }
 
-    const listenAddresses = this.addressManager.getAddresses().map(ma => ma.decapsulateCode(protocols('p2p').code))
-    const peerRecord = new PeerRecord({
-      peerId: this.peerId,
-      multiaddrs: listenAddresses
-    })
-    const signedPeerRecord = await RecordEnvelope.seal(peerRecord, this.privateKey)
-    const supportedProtocols = this.registrar.getProtocols()
-    const peer = await this.peerStore.get(this.peerId)
-    const agentVersion = uint8ArrayToString(peer.metadata.get('AgentVersion') ?? uint8ArrayFromString(this.host.agentVersion))
-    const protocolVersion = uint8ArrayToString(peer.metadata.get('ProtocolVersion') ?? uint8ArrayFromString(this.host.protocolVersion))
-    const self = this
-
-    async function * pushToConnections (): AsyncGenerator<() => Promise<void>> {
-      for (const connection of self.connectionManager.getConnections()) {
-        const peer = await self.peerStore.get(connection.remotePeer)
-
-        if (!peer.protocols.includes(self.protocol)) {
-          continue
-        }
+    try {
+      const listenAddresses = this.addressManager.getAddresses().map(ma => ma.decapsulateCode(protocols('p2p').code))
+      const peerRecord = new PeerRecord({
+        peerId: this.peerId,
+        multiaddrs: listenAddresses
+      })
+      const signedPeerRecord = await RecordEnvelope.seal(peerRecord, this.privateKey)
+      const supportedProtocols = this.registrar.getProtocols()
+      const peer = await this.peerStore.get(this.peerId)
+      const agentVersion = uint8ArrayToString(peer.metadata.get('AgentVersion') ?? uint8ArrayFromString(this.host.agentVersion))
+      const protocolVersion = uint8ArrayToString(peer.metadata.get('ProtocolVersion') ?? uint8ArrayFromString(this.host.protocolVersion))
+      const self = this
+
+      async function * pushToConnections (): AsyncGenerator<() => Promise<void>> {
+        for (const connection of self.connectionManager.getConnections()) {
+          const peer = await self.peerStore.get(connection.remotePeer)
+
+          if (!peer.protocols.includes(self.protocol)) {
+            continue
+          }
 
-        yield async () => {
-          let stream: Stream | undefined
-          const signal = AbortSignal.timeout(self.timeout)
-
-          setMaxListeners(Infinity, signal)
-
-          try {
-            stream = await connection.newStream(self.protocol, {
-              signal,
-              runOnLimitedConnection: self.runOnLimitedConnection
-            })
-
-            const pb = pbStream(stream, {
-              maxDataLength: self.maxMessageSize
-            }).pb(IdentifyMessage)
-
-            await pb.write({
-              listenAddrs: listenAddresses.map(ma => ma.bytes),
-              signedPeerRecord: signedPeerRecord.marshal(),
-              protocols: supportedProtocols,
-              agentVersion,
-              protocolVersion
-            }, {
-              signal
-            })
-
-            await stream.close({
-              signal
-            })
-          } catch (err: any) {
-            // Just log errors
-            self.log.error('could not push identify update to peer', err)
-            stream?.abort(err)
+          yield async () => {
+            let stream: Stream | undefined
+            const signal = AbortSignal.timeout(self.timeout)
+
+            setMaxListeners(Infinity, signal)
+
+            try {
+              stream = await connection.newStream(self.protocol, {
+                signal,
+                runOnLimitedConnection: self.runOnLimitedConnection
+              })
+
+              const pb = pbStream(stream, {
+                maxDataLength: self.maxMessageSize
+              }).pb(IdentifyMessage)
+
+              await pb.write({
+                listenAddrs: listenAddresses.map(ma => ma.bytes),
+                signedPeerRecord: signedPeerRecord.marshal(),
+                protocols: supportedProtocols,
+                agentVersion,
+                protocolVersion
+              }, {
+                signal
+              })
+
+              await stream.close({
+                signal
+              })
+            } catch (err: any) {
+              // Just log errors
+              self.log.error('could not push identify update to peer', err)
+              stream?.abort(err)
+            }
           }
         }
       }
-    }
 
-    await drain(parallel(pushToConnections(), {
-      concurrency: this.concurrency
-    }))
+      await drain(parallel(pushToConnections(), {
+        concurrency: this.concurrency
+      }))
+    } catch (err: any) {
+      this.log.error('error pushing updates to peers - %e', err)
+    }
   }
 
   /**

packages/protocol-identify/src/index.ts

@@ -126,6 +126,15 @@ export interface IdentifyPushInit extends Omit<IdentifyInit, 'runOnConnectionOpe
    * @default 32
    */
   concurrency?: number
+
+  /**
+   * To prevent rapid flurries of network activity when addresses/protocols
+   * change rapidly in succession, debounce the sending of push message by this
+   * amount in ms
+   *
+   * @default 1_000
+   */
+  debounce?: number
 }
 
 export interface IdentifyComponents {

packages/protocol-identify/test/push.spec.ts

@@ -4,6 +4,7 @@ import { defaultLogger } from '@libp2p/logger'
 import { peerIdFromPrivateKey } from '@libp2p/peer-id'
 import { multiaddr } from '@multiformats/multiaddr'
 import { expect } from 'aegir/chai'
+import delay from 'delay'
 import { pair } from 'it-pair'
 import { pbStream } from 'it-protobuf-stream'
 import { TypedEventEmitter } from 'main-event'
@@ -164,4 +165,35 @@ describe('identify (push)', () => {
     expect(components.peerStore.patch.callCount).to.equal(0, 'patched peer when push timed out')
     expect(stream.abort.callCount).to.equal(1, 'did not abort stream')
   })
+
+  it('should debounce outgoing pushes', async () => {
+    identify = new IdentifyPush(components, {
+      timeout: 10
+    })
+
+    await start(identify)
+
+    components.addressManager.getAddresses.returns([multiaddr(`/ip4/123.123.123.123/tcp/123/p2p/${components.peerId}`)])
+    components.registrar.getProtocols.returns(['/super/fun/protocol'])
+    components.peerStore.get.withArgs(components.peerId).resolves({
+      id: components.peerId,
+      addresses: [],
+      protocols: [],
+      metadata: new Map(),
+      tags: new Map()
+    })
+
+    expect(components.connectionManager.getConnections.called).to.be.false()
+
+    identify.push()
+    identify.push()
+    identify.push()
+    identify.push()
+    identify.push()
+    identify.push()
+
+    await delay(2_000)
+
+    expect(components.connectionManager.getConnections.callCount).to.equal(1)
+  })
 })