Feat/libp2p tuning

packages/backend/package.json

@@ -24,7 +24,7 @@
     "lint-staged": "lint-staged --no-stash",
     "test-nest": "cross-env NODE_OPTIONS=\"--experimental-vm-modules\" DEBUG=ipfs:*,backend:* node_modules/jest/bin/jest.js --detectOpenHandles --forceExit   ./src/nest/**/*.spec.ts",
     "test": "cross-env NODE_OPTIONS=\"--experimental-vm-modules\" DEBUG=ipfs:*,backend:* jest --runInBand --verbose --testPathIgnorePatterns=\".src/(!?nodeTest*)|(.node_modules*)\" --detectOpenHandles --forceExit",
-    "test-ci": "cross-env NODE_OPTIONS=\"--experimental-vm-modules\" jest --runInBand --colors --ci --silent --verbose --detectOpenHandles --forceExit --testPathIgnorePatterns \"/node_modules/\" --testPathIgnorePatterns \"/dist/\" --testPathIgnorePatterns \"src/nest/.*\\.tor\\.spec\\.(t|j)s$\" --testPathIgnorePatterns \"src/nest/ipfs-file-manager/big-files\\.long\\.spec\\.ts$\"",
+    "test-ci": "cross-env NODE_OPTIONS=\"--experimental-vm-modules\" jest --runInBand --colors --ci --silent --verbose --detectOpenHandles --forceExit --testPathIgnorePatterns \"/node_modules/\" --testPathIgnorePatterns \"/dist/\" --testPathIgnorePatterns \"src/.*\\.tor\\.spec\\.(t|j)s$\" --testPathIgnorePatterns \"src/nest/ipfs-file-manager/big-files\\.long\\.spec\\.ts$\"",
     "test-ci-tor": "cross-env NODE_OPTIONS=\"--experimental-vm-modules\" jest --runInBand --colors --ci --verbose --detectOpenHandles --forceExit ./src/nest/**/*.tor.spec.ts",
     "test-ci-long-running": "cross-env DEBUG=backend:* NODE_OPTIONS=\"--experimental-vm-modules\" jest --colors --ci --verbose --detectOpenHandles --forceExit ./src/nest/**/*.long.spec.ts",
     "test-connect": "cross-env NODE_OPTIONS=\"--experimental-vm-modules\" DEBUG='libp2p:websockets*' jest ./src/nodeTest/* --verbose",

packages/backend/src/nest/libp2p/libp2p.auth.ts

@@ -116,6 +116,68 @@ export class Libp2pAuth {
     this.libp2pService.emit(eventName, ...args)
   }
 
