From 6962836f351e180275682afb506ea2e77a43d6f9 Mon Sep 17 00:00:00 2001 From: taea Date: Thu, 2 Jul 2026 00:35:18 -0400 Subject: [PATCH 1/4] feat(libp2p): implement connection health checks and monitoring - Added connection health checks to monitor the status of peer connections. - Introduced environment variables for configuring connection health settings. - Enhanced connection lifecycle logging for better debugging. - Implemented retryable WebSocket upgrade error handling in WebSocket transport. - Updated WebSocket connection handling to include detailed logging on connection events. - Created a test suite to validate peer-to-peer disconnects over Tor with heartbeat checks. - Adjusted development environment variables to enable connection health monitoring. --- .../backend/src/nest/libp2p/libp2p.auth.ts | 103 +++++ .../src/nest/libp2p/libp2p.debug.types.ts | 36 ++ .../backend/src/nest/libp2p/libp2p.service.ts | 405 +++++++++++++++++- .../src/nest/websocketOverTor/index.ts | 171 +++++++- .../src/nest/websocketOverTor/listener.ts | 6 +- .../nest/websocketOverTor/socket-to-conn.ts | 87 +++- .../nodeTest/debugP2PDisconnects.tor.spec.ts | 258 +++++++++++ packages/desktop/.env.development | 11 +- 8 files changed, 1040 insertions(+), 37 deletions(-) create mode 100644 packages/backend/src/nest/libp2p/libp2p.debug.types.ts create mode 100644 packages/backend/src/nodeTest/debugP2PDisconnects.tor.spec.ts diff --git a/packages/backend/src/nest/libp2p/libp2p.auth.ts b/packages/backend/src/nest/libp2p/libp2p.auth.ts index 17d710cec8..3fee437e4a 100644 --- a/packages/backend/src/nest/libp2p/libp2p.auth.ts +++ b/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 { + 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 = {} + ) { + 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)) { diff --git a/packages/backend/src/nest/libp2p/libp2p.debug.types.ts b/packages/backend/src/nest/libp2p/libp2p.debug.types.ts new file mode 100644 index 0000000000..4da56c2a71 --- /dev/null +++ b/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 +} diff --git a/packages/backend/src/nest/libp2p/libp2p.service.ts b/packages/backend/src/nest/libp2p/libp2p.service.ts index 513288bbf3..29c3154f61 100644 --- a/packages/backend/src/nest/libp2p/libp2p.service.ts +++ b/packages/backend/src/nest/libp2p/libp2p.service.ts @@ -4,7 +4,7 @@ import { yamux } from '@chainsafe/libp2p-yamux' import { mplex } from '@libp2p/mplex' import { FaultTolerance } from '@libp2p/interface-transport' import { identify, identifyPush } from '@libp2p/identify' -import { type Libp2p } from '@libp2p/interface' +import { type Connection, type Libp2p } from '@libp2p/interface' import { kadDHT } from '@libp2p/kad-dht' import { peerIdFromString } from '@libp2p/peer-id' import { ping } from '@libp2p/ping' @@ -43,8 +43,68 @@ import { LocalDbService } from '../local-db/local-db.service' import { TimedQueue } from '../common/timed-queue' import { defaultLogger } from './libp2p.logger' import { QSSService } from '../qss/qss.service' +import type { + ConnectionHealthConfig, + ConnectionHealthDebugInfo, + ConnectionLifecycleDebugInfo, +} from './libp2p.debug.types' const CONNECTION_LIMIT = 20 +const HEARTBEAT_ABORT_ENV = 'LIBP2P_ABORT_CONNECTION_ON_PING_FAILURE' +const CONNECTION_MONITOR_PING_TIMEOUT_MIN_MS_ENV = 'LIBP2P_CONNECTION_MONITOR_PING_TIMEOUT_MIN_MS' +const PING_SERVICE_TIMEOUT_MS_ENV = 'LIBP2P_PING_TIMEOUT_MS' +const CONNECTION_HEALTH_CHECK_ENABLED_ENV = 'LIBP2P_CONNECTION_HEALTH_CHECK_ENABLED' +const CONNECTION_HEALTH_CHECK_INTERVAL_MS_ENV = 'LIBP2P_CONNECTION_HEALTH_CHECK_INTERVAL_MS' +const CONNECTION_HEALTH_CHECK_TIMEOUT_MS_ENV = 'LIBP2P_CONNECTION_HEALTH_CHECK_TIMEOUT_MS' +const CONNECTION_HEALTH_CHECK_FAILURE_THRESHOLD_ENV = 'LIBP2P_CONNECTION_HEALTH_CHECK_FAILURE_THRESHOLD' +const CONNECTION_HEALTH_CHECK_DEFAULT_INTERVAL_MS = 75_000 +const CONNECTION_HEALTH_CHECK_DEFAULT_TIMEOUT_MS = 20_000 +const CONNECTION_HEALTH_CHECK_DEFAULT_FAILURE_THRESHOLD = 3 +const REDIAL_QUEUE_CONCURRENCY = 4 +const REDIAL_QUEUE_BACKOFF_FACTOR = 1.6 +const REDIAL_QUEUE_FUZZ_FACTOR = 0.25 +const REDIAL_QUEUE_BASE_DELAY_MS = 15_000 +const REDIAL_QUEUE_MAX_DELAY_MS = 180_000 +const DIAL_QUEUE_INITIAL_STAGGER_MS = 1_000 + +const booleanEnv = (value: string | undefined, defaultValue: boolean): boolean => { + if (value == null || value.trim() === '') return defaultValue + + switch (value.trim().toLowerCase()) { + case 'false': + case '0': + case 'no': + return false + case 'true': + case '1': + case 'yes': + return true + default: + return defaultValue + } +} + +const numberEnv = (value: string | undefined): number | undefined => { + if (value == null || value.trim() === '') return undefined + + const parsed = Number(value) + return Number.isFinite(parsed) && parsed > 0 ? parsed : undefined +} + +const connectionHealthConfigFromEnv = (): ConnectionHealthConfig => ({ + enabled: booleanEnv(process.env[CONNECTION_HEALTH_CHECK_ENABLED_ENV], true), + intervalMs: + numberEnv(process.env[CONNECTION_HEALTH_CHECK_INTERVAL_MS_ENV]) ?? CONNECTION_HEALTH_CHECK_DEFAULT_INTERVAL_MS, + timeoutMs: + numberEnv(process.env[CONNECTION_HEALTH_CHECK_TIMEOUT_MS_ENV]) ?? CONNECTION_HEALTH_CHECK_DEFAULT_TIMEOUT_MS, + failureThreshold: Math.max( + 1, + Math.floor( + numberEnv(process.env[CONNECTION_HEALTH_CHECK_FAILURE_THRESHOLD_ENV]) ?? + CONNECTION_HEALTH_CHECK_DEFAULT_FAILURE_THRESHOLD + ) + ), +}) export enum Libp2pState { Started = 'started', @@ -68,6 +128,17 @@ export class Libp2pService extends EventEmitter implements OnModuleDestroy { public state: Libp2pState = Libp2pState.Stopped private torBootstrap?: TorBootstrapProvider private waitingForTorBootstrapToResumeDialQueue = false + private connectionLifecycleDebug = new Map() + private connectionHealthDebug = new Map() + private connectionHealthConfig: ConnectionHealthConfig = { + enabled: true, + intervalMs: CONNECTION_HEALTH_CHECK_DEFAULT_INTERVAL_MS, + timeoutMs: CONNECTION_HEALTH_CHECK_DEFAULT_TIMEOUT_MS, + failureThreshold: CONNECTION_HEALTH_CHECK_DEFAULT_FAILURE_THRESHOLD, + } + private connectionHealthChecksInFlight = new Set() + private pendingCloseTriggersByPeer = new Map() + private _connectionHealthInterval: NodeJS.Timeout | null = null private logger = createLogger(Libp2pService.name) @@ -84,11 +155,11 @@ export class Libp2pService extends EventEmitter implements OnModuleDestroy { this.dialedPeers = new Set() this.redialQueue = new TimedQueue({ start: true, - concurrency: 10, - backoffFactor: 1.25, - fuzzFactor: 0.05, - baseDelayMs: 8_000, - maxDelayMs: 20_000, + concurrency: REDIAL_QUEUE_CONCURRENCY, + backoffFactor: REDIAL_QUEUE_BACKOFF_FACTOR, + fuzzFactor: REDIAL_QUEUE_FUZZ_FACTOR, + baseDelayMs: REDIAL_QUEUE_BASE_DELAY_MS, + maxDelayMs: REDIAL_QUEUE_MAX_DELAY_MS, rolloverAtMaxDelay: false, }) } @@ -100,9 +171,219 @@ export class Libp2pService extends EventEmitter implements OnModuleDestroy { this.state = state } + private connectionDebugInfo( + connection: Connection, + existing?: ConnectionLifecycleDebugInfo + ): ConnectionLifecycleDebugInfo { + const connectionId = String(connection.id ?? 'unknown') + const timelineOpen = Number((connection as any).timeline?.open) + const openedAtMs = existing?.openedAtMs ?? (Number.isFinite(timelineOpen) ? timelineOpen : Date.now()) + const openedAtIso = + existing?.openedAtIso ?? DateTime.fromMillis(openedAtMs).toUTC().toISO() ?? new Date(openedAtMs).toISOString() + + return { + connectionId, + peerId: connection.remotePeer?.toString() ?? 'unknown', + direction: String(connection.direction ?? 'unknown'), + remoteAddr: connection.remoteAddr?.toString() ?? 'unknown', + status: String(connection.status ?? 'unknown'), + openedAtMs, + openedAtIso, + closedAtMs: existing?.closedAtMs, + closedAtIso: existing?.closedAtIso, + durationMs: existing?.durationMs, + closeTrigger: existing?.closeTrigger, + } + } + + private logConnectionLifecycle( + event: string, + info: ConnectionLifecycleDebugInfo, + extra: Record = {} + ) { + this.logger.debug('p2p-connection-lifecycle', JSON.stringify({ event, ...info, ...extra })) + } + + private trackConnectionOpen(connection: Connection) { + const info = this.connectionDebugInfo(connection) + this.connectionLifecycleDebug.set(info.connectionId, info) + this.connectionHealthDebug.set(info.peerId, { + peerId: info.peerId, + status: 'healthy', + failureCount: 0, + lastCheckedAtMs: Date.now(), + }) + this.logConnectionLifecycle('connection:open', info) + } + + private trackConnectionClose(connection: Connection) { + const connectionId = String(connection.id ?? 'unknown') + const existing = this.connectionLifecycleDebug.get(connectionId) + const peerId = connection.remotePeer?.toString() ?? existing?.peerId ?? 'unknown' + const closedAtMs = Date.now() + const info = { + ...this.connectionDebugInfo(connection, existing), + closedAtMs, + closedAtIso: DateTime.fromMillis(closedAtMs).toUTC().toISO() ?? new Date(closedAtMs).toISOString(), + durationMs: closedAtMs - (existing?.openedAtMs ?? closedAtMs), + closeTrigger: this.pendingCloseTriggersByPeer.get(peerId) ?? existing?.closeTrigger ?? 'libp2p-connection-close', + } + + this.connectionLifecycleDebug.set(connectionId, info) + this.logger.warn('p2p-connection-lifecycle', JSON.stringify({ event: 'connection:close', ...info })) + } + + private connectionDebugInfosForPeer(peerId: string): ConnectionLifecycleDebugInfo[] { + return [...this.connectionLifecycleDebug.values()].filter(info => info.peerId === peerId) + } + + private clearClosedConnectionDebugForPeer(peerId: string) { + for (const [connectionId, info] of this.connectionLifecycleDebug.entries()) { + if (info.peerId === peerId && info.closedAtMs != null) { + this.connectionLifecycleDebug.delete(connectionId) + } + } + } + + private markPeerCloseTrigger(peerId: string, closeTrigger: string) { + this.pendingCloseTriggersByPeer.set(peerId, closeTrigger) + } + + private logConnectionHealth(event: string, peerId: string, extra: Record = {}) { + this.logger.debug( + 'p2p-connection-health', + JSON.stringify({ + event, + peerId, + config: this.connectionHealthConfig, + health: this.connectionHealthDebug.get(peerId) ?? null, + ...extra, + }) + ) + } + + private startConnectionHealthChecks() { + if (!this.connectionHealthConfig.enabled) { + this.logger.debug('p2p-connection-health', JSON.stringify({ event: 'health:disabled' })) + return + } + if (this._connectionHealthInterval != null) return + + this.logger.debug( + 'p2p-connection-health', + JSON.stringify({ event: 'health:start', config: this.connectionHealthConfig }) + ) + this._connectionHealthInterval = setInterval(() => { + void this.checkConnectionHealth() + }, this.connectionHealthConfig.intervalMs) + this._connectionHealthInterval.unref?.() + } + + private stopConnectionHealthChecks() { + if (this._connectionHealthInterval != null) { + clearInterval(this._connectionHealthInterval) + this._connectionHealthInterval = null + } + this.connectionHealthChecksInFlight.clear() + } + + private async checkConnectionHealth() { + if (this.state !== Libp2pState.Started || this.libp2pInstance == null) { + return + } + + for (const [peerId, peer] of this.connectedPeers.entries()) { + await this.checkPeerHealth(peerId, peer.address) + } + } + + private async checkPeerHealth(peerId: string, peerAddress: string) { + if (this.libp2pInstance == null) return + if (this.connectionHealthChecksInFlight.has(peerId)) return + + const peerIdObject = peerIdFromString(peerId) + const openConnections = this.libp2pInstance.getConnections(peerIdObject).filter(conn => conn.status === 'open') + if (openConnections.length === 0) { + this.logConnectionHealth('health:skip-no-open-connection', peerId, { peerAddress }) + return + } + + this.connectionHealthChecksInFlight.add(peerId) + const startedAtMs = Date.now() + const previous = this.connectionHealthDebug.get(peerId) + const pingService = this.libp2pInstance.services.ping as + | { ping: (peer: ReturnType, options: { signal: AbortSignal }) => Promise } + | undefined + + if (pingService == null) { + this.logConnectionHealth('health:skip-no-ping-service', peerId, { peerAddress }) + this.connectionHealthChecksInFlight.delete(peerId) + return + } + + try { + this.logConnectionHealth('health:ping:start', peerId, { + peerAddress, + openConnectionCount: openConnections.length, + }) + const rtt = await pingService.ping(peerIdObject, { + signal: AbortSignal.timeout(this.connectionHealthConfig.timeoutMs), + }) + this.connectionHealthDebug.set(peerId, { + peerId, + status: 'healthy', + failureCount: 0, + lastCheckedAtMs: Date.now(), + lastSuccessAtMs: Date.now(), + lastRttMs: rtt, + }) + this.logConnectionHealth('health:ping:success', peerId, { + peerAddress, + rttMs: rtt, + durationMs: Date.now() - startedAtMs, + }) + } catch (error: any) { + const failureCount = (previous?.failureCount ?? 0) + 1 + const status = failureCount >= this.connectionHealthConfig.failureThreshold ? 'reconnecting' : 'degraded' + this.connectionHealthDebug.set(peerId, { + peerId, + status, + failureCount, + lastCheckedAtMs: Date.now(), + lastSuccessAtMs: previous?.lastSuccessAtMs, + lastFailureAtMs: Date.now(), + lastRttMs: previous?.lastRttMs, + lastErrorName: error?.name, + lastErrorCode: error?.code, + lastErrorMessage: error?.message, + reconnecting: status === 'reconnecting', + }) + this.logConnectionHealth('health:ping:failure', peerId, { + peerAddress, + durationMs: Date.now() - startedAtMs, + errorName: error?.name, + errorCode: error?.code, + errorMessage: error?.message, + failureCount, + }) + + if (failureCount >= this.connectionHealthConfig.failureThreshold) { + this.markPeerCloseTrigger(peerId, 'health-check-threshold') + this.logConnectionHealth('health:redial', peerId, { + peerAddress, + failureThreshold: this.connectionHealthConfig.failureThreshold, + }) + await this.hangUpPeer(peerAddress, true) + } + } finally { + this.connectionHealthChecksInFlight.delete(peerId) + } + } + public onModuleDestroy() { this.logger.log('Module is being destroyed') this.redialQueue.stop(true) + this.stopConnectionHealthChecks() if (this._dialQueueInterval) { clearInterval(this._dialQueueInterval) this._dialQueueInterval = null @@ -215,7 +496,19 @@ export class Libp2pService extends EventEmitter implements OnModuleDestroy { this.logger.debug('Local Address', this.localAddress) this.logger.debug(peerAddresses.length, dialable.length) - await Promise.all(dialable.map(addr => this.dialPeer(addr))) + let queuedFirstAttempts = 0 + for (const addr of dialable) { + if (this.redialQueue.hasTask(addr)) continue + const delayMs = this.dialedPeers.has(addr) ? undefined : queuedFirstAttempts * DIAL_QUEUE_INITIAL_STAGGER_MS + if (!this.dialedPeers.has(addr)) { + queuedFirstAttempts += 1 + } + await this.redialQueue.enqueue({ + key: addr, + delayMs, + task: async () => this.dialPeer(addr, { throwOnError: true, redialOnError: false }), + }) + } } public addPeersToDialQueue = async () => { @@ -227,13 +520,17 @@ export class Libp2pService extends EventEmitter implements OnModuleDestroy { return } + let queuedFirstAttempts = 0 for (const addr of sortedPeers) { const peerId = addr.split('/').pop() if (peerId === undefined) continue if (addr === this.localAddress) continue if (this.redialQueue.hasTask(addr)) continue if (this.connectedPeers.has(peerId)) continue - const delayMs = this.dialedPeers.has(addr) ? undefined : 0 // dial immediately if this is our first attempt at dialing this address + const delayMs = this.dialedPeers.has(addr) ? undefined : queuedFirstAttempts * DIAL_QUEUE_INITIAL_STAGGER_MS + if (!this.dialedPeers.has(addr)) { + queuedFirstAttempts += 1 + } await this.redialQueue.enqueue({ key: addr, @@ -386,6 +683,7 @@ export class Libp2pService extends EventEmitter implements OnModuleDestroy { try { const ma = multiaddr(peerAddress) const peerId = peerIdFromString(ma.getPeerId()!) + this.markPeerCloseTrigger(peerId.toString(), redial ? 'local-hangup-redial' : 'local-hangup') this.logger.debug('Disconnecting auth service gracefully') this.authService?.closeAuthConnection(peerId) @@ -463,6 +761,58 @@ export class Libp2pService extends EventEmitter implements OnModuleDestroy { let libp2p: Libp2p + const connectionMonitorPingTimeoutMinMs = numberEnv(process.env[CONNECTION_MONITOR_PING_TIMEOUT_MIN_MS_ENV]) + const pingServiceTimeoutMs = numberEnv(process.env[PING_SERVICE_TIMEOUT_MS_ENV]) + this.connectionHealthConfig = connectionHealthConfigFromEnv() + const connectionMonitorConfig = { + abortConnectionOnPingFailure: booleanEnv(process.env[HEARTBEAT_ABORT_ENV], false), + pingInterval: 60_000, + enabled: true, + ...(connectionMonitorPingTimeoutMinMs == null + ? {} + : { + pingTimeout: { + minTimeout: connectionMonitorPingTimeoutMinMs, + }, + }), + } satisfies ConnectionMonitorInit + + this.logger.debug( + 'p2p-heartbeat-monitor-config', + JSON.stringify({ + ...connectionMonitorConfig, + env: HEARTBEAT_ABORT_ENV, + envValue: process.env[HEARTBEAT_ABORT_ENV] ?? null, + pingTimeoutMinEnv: CONNECTION_MONITOR_PING_TIMEOUT_MIN_MS_ENV, + pingTimeoutMinEnvValue: process.env[CONNECTION_MONITOR_PING_TIMEOUT_MIN_MS_ENV] ?? null, + }) + ) + this.logger.debug( + 'p2p-ping-service-config', + JSON.stringify({ + timeout: pingServiceTimeoutMs ?? null, + env: PING_SERVICE_TIMEOUT_MS_ENV, + envValue: process.env[PING_SERVICE_TIMEOUT_MS_ENV] ?? null, + }) + ) + this.logger.debug( + 'p2p-connection-health', + JSON.stringify({ + event: 'health:config', + config: this.connectionHealthConfig, + env: { + enabled: CONNECTION_HEALTH_CHECK_ENABLED_ENV, + enabledValue: process.env[CONNECTION_HEALTH_CHECK_ENABLED_ENV] ?? null, + intervalMs: CONNECTION_HEALTH_CHECK_INTERVAL_MS_ENV, + intervalMsValue: process.env[CONNECTION_HEALTH_CHECK_INTERVAL_MS_ENV] ?? null, + timeoutMs: CONNECTION_HEALTH_CHECK_TIMEOUT_MS_ENV, + timeoutMsValue: process.env[CONNECTION_HEALTH_CHECK_TIMEOUT_MS_ENV] ?? null, + failureThreshold: CONNECTION_HEALTH_CHECK_FAILURE_THRESHOLD_ENV, + failureThresholdValue: process.env[CONNECTION_HEALTH_CHECK_FAILURE_THRESHOLD_ENV] ?? null, + }, + }) + ) + this.logger.debug(`Creating libp2p`) try { libp2p = await createLibp2p({ @@ -481,11 +831,7 @@ export class Libp2pService extends EventEmitter implements OnModuleDestroy { }, privateKey: params.peerId.privKey, addresses: { listen: params.listenAddresses }, - connectionMonitor: { - abortConnectionOnPingFailure: true, - pingInterval: 60_000, - enabled: true, - } satisfies ConnectionMonitorInit, + connectionMonitor: connectionMonitorConfig, connectionProtector: params.useConnectionProtector || params.useConnectionProtector == null ? preSharedKey({ psk: params.psk }) @@ -531,7 +877,7 @@ export class Libp2pService extends EventEmitter implements OnModuleDestroy { }, services: { auth: libp2pAuth(this.sigchainService, this.qssService, this), - ping: ping(), + ping: ping(pingServiceTimeoutMs == null ? {} : { timeout: pingServiceTimeoutMs }), pubsub: gossipsub({ // neccessary to run a single peer allowPublishToZeroTopicPeers: true, @@ -582,6 +928,7 @@ export class Libp2pService extends EventEmitter implements OnModuleDestroy { this.serverIoProvider.io.emit(SocketEvents.CONNECTION_PROCESS_INFO, ConnectionProcessInfo.INITIALIZING_LIBP2P) this.libp2pInstance.addEventListener('connection:open', openEvent => { + this.trackConnectionOpen(openEvent.detail) this.logger.debug( `Opened connection with ID ${openEvent.detail.id} with peer`, openEvent.detail.remotePeer.toString() @@ -607,6 +954,7 @@ export class Libp2pService extends EventEmitter implements OnModuleDestroy { }) this.libp2pInstance.addEventListener('connection:close', event => { + this.trackConnectionClose(event.detail) this.logger.warn(`Connection with ID ${event.detail.id} closing with peer`, event.detail.remotePeer.toString()) }) @@ -632,6 +980,18 @@ export class Libp2pService extends EventEmitter implements OnModuleDestroy { return } const localPeerId = peerId.peerId.toString() + this.logger.debug( + 'p2p-peer-lifecycle', + JSON.stringify({ + event: 'peer:connect', + peerId: remotePeerId, + localPeerId, + connectionCount: connection?.length ?? 0, + connections: connection?.map(conn => + this.connectionDebugInfo(conn, this.connectionLifecycleDebug.get(String(conn.id ?? 'unknown'))) + ), + }) + ) this.logger.debug(`Connection established with ${remotePeerId}`, JSON.stringify(connection)) this.logger.debug(`${localPeerId} connected to ${remotePeerId}`) @@ -676,6 +1036,17 @@ export class Libp2pService extends EventEmitter implements OnModuleDestroy { this.libp2pInstance.addEventListener('peer:disconnect', async event => { const remotePeerId = event.detail.toString() const localPeerId = peerId.peerId.toString() + const connectionDebugInfos = this.connectionDebugInfosForPeer(remotePeerId) + this.logger.debug( + 'p2p-peer-lifecycle', + JSON.stringify({ + event: 'peer:disconnect', + peerId: remotePeerId, + localPeerId, + closeTrigger: this.pendingCloseTriggersByPeer.get(remotePeerId) ?? 'peer-disconnect-after-connection-close', + connections: connectionDebugInfos, + }) + ) this.logger.debug(`Connection closed with ${remotePeerId}`, JSON.stringify(event)) this.logger.debug(`${localPeerId} disconnected from ${remotePeerId}`) if (!this.libp2pInstance) { @@ -693,6 +1064,10 @@ export class Libp2pService extends EventEmitter implements OnModuleDestroy { const connectionDuration: number = connectionEndTime - connectionStartTime this.connectedPeers.delete(remotePeerId) + this.pendingCloseTriggersByPeer.delete(remotePeerId) + this.connectionHealthDebug.delete(remotePeerId) + this.connectionHealthChecksInFlight.delete(remotePeerId) + this.clearClosedConnectionDebugForPeer(remotePeerId) this.logger.debug(`${localPeerId} is now connected to ${this.connectedPeers.size} peers`) const peerPrevStats = await this.localDbService.getPeerStats(remotePeerId) @@ -725,6 +1100,7 @@ export class Libp2pService extends EventEmitter implements OnModuleDestroy { this.setState(Libp2pState.Starting) await this.libp2pInstance.start() this.setState(Libp2pState.Started) + this.startConnectionHealthChecks() this.logger.debug('Queueing peers for initial dialing') await this.resumeDialQueueWhenTorReady() @@ -767,6 +1143,7 @@ export class Libp2pService extends EventEmitter implements OnModuleDestroy { clearInterval(this._connectedPeersInterval) this.redialQueue.stop(true) + this.stopConnectionHealthChecks() await this.hangUpPeers() await this.libp2pInstance?.stop() diff --git a/packages/backend/src/nest/websocketOverTor/index.ts b/packages/backend/src/nest/websocketOverTor/index.ts index 4d9914ea96..25aa2220bf 100644 --- a/packages/backend/src/nest/websocketOverTor/index.ts +++ b/packages/backend/src/nest/websocketOverTor/index.ts @@ -29,6 +29,7 @@ import type { ComponentLogger, Logger, Connection, + MultiaddrConnection, OutboundConnectionUpgradeEvents, Metrics, CounterGroup, @@ -37,7 +38,7 @@ import type { Multiaddr } from '@multiformats/multiaddr' import type { Server } from 'http' import type { DuplexWebSocket } from 'it-ws/duplex' import type { ProgressEvent } from 'progress-events' -import type { ClientOptions } from 'ws' +import type { ClientOptions, CloseEvent } from 'ws' import http from 'node:http' import https from 'node:https' import { QuietLibp2pLogger } from '../libp2p/libp2p.logger' @@ -87,6 +88,48 @@ export interface WebSocketsMetrics { export type WebSocketsDialEvents = OutboundConnectionUpgradeEvents | ProgressEvent<'websockets:open-connection'> +const EXPECTED_SERVER_RESPONSES = new Set(['Unexpected server response: 404', 'Unexpected server response: 503']) +const CONNECT_PHASE_RETRYABLE_SERVER_RESPONSES = new Set(['Unexpected server response: 503']) + +export class RetryableWebSocketConnectError extends ConnectionFailedError { + public readonly code = 'ERR_RETRYABLE_WEBSOCKET_CONNECT' + public readonly retryable = true + public readonly phase = 'connect' + public readonly cause: unknown + + constructor(message: string, cause: unknown) { + super(message) + this.name = 'RetryableWebSocketConnectError' + this.cause = cause + } +} + +export class RetryableWebSocketUpgradeError extends ConnectionFailedError { + public readonly code = 'ERR_RETRYABLE_WEBSOCKET_UPGRADE' + public readonly retryable = true + public readonly phase: string + public readonly cause: unknown + + constructor(message: string, phase: string, cause: unknown) { + super(message) + this.name = 'RetryableWebSocketUpgradeError' + this.phase = phase + this.cause = cause + } +} + +const isRetryableUpgradeFailure = (err: any): boolean => { + const message = String(err?.message ?? '').toLowerCase() + return ( + err?.code === 'ERR_UNEXPECTED_EOF' || + err?.code === 'ABORT_ERR' || + ['UnexpectedEOFError', 'AbortError', 'TimeoutError'].includes(err?.name) || + message.includes('unexpected end of input') || + message.includes('read aborted') || + message.includes('websocket was closed before the connection was established') + ) +} + export class WebSockets implements Transport { private readonly init: WebSocketsInit private readonly logger: ComponentLogger @@ -118,19 +161,96 @@ export class WebSockets implements Transport { const _log = this.components.logger.forComponent(`libp2p:websockets:dial:${ma.getPeerId()}`) as QuietLibp2pLogger _log('dialing %s', ma) options = options ?? ({} as DialTransportOptions) + const peerId = ma.getPeerId() ?? 'unknown' + const dialStartedAtMs = Date.now() + const dialId = `${peerId}:${dialStartedAtMs}` + let phase: 'connect' | 'upgrade' = 'connect' + let socket: DuplexWebSocket | undefined + let maConn: MultiaddrConnection | undefined + + _log( + 'p2p-websocket-dial %s', + JSON.stringify({ + event: 'dial:start', + dialId, + peerId, + remoteAddr: ma.toString(), + startedAtMs: dialStartedAtMs, + }) + ) - const socket = await this._connect(ma, options) - const maConn = socketToMaConn(socket, ma, { - logger: this.logger, - metrics: this.metrics?.dialerEvents, - signal: options.signal, - }) - _log('new outbound connection %s', maConn.remoteAddr) - - const conn = await options.upgrader.upgradeOutbound(maConn, options) - _log('outbound connection %s upgraded', maConn.remoteAddr) - - return conn + try { + socket = await this._connect(ma, options) + _log( + 'p2p-websocket-dial %s', + JSON.stringify({ event: 'dial:websocket-connected', dialId, peerId, ageMs: Date.now() - dialStartedAtMs }) + ) + maConn = socketToMaConn(socket, ma, { + logger: this.logger, + metrics: this.metrics?.dialerEvents, + signal: options.signal, + }) + _log('new outbound connection %s', maConn.remoteAddr) + phase = 'upgrade' + _log( + 'p2p-websocket-dial %s', + JSON.stringify({ event: 'dial:upgrade:start', dialId, peerId, remoteAddr: maConn.remoteAddr.toString() }) + ) + + const conn = await options.upgrader.upgradeOutbound(maConn, options) + _log('outbound connection %s upgraded', maConn.remoteAddr) + _log( + 'p2p-websocket-dial %s', + JSON.stringify({ + event: 'dial:upgrade:complete', + dialId, + peerId, + connectionId: conn.id, + direction: conn.direction, + status: conn.status, + ageMs: Date.now() - dialStartedAtMs, + }) + ) + + return conn + } catch (err: any) { + const retryable = err?.retryable === true || (phase === 'upgrade' && isRetryableUpgradeFailure(err)) + const errorToThrow = retryable + ? err?.retryable === true + ? err + : new RetryableWebSocketUpgradeError( + `Retryable Tor WebSocket upgrade failure for ${ma.toString()}: ${err?.message ?? err}`, + phase, + err + ) + : err + const logDialFailure = retryable ? _log.warn.bind(_log) : _log.error.bind(_log) + logDialFailure( + 'p2p-websocket-dial %s', + JSON.stringify({ + event: 'dial:failed', + dialId, + peerId, + phase, + retryable, + ageMs: Date.now() - dialStartedAtMs, + errorName: errorToThrow?.name, + errorCode: errorToThrow?.code, + errorMessage: errorToThrow?.message, + causeName: err?.name, + causeCode: err?.code, + causeMessage: err?.message, + }) + ) + if (phase === 'upgrade' && maConn != null) { + maConn.abort(errorToThrow instanceof Error ? errorToThrow : new Error(String(errorToThrow))) + } else if (socket != null) { + socket.close().catch(closeErr => { + _log.error('error closing raw socket after dial failure', closeErr) + }) + } + throw errorToThrow + } } async _connect(ma: Multiaddr, options: DialTransportOptions): Promise { @@ -146,6 +266,8 @@ export class WebSockets implements Transport { const errorPromise = pDefer() const addr = `${toUri(ma)}/?remoteAddress=${encodeURIComponent(this.init.localAddress)}` _log('CONNECTING TO ADDR', addr) + const peerId = ma.getPeerId() ?? 'unknown' + const rawSocketStartedAtMs = Date.now() const rawSocket = connect(addr, this.init) rawSocket.socket.addEventListener('error', errorEvent => { // the WebSocket.ErrorEvent type doesn't actually give us any useful @@ -153,9 +275,13 @@ export class WebSockets implements Transport { // https://developer.mozilla.org/en-US/docs/Web/API/WebSocket/error_event this.metrics?.dialerEvents.increment({ error: true }) const message = `Could not connect to ${ma.toString()}: ${errorEvent.message}` - // 404 errors are plentiful and logging them as errors muddies the logs - if (errorEvent.message === 'Unexpected server response: 404') { + // Expected HTTP responses are common while peers are offline, starting, or not yet listening. + if (EXPECTED_SERVER_RESPONSES.has(errorEvent.message)) { _log.warn(message) + if (CONNECT_PHASE_RETRYABLE_SERVER_RESPONSES.has(errorEvent.message)) { + const err = new RetryableWebSocketConnectError(message, errorEvent.error) + errorPromise.reject(err) + } } else { const err = new ConnectionFailedError(message) _log.error('Connection Error:', err) @@ -163,6 +289,21 @@ export class WebSockets implements Transport { errorPromise.reject(err) } }) + rawSocket.socket.addEventListener('close', (closeEvent: CloseEvent) => { + _log( + 'p2p-websocket-dial %s', + JSON.stringify({ + event: 'raw-socket:close', + peerId, + remoteAddr: ma.toString(), + code: closeEvent.code, + reason: closeEvent.reason, + wasClean: closeEvent.wasClean, + ageMs: Date.now() - rawSocketStartedAtMs, + signalAborted: options.signal?.aborted ?? false, + }) + ) + }) try { options.onProgress?.(new CustomProgressEvent('websockets:open-connection')) diff --git a/packages/backend/src/nest/websocketOverTor/listener.ts b/packages/backend/src/nest/websocketOverTor/listener.ts index 92f6ff0085..5b5801b9a0 100644 --- a/packages/backend/src/nest/websocketOverTor/listener.ts +++ b/packages/backend/src/nest/websocketOverTor/listener.ts @@ -166,11 +166,7 @@ export class WebSocketListener extends TypedEventEmitter impleme .catch(async err => { _log.error('inbound connection failed to upgrade - %e', err) this.metrics.errors?.increment({ [`${this.addr} inbound_upgrade`]: true }) - - await maConn.close().catch(err => { - _log.error('inbound connection failed to close after upgrade failed', err) - this.metrics.errors?.increment({ [`${this.addr} inbound_closing_failed`]: true }) - }) + maConn.abort(err instanceof Error ? err : new Error(String(err))) }) // store the socket so we can close it when the listener closes diff --git a/packages/backend/src/nest/websocketOverTor/socket-to-conn.ts b/packages/backend/src/nest/websocketOverTor/socket-to-conn.ts index 0dc5467a2e..424ef5f136 100644 --- a/packages/backend/src/nest/websocketOverTor/socket-to-conn.ts +++ b/packages/backend/src/nest/websocketOverTor/socket-to-conn.ts @@ -28,6 +28,10 @@ export function socketToMaConn( const log = options.logger.forComponent(`libp2p:websockets:maconn:${remoteAddr.getPeerId()}`) const metrics = options.metrics const metricPrefix = options.metricPrefix ?? '' + const peerId = remoteAddr.getPeerId() ?? 'unknown' + const openedAtMs = Date.now() + let localCloseTrigger: 'maConn.close' | 'maConn.abort' | undefined + let localCloseReason: string | undefined stream.source = abortableAsyncIterable(stream.source, options.signal) const generateSink = ( @@ -50,6 +54,17 @@ export function socketToMaConn( try { await stream.sink(generateSink(source)) } catch (err: any) { + log.error( + 'p2p-websocket-raw %s', + JSON.stringify( + rawSocketLogContext({ + event: 'stream:sink:error', + errorName: err?.name, + errorMessage: err?.message, + errorType: err?.type, + }) + ) + ) if (err.type !== 'aborted') { log.error(`Stream abort error`, err) } else { @@ -66,6 +81,8 @@ export function socketToMaConn( async close(options: AbortOptions = {}) { const start = Date.now() + localCloseTrigger = 'maConn.close' + localCloseReason = options.signal?.aborted === true ? 'signal-already-aborted' : 'close-called' if (options.signal == null) { const signal = AbortSignal.timeout(CLOSE_TIMEOUT) @@ -78,12 +95,22 @@ export function socketToMaConn( const listener = (): void => { const { host, port } = maConn.remoteAddr.toOptions() + localCloseReason = 'close-timeout' log('timeout closing stream to %s:%s after %dms, destroying it manually', host, port, Date.now() - start) this.abort(new TimeoutError('Socket close timeout')) } - options.signal?.addEventListener('abort', listener) + options.signal?.addEventListener('abort', listener, { once: true }) + log( + 'p2p-websocket-raw %s', + JSON.stringify( + rawSocketLogContext({ + event: 'maConn:close:start', + signalAborted: options.signal?.aborted ?? false, + }) + ) + ) try { await stream.close() @@ -93,11 +120,32 @@ export function socketToMaConn( } finally { options.signal?.removeEventListener('abort', listener) maConn.timeline.close = Date.now() + log( + 'p2p-websocket-raw %s', + JSON.stringify( + rawSocketLogContext({ + event: 'maConn:close:end', + closeDurationMs: Date.now() - start, + }) + ) + ) } }, abort(err: Error): void { const { host, port } = maConn.remoteAddr.toOptions() + localCloseTrigger = 'maConn.abort' + localCloseReason = err.message + log.error( + 'p2p-websocket-raw %s', + JSON.stringify( + rawSocketLogContext({ + event: 'maConn:abort', + errorName: err.name, + errorMessage: err.message, + }) + ) + ) log('timeout closing stream to %s:%s due to error', host, port, err) stream.destroy() @@ -111,13 +159,50 @@ export function socketToMaConn( }, } + function rawSocketLogContext(extra: Record = {}): Record { + return { + peerId, + remoteAddr: remoteAddr.toString(), + closeOrigin: localCloseTrigger == null ? 'remote-or-network' : 'local', + localCloseTrigger: localCloseTrigger ?? null, + localCloseReason: localCloseReason ?? null, + socketReadyState: stream.socket.readyState, + openedAtMs, + ageMs: Date.now() - openedAtMs, + timeline: maConn.timeline, + ...extra, + } + } + stream.socket.addEventListener('error', (errorEvent: ErrorEvent) => { + log.error( + 'p2p-websocket-raw %s', + JSON.stringify( + rawSocketLogContext({ + event: 'socket:error', + errorMessage: errorEvent.message, + errorName: errorEvent.error?.name, + errorCause: errorEvent.error?.message, + }) + ) + ) log.error(`Error on socket: ${errorEvent.message}`, errorEvent.error) }) stream.socket.addEventListener( 'close', (closeEvent: CloseEvent) => { + log( + 'p2p-websocket-raw %s', + JSON.stringify( + rawSocketLogContext({ + event: 'socket:close', + code: closeEvent.code, + reason: closeEvent.reason, + wasClean: closeEvent.wasClean, + }) + ) + ) switch (closeEvent.code) { case SocketCloseCode.ERROR: case SocketCloseCode.INVALID_DATA: diff --git a/packages/backend/src/nodeTest/debugP2PDisconnects.tor.spec.ts b/packages/backend/src/nodeTest/debugP2PDisconnects.tor.spec.ts new file mode 100644 index 0000000000..1caefdbc83 --- /dev/null +++ b/packages/backend/src/nodeTest/debugP2PDisconnects.tor.spec.ts @@ -0,0 +1,258 @@ +import { jest } from '@jest/globals' +import { Test, TestingModule } from '@nestjs/testing' +import getPort from 'get-port' +import { once } from 'events' + +import { ConnectionsManagerModule } from '../nest/connections-manager/connections-manager.module' +import { Libp2pModule } from '../nest/libp2p/libp2p.module' +import { Libp2pService } from '../nest/libp2p/libp2p.service' +import { Libp2pEvents, Libp2pNodeParams } from '../nest/libp2p/libp2p.types' +import { IpfsModule } from '../nest/ipfs/ipfs.module' +import { StorageModule } from '../nest/storage/storage.module' +import { SigChainService } from '../nest/auth/sigchain.service' +import { TestModule } from '../nest/common/test.module' +import { createPeerId, generateLibp2pPSK } from '../nest/common/utils' +import { SOCKS_PROXY_AGENT } from '../nest/const' +import { TorModule } from '../nest/tor/tor.module' +import { Tor } from '../nest/tor/tor.service' + +const holdMs = Number(process.env.P2P_DEBUG_HOLD_MS ?? '150000') +const sampleMs = Number(process.env.P2P_DEBUG_SAMPLE_MS ?? '5000') +const listenOnLocalhost = process.env.P2P_DEBUG_LISTEN_ON_LOCALHOST === 'true' +const abortOnPingFailure = process.env.LIBP2P_ABORT_CONNECTION_ON_PING_FAILURE ?? 'true' +const dialRetries = Number(process.env.P2P_DEBUG_DIAL_RETRIES ?? '4') +const dialRetryBaseDelayMs = Number(process.env.P2P_DEBUG_DIAL_RETRY_BASE_DELAY_MS ?? '5000') + +jest.setTimeout(420_000) + +interface PeerHarness { + index: number + module: TestingModule + sigchain: SigChainService + libp2p: Libp2pService + targetPort: number + onionAddress: string +} + +interface DisconnectRecord { + peer: string + event: string + atMs: number + payload: unknown +} + +const withTimeout = async (promise: Promise, timeoutMs: number, label: string): Promise => { + let timeout: NodeJS.Timeout | undefined + const timeoutPromise = new Promise((_, reject) => { + timeout = setTimeout(() => reject(new Error(`${label} timed out after ${timeoutMs}ms`)), timeoutMs) + }) + + try { + return await Promise.race([promise, timeoutPromise]) + } finally { + if (timeout) clearTimeout(timeout) + } +} + +const waitUntil = async (predicate: () => boolean, timeoutMs: number, label: string) => { + const startedAt = Date.now() + while (Date.now() - startedAt < timeoutMs) { + if (predicate()) return + await new Promise(resolve => setTimeout(resolve, 250)) + } + throw new Error(`${label} timed out after ${timeoutMs}ms`) +} + +const isRetryableDialError = (error: any): boolean => { + const message = String(error?.message ?? '').toLowerCase() + return ( + error?.retryable === true || + error?.code === 'ERR_RETRYABLE_WEBSOCKET_UPGRADE' || + error?.code === 'ERR_UNEXPECTED_EOF' || + error?.code === 'ABORT_ERR' || + ['RetryableWebSocketUpgradeError', 'UnexpectedEOFError', 'AbortError', 'TimeoutError'].includes(error?.name) || + message.includes('unexpected end of input') || + message.includes('read aborted') + ) +} + +const dialWithRetries = async (dialer: Libp2pService, peerAddress: string) => { + for (let attempt = 0; attempt <= dialRetries; attempt += 1) { + try { + await dialer.dialPeer(peerAddress, { throwOnError: true, redialOnError: false }) + return + } catch (error: any) { + if (attempt >= dialRetries || !isRetryableDialError(error)) { + throw error + } + const delayMs = dialRetryBaseDelayMs * (attempt + 1) + console.log( + `P2P_DEBUG_DIAL_RETRY ${JSON.stringify({ + attempt: attempt + 1, + nextDelayMs: delayMs, + errorName: error?.name, + errorCode: error?.code, + errorMessage: error?.message, + })}` + ) + await new Promise(resolve => setTimeout(resolve, delayMs)) + } + } +} + +const connectionSnapshot = (peer: PeerHarness) => { + const connections = peer.libp2p.libp2pInstance?.getConnections() ?? [] + return { + peer: peer.index, + connectedPeers: peer.libp2p.connectedPeers.size, + libp2pConnections: connections.length, + connectionStatuses: connections.map(conn => ({ + id: conn.id, + direction: conn.direction, + status: conn.status, + remotePeer: conn.remotePeer.toString(), + remoteAddr: conn.remoteAddr.toString(), + })), + } +} + +const makePeerModule = async (index: number): Promise => { + const module = await Test.createTestingModule({ + imports: [TestModule, Libp2pModule, StorageModule, ConnectionsManagerModule, IpfsModule], + }).compile() + + return { + index, + module, + sigchain: await module.resolve(SigChainService), + libp2p: await module.resolve(Libp2pService), + targetPort: await getPort(), + onionAddress: '', + } +} + +const closePeer = async (peer: PeerHarness) => { + await peer.libp2p.close().catch(err => { + console.error(`peer ${peer.index} libp2p close failed`, err) + }) + await peer.module.close().catch(err => { + console.error(`peer ${peer.index} module close failed`, err) + }) +} + +describe('debug p2p disconnects over tor', () => { + it('holds a Tor WebSocket libp2p auth connection through heartbeat checks', async () => { + process.env.NODE_ENV = process.env.NODE_ENV ?? 'test' + process.env.NETWORK_LOGGING = process.env.NETWORK_LOGGING ?? 'true' + process.env.LIBP2P_ABORT_CONNECTION_ON_PING_FAILURE = abortOnPingFailure + + const torModule = await Test.createTestingModule({ + imports: [TestModule, TorModule], + }).compile() + const tor = await torModule.resolve(Tor) + const agent = await torModule.resolve(SOCKS_PROXY_AGENT) + const disconnects: DisconnectRecord[] = [] + const samples: Array> = [] + const peers: PeerHarness[] = [] + + try { + await tor.init() + await waitUntil(() => tor.bootstrapped, 120_000, 'tor bootstrap') + + peers.push(await makePeerModule(0)) + peers.push(await makePeerModule(1)) + + await peers[0].sigchain.createChain('p2p-debug-team', 'peer0', true) + const invite = peers[0].sigchain.getActiveChain().invites.createLongLivedUserInvite() + await peers[1].sigchain.createChainFromInvite( + 'peer1', + 'p2p-debug-team', + invite.seed, + peers[0].sigchain.team.id, + true + ) + + const psk = generateLibp2pPSK().fullKey + for (const peer of peers) { + const hiddenService = await tor.createNewHiddenService({ targetPort: peer.targetPort }) + peer.onionAddress = hiddenService.onionAddress + const peerId = await createPeerId() + const listenAddress = listenOnLocalhost + ? `/ip4/127.0.0.1/tcp/${peer.targetPort}/ws` + : peer.libp2p.createLibp2pListenAddress(peer.onionAddress) + const params: Libp2pNodeParams = { + peerId, + listenAddresses: [listenAddress], + agent, + localAddress: peer.libp2p.createLibp2pAddress(peer.onionAddress, peerId.peerId.toString()), + targetPort: peer.targetPort, + psk, + torBootstrap: tor, + instanceName: `p2p-debug-${peer.index}`, + } + + peer.libp2p.on(Libp2pEvents.PEER_DISCONNECTED, payload => { + disconnects.push({ + peer: `peer${peer.index}`, + event: Libp2pEvents.PEER_DISCONNECTED, + atMs: Date.now(), + payload, + }) + }) + await peer.libp2p.createInstance(params) + } + + const joinedPromise = once(peers[1].libp2p, Libp2pEvents.AUTH_JOINED) + await dialWithRetries(peers[1].libp2p, peers[0].libp2p.localAddress) + await withTimeout(joinedPromise, 120_000, 'AUTH_JOINED') + await waitUntil( + () => + peers.every(peer => { + const openConnections = peer.libp2p.libp2pInstance?.getConnections().filter(conn => conn.status === 'open') + return (openConnections?.length ?? 0) > 0 + }), + 60_000, + 'open connections on both peers' + ) + + const holdStartedAt = Date.now() + while (Date.now() - holdStartedAt < holdMs) { + samples.push({ + atMs: Date.now(), + ageMs: Date.now() - holdStartedAt, + peers: peers.map(connectionSnapshot), + disconnects: disconnects.length, + }) + if (disconnects.length > 0) break + await new Promise(resolve => setTimeout(resolve, sampleMs)) + } + + const result = { + ok: disconnects.length === 0, + abortOnPingFailure, + dialRetries, + holdMs, + sampleMs, + listenOnLocalhost, + disconnects, + finalSnapshot: peers.map(connectionSnapshot), + sampleCount: samples.length, + firstSample: samples[0], + lastSample: samples[samples.length - 1], + } + + console.log(`P2P_DEBUG_RESULT ${JSON.stringify(result)}`) + expect(result.ok).toBe(true) + } finally { + for (const peer of peers.reverse()) { + await closePeer(peer) + } + await tor.kill().catch(err => { + console.error('tor kill failed', err) + }) + await torModule.close().catch(err => { + console.error('tor module close failed', err) + }) + } + }) +}) diff --git a/packages/desktop/.env.development b/packages/desktop/.env.development index 61cd349504..35c59b092e 100644 --- a/packages/desktop/.env.development +++ b/packages/desktop/.env.development @@ -5,5 +5,12 @@ QPS_ALLOWED=true QSS_ENDPOINT=ws://127.0.0.1:3003 LOG_TO_FILE=true ENVFILE=.env.development -DEBUG=state-manager*,desktop*,utils*,identity*,backend*,-backend:Libp2p*,-backend:auth:cryptoService,-localfirst*,-backend:Tor*,-backend:TimedQueue,-backend:libp2p:auth* -NETWORK_LOGGING=false +DEBUG=state-manager*,desktop*,utils*,identity*,backend*,localfirst*,libp2p:*,-backend:auth:cryptoService,-backend:TimedQueue +NETWORK_LOGGING=true +LIBP2P_ABORT_CONNECTION_ON_PING_FAILURE=false +LIBP2P_CONNECTION_MONITOR_PING_TIMEOUT_MIN_MS=20000 +LIBP2P_PING_TIMEOUT_MS=20000 +LIBP2P_CONNECTION_HEALTH_CHECK_ENABLED=true +LIBP2P_CONNECTION_HEALTH_CHECK_INTERVAL_MS=75000 +LIBP2P_CONNECTION_HEALTH_CHECK_TIMEOUT_MS=20000 +LIBP2P_CONNECTION_HEALTH_CHECK_FAILURE_THRESHOLD=3 From d7c70e94c116b8f6dab4a9a2a0c755c593715cac Mon Sep 17 00:00:00 2001 From: taea Date: Thu, 2 Jul 2026 01:12:57 -0400 Subject: [PATCH 2/4] disable libp2p connection monitor --- .../src/nest/libp2p/libp2p.service.spec.ts | 41 +++++++++++++++++++ .../backend/src/nest/libp2p/libp2p.service.ts | 41 ++++++++++++++++++- .../src/nest/websocketOverTor/index.ts | 26 +++++++++--- .../nest/websocketOverTor/socket-to-conn.ts | 28 ++++++++----- 4 files changed, 118 insertions(+), 18 deletions(-) diff --git a/packages/backend/src/nest/libp2p/libp2p.service.spec.ts b/packages/backend/src/nest/libp2p/libp2p.service.spec.ts index cdb54ab051..42cbd1b8b9 100644 --- a/packages/backend/src/nest/libp2p/libp2p.service.spec.ts +++ b/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>().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 + }) }) diff --git a/packages/backend/src/nest/libp2p/libp2p.service.ts b/packages/backend/src/nest/libp2p/libp2p.service.ts index 29c3154f61..d6cf8c64aa 100644 --- a/packages/backend/src/nest/libp2p/libp2p.service.ts +++ b/packages/backend/src/nest/libp2p/libp2p.service.ts @@ -51,6 +51,7 @@ import type { const CONNECTION_LIMIT = 20 const HEARTBEAT_ABORT_ENV = 'LIBP2P_ABORT_CONNECTION_ON_PING_FAILURE' +const CONNECTION_MONITOR_ENABLED_ENV = 'LIBP2P_CONNECTION_MONITOR_ENABLED' const CONNECTION_MONITOR_PING_TIMEOUT_MIN_MS_ENV = 'LIBP2P_CONNECTION_MONITOR_PING_TIMEOUT_MIN_MS' const PING_SERVICE_TIMEOUT_MS_ENV = 'LIBP2P_PING_TIMEOUT_MS' const CONNECTION_HEALTH_CHECK_ENABLED_ENV = 'LIBP2P_CONNECTION_HEALTH_CHECK_ENABLED' @@ -58,7 +59,7 @@ const CONNECTION_HEALTH_CHECK_INTERVAL_MS_ENV = 'LIBP2P_CONNECTION_HEALTH_CHECK_ const CONNECTION_HEALTH_CHECK_TIMEOUT_MS_ENV = 'LIBP2P_CONNECTION_HEALTH_CHECK_TIMEOUT_MS' const CONNECTION_HEALTH_CHECK_FAILURE_THRESHOLD_ENV = 'LIBP2P_CONNECTION_HEALTH_CHECK_FAILURE_THRESHOLD' const CONNECTION_HEALTH_CHECK_DEFAULT_INTERVAL_MS = 75_000 -const CONNECTION_HEALTH_CHECK_DEFAULT_TIMEOUT_MS = 20_000 +const CONNECTION_HEALTH_CHECK_DEFAULT_TIMEOUT_MS = 30_000 const CONNECTION_HEALTH_CHECK_DEFAULT_FAILURE_THRESHOLD = 3 const REDIAL_QUEUE_CONCURRENCY = 4 const REDIAL_QUEUE_BACKOFF_FACTOR = 1.6 @@ -91,6 +92,14 @@ const numberEnv = (value: string | undefined): number | undefined => { return Number.isFinite(parsed) && parsed > 0 ? parsed : undefined } +const isPingStreamContentionError = (error: any): boolean => { + const message = String(error?.message ?? '') + return ( + error?.name === 'TooManyOutboundProtocolStreamsError' && + message.includes('Too many outbound protocol streams for protocol "/ipfs/ping/1.0.0"') + ) +} + const connectionHealthConfigFromEnv = (): ConnectionHealthConfig => ({ enabled: booleanEnv(process.env[CONNECTION_HEALTH_CHECK_ENABLED_ENV], true), intervalMs: @@ -343,6 +352,28 @@ export class Libp2pService extends EventEmitter implements OnModuleDestroy { durationMs: Date.now() - startedAtMs, }) } catch (error: any) { + if (isPingStreamContentionError(error)) { + this.connectionHealthDebug.set(peerId, { + peerId, + status: previous?.status ?? 'healthy', + failureCount: previous?.failureCount ?? 0, + lastCheckedAtMs: Date.now(), + lastSuccessAtMs: previous?.lastSuccessAtMs, + lastFailureAtMs: previous?.lastFailureAtMs, + lastRttMs: previous?.lastRttMs, + reconnecting: previous?.reconnecting ?? false, + }) + this.logConnectionHealth('health:ping:contention', peerId, { + peerAddress, + durationMs: Date.now() - startedAtMs, + errorName: error?.name, + errorCode: error?.code, + errorMessage: error?.message, + failureCount: previous?.failureCount ?? 0, + }) + return + } + const failureCount = (previous?.failureCount ?? 0) + 1 const status = failureCount >= this.connectionHealthConfig.failureThreshold ? 'reconnecting' : 'degraded' this.connectionHealthDebug.set(peerId, { @@ -764,10 +795,14 @@ export class Libp2pService extends EventEmitter implements OnModuleDestroy { const connectionMonitorPingTimeoutMinMs = numberEnv(process.env[CONNECTION_MONITOR_PING_TIMEOUT_MIN_MS_ENV]) const pingServiceTimeoutMs = numberEnv(process.env[PING_SERVICE_TIMEOUT_MS_ENV]) this.connectionHealthConfig = connectionHealthConfigFromEnv() + const connectionMonitorEnabled = booleanEnv( + process.env[CONNECTION_MONITOR_ENABLED_ENV], + !this.connectionHealthConfig.enabled + ) const connectionMonitorConfig = { abortConnectionOnPingFailure: booleanEnv(process.env[HEARTBEAT_ABORT_ENV], false), pingInterval: 60_000, - enabled: true, + enabled: connectionMonitorEnabled, ...(connectionMonitorPingTimeoutMinMs == null ? {} : { @@ -781,6 +816,8 @@ export class Libp2pService extends EventEmitter implements OnModuleDestroy { 'p2p-heartbeat-monitor-config', JSON.stringify({ ...connectionMonitorConfig, + enabledEnv: CONNECTION_MONITOR_ENABLED_ENV, + enabledEnvValue: process.env[CONNECTION_MONITOR_ENABLED_ENV] ?? null, env: HEARTBEAT_ABORT_ENV, envValue: process.env[HEARTBEAT_ABORT_ENV] ?? null, pingTimeoutMinEnv: CONNECTION_MONITOR_PING_TIMEOUT_MIN_MS_ENV, diff --git a/packages/backend/src/nest/websocketOverTor/index.ts b/packages/backend/src/nest/websocketOverTor/index.ts index 25aa2220bf..22c28ddb4c 100644 --- a/packages/backend/src/nest/websocketOverTor/index.ts +++ b/packages/backend/src/nest/websocketOverTor/index.ts @@ -88,8 +88,16 @@ export interface WebSocketsMetrics { export type WebSocketsDialEvents = OutboundConnectionUpgradeEvents | ProgressEvent<'websockets:open-connection'> -const EXPECTED_SERVER_RESPONSES = new Set(['Unexpected server response: 404', 'Unexpected server response: 503']) -const CONNECT_PHASE_RETRYABLE_SERVER_RESPONSES = new Set(['Unexpected server response: 503']) +const EXPECTED_SERVER_RESPONSE_CODES = new Set([404, 503]) +const CONNECT_PHASE_RETRYABLE_SERVER_RESPONSE_CODES = new Set([503]) + +const websocketServerResponseCode = (message: string): number | undefined => { + const match = /Unexpected server response:\s*(\d+)/.exec(message) + if (match == null) return undefined + + const statusCode = Number(match[1]) + return Number.isFinite(statusCode) ? statusCode : undefined +} export class RetryableWebSocketConnectError extends ConnectionFailedError { public readonly code = 'ERR_RETRYABLE_WEBSOCKET_CONNECT' @@ -274,13 +282,21 @@ export class WebSockets implements Transport { // information about what happened // https://developer.mozilla.org/en-US/docs/Web/API/WebSocket/error_event this.metrics?.dialerEvents.increment({ error: true }) - const message = `Could not connect to ${ma.toString()}: ${errorEvent.message}` + const errorMessage = String(errorEvent.message ?? errorEvent.error?.message ?? 'unknown WebSocket error') + const serverResponseCode = websocketServerResponseCode(errorMessage) + const message = `Could not connect to ${ma.toString()}: ${errorMessage}` + const expectedServerResponse = + serverResponseCode != null && EXPECTED_SERVER_RESPONSE_CODES.has(serverResponseCode) + const retryableServerResponse = + serverResponseCode != null && CONNECT_PHASE_RETRYABLE_SERVER_RESPONSE_CODES.has(serverResponseCode) // Expected HTTP responses are common while peers are offline, starting, or not yet listening. - if (EXPECTED_SERVER_RESPONSES.has(errorEvent.message)) { + if (expectedServerResponse) { _log.warn(message) - if (CONNECT_PHASE_RETRYABLE_SERVER_RESPONSES.has(errorEvent.message)) { + if (retryableServerResponse) { const err = new RetryableWebSocketConnectError(message, errorEvent.error) errorPromise.reject(err) + } else { + errorPromise.reject(new ConnectionFailedError(message)) } } else { const err = new ConnectionFailedError(message) diff --git a/packages/backend/src/nest/websocketOverTor/socket-to-conn.ts b/packages/backend/src/nest/websocketOverTor/socket-to-conn.ts index 424ef5f136..003ef5d68a 100644 --- a/packages/backend/src/nest/websocketOverTor/socket-to-conn.ts +++ b/packages/backend/src/nest/websocketOverTor/socket-to-conn.ts @@ -32,6 +32,7 @@ export function socketToMaConn( const openedAtMs = Date.now() let localCloseTrigger: 'maConn.close' | 'maConn.abort' | undefined let localCloseReason: string | undefined + let socketCloseObserved = stream.socket.readyState === WebSocket.CLOSED stream.source = abortableAsyncIterable(stream.source, options.signal) const generateSink = ( @@ -81,8 +82,11 @@ export function socketToMaConn( async close(options: AbortOptions = {}) { const start = Date.now() - localCloseTrigger = 'maConn.close' - localCloseReason = options.signal?.aborted === true ? 'signal-already-aborted' : 'close-called' + const closeAlreadyObserved = socketCloseObserved || stream.socket.readyState === WebSocket.CLOSED + if (!closeAlreadyObserved) { + localCloseTrigger = 'maConn.close' + localCloseReason = options.signal?.aborted === true ? 'signal-already-aborted' : 'close-called' + } if (options.signal == null) { const signal = AbortSignal.timeout(CLOSE_TIMEOUT) @@ -108,6 +112,7 @@ export function socketToMaConn( rawSocketLogContext({ event: 'maConn:close:start', signalAborted: options.signal?.aborted ?? false, + closeAlreadyObserved, }) ) ) @@ -119,7 +124,9 @@ export function socketToMaConn( this.abort(err) } finally { options.signal?.removeEventListener('abort', listener) - maConn.timeline.close = Date.now() + if (maConn.timeline.close == null) { + maConn.timeline.close = Date.now() + } log( 'p2p-websocket-raw %s', JSON.stringify( @@ -149,7 +156,9 @@ export function socketToMaConn( log('timeout closing stream to %s:%s due to error', host, port, err) stream.destroy() - maConn.timeline.close = Date.now() + if (maConn.timeline.close == null) { + maConn.timeline.close = Date.now() + } // ws WebSocket.terminate does not accept an Error arg to emit an 'error' // event on destroy like other node streams so we can't update a metric @@ -192,6 +201,10 @@ export function socketToMaConn( stream.socket.addEventListener( 'close', (closeEvent: CloseEvent) => { + socketCloseObserved = true + if (maConn.timeline.close == null) { + maConn.timeline.close = Date.now() + } log( 'p2p-websocket-raw %s', JSON.stringify( @@ -216,13 +229,6 @@ export function socketToMaConn( } metrics?.increment({ [`${metricPrefix}close`]: true }) - - // In instances where `close` was not explicitly called, - // such as an iterable stream ending, ensure we have set the close - // timeline - if (maConn.timeline.close == null) { - maConn.timeline.close = Date.now() - } }, { once: true } ) From 69a39ea386e5443f565bfcc0318f48b9b5181727 Mon Sep 17 00:00:00 2001 From: taea Date: Thu, 2 Jul 2026 13:29:21 -0400 Subject: [PATCH 3/4] Fix libp2p tuning CI failures --- packages/backend/package.json | 2 +- packages/backend/src/nest/libp2p/libp2p.service.ts | 11 ++++++++++- packages/desktop/.env.development | 6 +++--- 3 files changed, 14 insertions(+), 5 deletions(-) diff --git a/packages/backend/package.json b/packages/backend/package.json index d95da929ca..7101035ff3 100644 --- a/packages/backend/package.json +++ b/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", diff --git a/packages/backend/src/nest/libp2p/libp2p.service.ts b/packages/backend/src/nest/libp2p/libp2p.service.ts index d6cf8c64aa..4d5da404b8 100644 --- a/packages/backend/src/nest/libp2p/libp2p.service.ts +++ b/packages/backend/src/nest/libp2p/libp2p.service.ts @@ -101,7 +101,7 @@ const isPingStreamContentionError = (error: any): boolean => { } const connectionHealthConfigFromEnv = (): ConnectionHealthConfig => ({ - enabled: booleanEnv(process.env[CONNECTION_HEALTH_CHECK_ENABLED_ENV], true), + enabled: booleanEnv(process.env[CONNECTION_HEALTH_CHECK_ENABLED_ENV], false), intervalMs: numberEnv(process.env[CONNECTION_HEALTH_CHECK_INTERVAL_MS_ENV]) ?? CONNECTION_HEALTH_CHECK_DEFAULT_INTERVAL_MS, timeoutMs: @@ -296,6 +296,13 @@ export class Libp2pService extends EventEmitter implements OnModuleDestroy { this.connectionHealthChecksInFlight.clear() } + private clearConnectionDebugState() { + this.connectionLifecycleDebug.clear() + this.connectionHealthDebug.clear() + this.connectionHealthChecksInFlight.clear() + this.pendingCloseTriggersByPeer.clear() + } + private async checkConnectionHealth() { if (this.state !== Libp2pState.Started || this.libp2pInstance == null) { return @@ -415,6 +422,7 @@ export class Libp2pService extends EventEmitter implements OnModuleDestroy { this.logger.log('Module is being destroyed') this.redialQueue.stop(true) this.stopConnectionHealthChecks() + this.clearConnectionDebugState() if (this._dialQueueInterval) { clearInterval(this._dialQueueInterval) this._dialQueueInterval = null @@ -1193,6 +1201,7 @@ export class Libp2pService extends EventEmitter implements OnModuleDestroy { this.libp2pInstance = null this.connectedPeers = new Map() this.dialedPeers = new Set() + this.clearConnectionDebugState() this.setState(Libp2pState.Stopped) } diff --git a/packages/desktop/.env.development b/packages/desktop/.env.development index 35c59b092e..d45cfaee11 100644 --- a/packages/desktop/.env.development +++ b/packages/desktop/.env.development @@ -5,12 +5,12 @@ QPS_ALLOWED=true QSS_ENDPOINT=ws://127.0.0.1:3003 LOG_TO_FILE=true ENVFILE=.env.development -DEBUG=state-manager*,desktop*,utils*,identity*,backend*,localfirst*,libp2p:*,-backend:auth:cryptoService,-backend:TimedQueue -NETWORK_LOGGING=true +DEBUG=state-manager*,desktop*,utils*,identity*,backend*,-backend:Libp2p*,-backend:auth:cryptoService,-localfirst*,-backend:Tor*,-backend:TimedQueue,-backend:libp2p:auth* +NETWORK_LOGGING=false LIBP2P_ABORT_CONNECTION_ON_PING_FAILURE=false LIBP2P_CONNECTION_MONITOR_PING_TIMEOUT_MIN_MS=20000 LIBP2P_PING_TIMEOUT_MS=20000 -LIBP2P_CONNECTION_HEALTH_CHECK_ENABLED=true +LIBP2P_CONNECTION_HEALTH_CHECK_ENABLED=false LIBP2P_CONNECTION_HEALTH_CHECK_INTERVAL_MS=75000 LIBP2P_CONNECTION_HEALTH_CHECK_TIMEOUT_MS=20000 LIBP2P_CONNECTION_HEALTH_CHECK_FAILURE_THRESHOLD=3 From 5e8eb559b2dc88a6750c3b54e77f248c77bf7a31 Mon Sep 17 00:00:00 2001 From: taea Date: Thu, 2 Jul 2026 13:48:37 -0400 Subject: [PATCH 4/4] Stabilize Channel scroll Cypress assertions --- .../renderer/components/Channel/Channel.cy.tsx | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 deletions(-) diff --git a/packages/desktop/src/renderer/components/Channel/Channel.cy.tsx b/packages/desktop/src/renderer/components/Channel/Channel.cy.tsx index a8874ed447..88ef5f4170 100644 --- a/packages/desktop/src/renderer/components/Channel/Channel.cy.tsx +++ b/packages/desktop/src/renderer/components/Channel/Channel.cy.tsx @@ -17,9 +17,13 @@ declare global { // Custom command to check if the channel content is scrolled to the bottom Cypress.Commands.add('assertScrolledToBottom', { prevSubject: 'element' }, subject => { - const el = subject[0] - const isScrolledToBottom = Math.abs(el.scrollHeight - el.scrollTop - el.clientHeight) <= 1 // Allow 1px difference for rounding - cy.wrap(isScrolledToBottom).should('be.true') + cy.wrap(subject).should($subject => { + const el = $subject[0] + const isScrolledToBottom = Math.abs(el.scrollHeight - el.scrollTop - el.clientHeight) <= 1 // Allow 1px difference for rounding + if (!isScrolledToBottom) { + throw new Error(`Expected message list to be scrolled to bottom, got ${el.scrollTop}`) + } + }) }) const resizeObserverLoopErrRe = /^[^(ResizeObserver loop limit exceeded)]/ @@ -82,10 +86,12 @@ describe('Scroll behavior test', () => { it('PageUp keydown should scroll message list up.', () => { cy.get(messageInput).focus().type('{pageup}{pageup}{pageup}{pageup}{pageup}{pageup}{pageup}{pageup}{pageup}') - cy.get(channelContent).then($el => { + cy.get(channelContent).should($el => { const container = $el[0] const isScrolledToTop = Math.abs(container.scrollTop) <= 1 // Allow 1px difference for rounding - cy.wrap(isScrolledToTop).should('be.true') + if (!isScrolledToTop) { + throw new Error(`Expected message list to be scrolled to top, got ${container.scrollTop}`) + } }) })