diff --git a/.changeset/quiet-planets-fix.md b/.changeset/quiet-planets-fix.md new file mode 100644 index 0000000000..c843c0245d --- /dev/null +++ b/.changeset/quiet-planets-fix.md @@ -0,0 +1,5 @@ +--- +"livekit-client": patch +--- + +fix: recover broken publish paths — act on local `ConnectionQuality.Lost`, add outbound-RTP liveness to the connection reconcile, recreate the peer connection when an ICE restart has no remote description, and reconnect (instead of disconnecting) on a detected connection state mismatch diff --git a/src/room/PCTransport.ts b/src/room/PCTransport.ts index c4367c7bc8..a7f8e0d069 100644 --- a/src/room/PCTransport.ts +++ b/src/room/PCTransport.ts @@ -425,9 +425,15 @@ export default class PCTransport extends (EventEmitter as new () => TypedEmitter // the only exception to this is when ICE restart is needed const currentSD = this._pc.remoteDescription; if (options?.iceRestart && currentSD) { - // TODO: handle when ICE restart is needed but we don't have a remote description - // the best thing to do is to recreate the peerconnection + // roll the remote description back in so createOffer produces a valid + // ICE-restart offer on top of the already-negotiated state await this._pc.setRemoteDescription(currentSD); + } else if (options?.iceRestart) { + // ICE restart with no remote description to restart on: `renegotiate` would stall + // (the pending offer is never answered), so throw for the caller to recreate the PC. + throw new NegotiationError( + 'ICE restart requested without a remote description, peer connection must be recreated', + ); } else { this.renegotiate = true; this.log.debug('requesting renegotiation'); diff --git a/src/room/RTCEngine.ts b/src/room/RTCEngine.ts index ab0a7bb4ae..99b9fe3086 100644 --- a/src/room/RTCEngine.ts +++ b/src/room/RTCEngine.ts @@ -18,6 +18,7 @@ import { LeaveRequest_Action, MediaSectionsRequirement, ParticipantInfo, + ConnectionQuality as ProtoConnectionQuality, PublishDataTrackResponse, ReconnectReason, type ReconnectResponse, @@ -108,6 +109,12 @@ import { const minReconnectWait = 2 * 1000; const leaveReconnect = 'leave-reconnect'; + +/** + * How long local connection quality must stay `LOST` while connected and publishing before we + * force a full reconnect — `LOST` is the server's verdict that it isn't receiving our media. + */ +const connectionQualityLostTimeout = 5 * 1000; const reliabeReceiveStateTTL = 30_000; const initialMediaSectionsAudio = 3; @@ -246,6 +253,15 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit /** used to indicate whether the browser is currently waiting to reconnect */ private isWaitingForNetworkReconnect: boolean = false; + /** set while the local participant's connection quality is `LOST`; forces a full reconnect on timeout */ + private lostQualityTimeout?: ReturnType; + + /** timestamp (ms) the primary transport entered `CONNECTING`, used to bound how long we tolerate it */ + private transportConnectingSince?: number; + + /** last observed publisher outbound `bytesSent`, used to detect a stalled publish path in {@link verifyTransport} */ + private lastPublisherBytesSent?: number; + constructor(private options: InternalRoomOptions) { super(); this.log = getLogger(options.loggerName ?? LoggerNames.Engine, () => this.logContext); @@ -271,8 +287,10 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit this.client.onParticipantUpdate = (updates) => this.emit(EngineEvent.ParticipantUpdate, updates); - this.client.onConnectionQuality = (update) => + this.client.onConnectionQuality = (update) => { + this.handleLocalConnectionQuality(update); this.emit(EngineEvent.ConnectionQualityUpdate, update); + }; this.client.onRoomUpdate = (update) => this.emit(EngineEvent.RoomUpdate, update); this.client.onSubscriptionError = (resp) => this.emit(EngineEvent.SubscriptionError, resp); this.client.onSubscriptionPermissionUpdate = (update) => @@ -429,6 +447,7 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit this.removeAllListeners(); this.deregisterOnLineListener(); this.clearPendingReconnect(); + this.clearLostQualityTimeout(); this.cleanupLossyDataStats(); await this.cleanupPeerConnections(); await this.cleanupClient(); @@ -1157,6 +1176,73 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit ); }; + /** + * A sustained local `LOST` while connected and publishing means the server isn't receiving + * our media, so force a full reconnect; any non-`LOST` value cancels a pending trigger. + */ + private handleLocalConnectionQuality(update: ConnectionQualityUpdate) { + if (!this.participantSid) { + return; + } + const localUpdate = update.updates.find((u) => u.participantSid === this.participantSid); + if (!localUpdate) { + return; + } + if (localUpdate.quality === ProtoConnectionQuality.LOST) { + this.scheduleLostQualityReconnect(); + } else { + this.clearLostQualityTimeout(); + } + } + + private scheduleLostQualityReconnect() { + if (this.lostQualityTimeout) { + // already counting down towards a reconnect + return; + } + this.lostQualityTimeout = CriticalTimers.setTimeout(() => { + this.lostQualityTimeout = undefined; + if (this._isClosed || this.pcState !== PCState.Connected || this.attemptingReconnect) { + return; + } + if (!this.hasActivePublisherSenders()) { + return; + } + this.log.warn( + 'local connection quality lost while publishing, triggering full reconnect', + this.logContext, + ); + this.fullReconnectOnNext = true; + this.handleDisconnect('connection quality lost', ReconnectReason.RR_PUBLISHER_FAILED); + }, connectionQualityLostTimeout); + } + + private clearLostQualityTimeout() { + if (this.lostQualityTimeout) { + clearTimeout(this.lostQualityTimeout); + this.lostQualityTimeout = undefined; + } + } + + /** Whether the publisher currently has any sender with a live track. */ + private hasActivePublisherSenders(): boolean { + return ( + this.pcManager?.publisher + .getSenders() + .some((sender) => !!sender.track && sender.track.readyState === 'live') ?? false + ); + } + + /** + * Forces a full reconnect while keeping the engine (and its saved credentials) alive. Used by + * Room's connection-reconcile safety net when the transport silently died but we looked connected. + * @internal + */ + reconnect(reason: ReconnectReason = ReconnectReason.RR_UNKNOWN) { + this.fullReconnectOnNext = true; + this.handleDisconnect('reconcile', reason); + } + private async attemptReconnect(reason?: ReconnectReason) { if (this._isClosed) { return; @@ -1175,15 +1261,23 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit this.fullReconnectOnNext = true; } + let succeeded = false; + let performedFullReconnect = false; try { this.attemptingReconnect = true; if (this.fullReconnectOnNext) { + performedFullReconnect = true; await this.restartConnection(); } else { await this.resumeConnection(reason); } this.clearPendingReconnect(); - this.fullReconnectOnNext = false; + // Only clear the flag if we actually did a full reconnect, so a full reconnect requested + // mid-attempt (e.g. a server leave during a resume) survives a successful resume. + if (performedFullReconnect) { + this.fullReconnectOnNext = false; + } + succeeded = true; } catch (e) { this.reconnectAttempts += 1; let recoverable = true; @@ -1209,6 +1303,13 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit } } finally { this.attemptingReconnect = false; + + // A full reconnect requested mid-attempt (e.g. a `RECONNECT` leave during a resume) that + // a successful attempt didn't act on; dispatch it now (the failure path already retries). + if (succeeded && this.fullReconnectOnNext && !this._isClosed) { + this.log.debug('full reconnect requested during in-progress attempt, dispatching'); + this.handleDisconnect('reconnect'); + } } } @@ -1587,15 +1688,18 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit } /* @internal */ - verifyTransport(): boolean { + async verifyTransport(): Promise { if (!this.pcManager) { return false; } + const state = this.pcManager.currentState; const allowedConnectionStates: PCTransportState[] = [ PCTransportState.CONNECTING, PCTransportState.CONNECTED, ]; - if (!allowedConnectionStates.includes(this.pcManager.currentState)) { + if (!allowedConnectionStates.includes(state)) { + this.transportConnectingSince = undefined; + this.lastPublisherBytesSent = undefined; return false; } @@ -1603,9 +1707,66 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit if (!this.client.ws || this.client.ws.readyState === WebSocket.CLOSED) { return false; } + + // A transport stuck in CONNECTING never reaches CONNECTED nor reports FAILED, so it would + // otherwise look healthy forever; bound how long we tolerate it. + if (state === PCTransportState.CONNECTING) { + const now = Date.now(); + if (this.transportConnectingSince === undefined) { + this.transportConnectingSince = now; + } else if (now - this.transportConnectingSince > this.peerConnectionTimeout) { + this.log.warn('transport stuck in connecting state', this.logContext); + return false; + } + // can't assert media liveness until connected + this.lastPublisherBytesSent = undefined; + return true; + } + this.transportConnectingSince = undefined; + + // Outbound-RTP liveness: with active senders `bytesSent` must keep advancing between checks; + // if it stalls while connected the publish path is broken even though the PC looks connected. + if (this.hasActivePublisherSenders()) { + const bytesSent = await this.getPublisherBytesSent(); + if (bytesSent !== undefined) { + const advanced = + this.lastPublisherBytesSent === undefined || bytesSent > this.lastPublisherBytesSent; + this.lastPublisherBytesSent = bytesSent; + if (!advanced) { + this.log.warn('publisher outbound bytes not advancing while senders active', { + ...this.logContext, + bytesSent, + }); + return false; + } + } + } else { + this.lastPublisherBytesSent = undefined; + } + return true; } + /** Sum of `bytesSent` across the publisher's outbound-rtp stats, or undefined if unavailable. */ + private async getPublisherBytesSent(): Promise { + try { + const stats = await this.pcManager?.publisher.getStats(); + if (!stats) { + return undefined; + } + let bytesSent = 0; + stats.forEach((report) => { + if (report.type === 'outbound-rtp') { + bytesSent += report.bytesSent ?? 0; + } + }); + return bytesSent; + } catch (e) { + this.log.debug('could not read publisher stats', { ...this.logContext, error: e }); + return undefined; + } + } + /** @internal */ async negotiate(): Promise { // observe signal state diff --git a/src/room/Room.ts b/src/room/Room.ts index 47b3087777..a190b5d6bb 100644 --- a/src/room/Room.ts +++ b/src/room/Room.ts @@ -2584,14 +2584,17 @@ class Room extends (EventEmitter as new () => TypedEmitter) private registerConnectionReconcile() { this.clearConnectionReconcile(); let consecutiveFailures = 0; - this.connectionReconcileInterval = CriticalTimers.setInterval(() => { + this.connectionReconcileInterval = CriticalTimers.setInterval(async () => { + // `verifyTransport` is async (it samples outbound-rtp stats), so resolve it once + // and reuse the result for both the decision and the diagnostic log. + const transportHealthy = this.engine ? await this.engine.verifyTransport() : false; if ( // ensure we didn't tear it down !this.engine || // engine detected close, but Room missed it this.engine.isClosed || // transports failed without notifying engine - !this.engine.verifyTransport() + !transportHealthy ) { consecutiveFailures++; this.log.warn('detected connection state mismatch', { @@ -2599,16 +2602,25 @@ class Room extends (EventEmitter as new () => TypedEmitter) engine: this.engine ? { closed: this.engine.isClosed, - transportsConnectedOrConnecting: this.engine.verifyTransport(), + transportsConnectedOrConnecting: transportHealthy, } : undefined, }); if (consecutiveFailures >= 3) { - this.recreateEngine(); - this.handleDisconnect( - this.options.stopLocalTrackOnUnpublish, - DisconnectReason.STATE_MISMATCH, - ); + this.clearConnectionReconcile(); + if (this.engine && !this.engine.isClosed) { + // The transport silently died while we still looked connected. Try a full reconnect + // (keeps the room alive; the engine falls back to Disconnected if it ultimately fails). + this.log.warn('detected connection state mismatch, attempting full reconnect'); + this.engine.reconnect(); + } else { + // No usable engine to reconnect with; tear down. + this.recreateEngine(); + this.handleDisconnect( + this.options.stopLocalTrackOnUnpublish, + DisconnectReason.STATE_MISMATCH, + ); + } } } else { consecutiveFailures = 0;