+  private authConnectionDebugInfo(connection?: Connection): Record<string, unknown> {
+    if (connection == null) {
+      return {
+        connectionId: null,
+        direction: null,
+        status: null,
+        remoteAddr: null,
+        remotePeer: null,
+      }
+    }
+
+    return {
+      connectionId: connection.id,
+      direction: connection.direction,
+      status: connection.status,
+      remoteAddr: connection.remoteAddr?.toString() ?? null,
+      remotePeer: connection.remotePeer?.toString() ?? (connection as any).remotePeerId?.toString?.() ?? null,
+    }
+  }
+
+  private authCleanupCause(event: any): string {
+    switch (event?.type) {
+      case 'LOCAL_ERROR':
+        return 'local-error'
+      case 'REMOTE_ERROR':
+        return 'remote-error'
+      case 'ERROR':
+        return 'auth-error'
+      default:
+        return 'lfa-disconnected'
+    }
+  }
+
+  private logAuthCleanup(
+    event: string,
+    peerId: PeerId | string,
+    cause: string,
+    connection?: Connection,
+    extra: Record<string, unknown> = {}
+  ) {
+    this.logger.debug(
+      'p2p-auth-cleanup',
+      JSON.stringify({
+        event,
+        peerId: peerId.toString(),
+        cause,
+        ...this.authConnectionDebugInfo(connection),
+        ...extra,
+      })
+    )
+  }
+
+  private async closeDuplicateConnection(peerId: PeerId, connection: Connection) {
+    this.logAuthCleanup('duplicate-auth-connection-close', peerId, 'duplicate-auth-connection-close', connection)
+    try {
+      await connection.close({ signal: AbortSignal.timeout(15_000) })
+    } catch (error: any) {
+      this.logger.warn(`Failed to gracefully close duplicate auth connection with ${peerId.toString()}`, error)
+      connection.abort(error instanceof Error ? error : new Error(String(error)))
+    }
+  }
+
   // Process any connections that were buffered because we were waiting for a chain
   private async unblockConnections(conns: { peerId: PeerId; connection: Connection }[]) {
     if (this.joinStatus === JoinStatus.NOT_STARTED && this.sigChainService.activeChainTeamName != null) {
@@ -300,7 +362,21 @@ export class Libp2pAuth {
    */
   private async onPeerConnected(peerId: PeerId, connection: Connection) {
     if (this.authConnections.has(peerId.toString())) {
+      const existingConnection = this.peerConnections.get(peerId.toString())
+      const duplicateIsTrackedConnection = existingConnection?.id === connection.id
+      this.logAuthCleanup(
+        'duplicate-auth-connection',
+        peerId,
+        'duplicate-auth-connection-return-without-close',
+        connection,
+        {
+          existingConnection: this.authConnectionDebugInfo(existingConnection),
+        }
+      )
       this.logger.info(`Auth connection with ${peerId.toString()} already exists`)
+      if (!duplicateIsTrackedConnection) {
+        await this.closeDuplicateConnection(peerId, connection)
+      }
       return
     }
     if (this.joinStatus === JoinStatus.JOINING) {
@@ -319,6 +395,7 @@ export class Libp2pAuth {
     }
 
     this.logger.info(`Peer connected (direction = ${connection.direction})! (status = ${connection.status})`)
+    this.logAuthCleanup('peer:connect', peerId, 'auth-connection-start', connection)
     if (connection.status !== 'open') {
       this.logger.warn(`The connection with ${peerId.toString()} was not in an open state!`)
       return
@@ -377,6 +454,9 @@ export class Libp2pAuth {
 
     authConnection.on(LFAEvents.DISCONNECTED, event => {
       this.logger.info(`LFA Disconnected!`, event)
+      this.logAuthCleanup('lfa:disconnected', peerId, this.authCleanupCause(event), connection, {
+        lfaEventType: (event as any)?.type ?? null,
+      })
       this.libp2pService.emit(Libp2pEvents.AUTH_DISCONNECTED, {
         event,
         connection,
@@ -417,9 +497,17 @@ export class Libp2pAuth {
 
     // Handle errors from local or remote sources.
     authConnection.on(LFAEvents.LOCAL_ERROR, error => {
+      this.logAuthCleanup('lfa:local-error', peerId, 'local-error', connection, {
+        errorType: (error as any)?.type ?? null,
+        errorMessage: (error as any)?.message ?? null,
+      })
       this.emit(Libp2pEvents.AUTH_LOCAL_ERROR, { error, connection })
     })
     authConnection.on(LFAEvents.REMOTE_ERROR, error => {
+      this.logAuthCleanup('lfa:remote-error', peerId, 'remote-error', connection, {
+        errorType: (error as any)?.type ?? null,
+        errorMessage: (error as any)?.message ?? null,
+      })
       this.emit(Libp2pEvents.AUTH_REMOTE_ERROR, { error, connection })
     })
 
@@ -432,6 +520,9 @@ export class Libp2pAuth {
   }
 
   private async onPeerDisconnected(peerId: PeerId) {
+    this.logAuthCleanup('peer:disconnect', peerId, 'peer:disconnect', this.peerConnections.get(peerId.toString()), {
+      hadAuthConnection: this.authConnections.has(peerId.toString()),
+    })
     if (this.authConnections.has(peerId.toString())) {
       this.logger.warn(`Auth connection with ${peerId.toString()} was disconnected`)
       this.closeAuthConnection(peerId, false)
@@ -467,6 +558,18 @@ export class Libp2pAuth {
   public closeAuthConnection(peerId: PeerId | string, sendPeerDisconnect = true) {
     this.logger.info(`Attempting to close auth connection with ${peerId.toString()}`)
     const key = peerId.toString()
+    const peerConnection = this.peerConnections.get(key)
+    this.logAuthCleanup(
+      'closeAuthConnection',
+      key,
+      sendPeerDisconnect ? 'manual-close' : 'peer:disconnect',
+      peerConnection,
+      {
+        sendPeerDisconnect,
+        hadAuthConnection: this.authConnections.has(key),
+        hadPeerConnection: this.peerConnections.has(key),
+      }
+    )
 
     // Remove the stored connection (ephemeral streams are used for each message)
     if (this.peerConnections.has(key)) {

packages/backend/src/nest/libp2p/libp2p.debug.types.ts

@@ -0,0 +1,36 @@
+export interface ConnectionLifecycleDebugInfo {
+  connectionId: string
+  peerId: string
+  direction: string
+  remoteAddr: string
+  status: string
+  openedAtMs: number
+  openedAtIso: string
+  closedAtMs?: number
+  closedAtIso?: string
+  durationMs?: number
+  closeTrigger?: string
+}
+
+export interface ConnectionHealthConfig {
+  enabled: boolean
+  intervalMs: number
+  timeoutMs: number
+  failureThreshold: number
+}
+
+export type ConnectionHealthStatus = 'healthy' | 'degraded' | 'reconnecting'
+
+export interface ConnectionHealthDebugInfo {
+  peerId: string
+  status: ConnectionHealthStatus
+  failureCount: number
+  lastCheckedAtMs?: number
+  lastSuccessAtMs?: number
+  lastFailureAtMs?: number
+  lastRttMs?: number
+  lastErrorName?: string
+  lastErrorCode?: string
+  lastErrorMessage?: string
+  reconnecting?: boolean
+}

packages/backend/src/nest/libp2p/libp2p.service.spec.ts

@@ -16,6 +16,8 @@ describe('Libp2pService', () => {
   const localPeerAddress = '/dns4/local.onion/tcp/80/ws/p2p/local-peer'
   const remotePeerAddress = '/dns4/remote.onion/tcp/80/ws/p2p/remote-peer'
   const connectedRemotePeerAddress = '/dns4/connected-remote.onion/tcp/80/ws/p2p/remote-peer'
+  const validRemotePeerId = '12D3KooWKBpNVUqZjF4LDhkBMs2mRTSCYSaoBhCjQ9dfFJXogaWq'
+  const validRemotePeerAddress = `/dns4/remote.onion/tcp/80/ws/p2p/${validRemotePeerId}`
 
   beforeAll(async () => {
     module = await Test.createTestingModule({
@@ -100,4 +102,43 @@ describe('Libp2pService', () => {
     expect(hangUpPeers).toHaveBeenCalledWith([connectedRemotePeerAddress])
     expect(dialPeers).toHaveBeenCalledWith([remotePeerAddress])
   })
+
+  it('does not count ping stream contention as a health failure', async () => {
+    const contentionError = new Error('Too many outbound protocol streams for protocol "/ipfs/ping/1.0.0" - 1/1')
+    contentionError.name = 'TooManyOutboundProtocolStreamsError'
+    const ping = jest.fn<() => Promise<number>>().mockRejectedValue(contentionError)
+    const service = libp2pService as any
+    service.state = 'started'
+    service.connectionHealthConfig = {
+      enabled: true,
+      intervalMs: 75_000,
+      timeoutMs: 30_000,
+      failureThreshold: 3,
+    }
+    service.connectionHealthDebug.set(validRemotePeerId, {
+      peerId: validRemotePeerId,
+      status: 'degraded',
+      failureCount: 2,
+      lastFailureAtMs: 1,
+    })
+    service.libp2pInstance = {
+      getConnections: jest.fn(() => [{ status: 'open' }]),
+      services: {
+        ping: { ping },
+      },
+    }
+    const hangUpPeer = jest.spyOn(libp2pService, 'hangUpPeer').mockResolvedValue(undefined)
+
+    await service.checkPeerHealth(validRemotePeerId, validRemotePeerAddress)
+
+    expect(service.connectionHealthDebug.get(validRemotePeerId)).toEqual(
+      expect.objectContaining({
+        status: 'degraded',
+        failureCount: 2,
+        lastFailureAtMs: 1,
+      })
+    )
+    expect(hangUpPeer).not.toHaveBeenCalled()
+    service.libp2pInstance = null
+  })
 })