From 7b8b0c5dee102dfeb88c5656a8c549433835bce5 Mon Sep 17 00:00:00 2001 From: tinalenguyen Date: Tue, 1 Sep 2026 09:49:32 -0400 Subject: [PATCH 1/3] feat(rtc): surface ParticipantActive and Participant.state The FFI already emits a `ParticipantActive` event, but `Room` dropped it on the floor: JS had no way to observe when a remote participant transitions past JOINED, and `Participant` exposed no `state` at all (`info.state` was set once at construction and never refreshed). A remote participant can only receive data messages once it reaches `ParticipantState.ACTIVE`, so callers waiting on `ParticipantConnected` could start sending to a participant that was not yet reachable. The Python SDK has had `participant_active` for this; this brings Node to parity. - handle the `participantActive` FFI event, flip `info.state` to ACTIVE, and emit `RoomEvent.ParticipantActive` - flip `info.state` to DISCONNECTED alongside the existing `disconnectReason` - add the `Participant.state` getter and export `ParticipantState` --- .changeset/participant-active-event.md | 10 ++ packages/livekit-rtc/src/index.ts | 1 + packages/livekit-rtc/src/participant.ts | 13 ++ .../src/participant_active.test.ts | 134 ++++++++++++++++++ packages/livekit-rtc/src/room.ts | 21 ++- 5 files changed, 178 insertions(+), 1 deletion(-) create mode 100644 .changeset/participant-active-event.md create mode 100644 packages/livekit-rtc/src/participant_active.test.ts diff --git a/.changeset/participant-active-event.md b/.changeset/participant-active-event.md new file mode 100644 index 00000000..62dd93df --- /dev/null +++ b/.changeset/participant-active-event.md @@ -0,0 +1,10 @@ +--- +'@livekit/rtc-node': minor +--- + +Surface the `ParticipantActive` FFI event as `RoomEvent.ParticipantActive` and expose `Participant.state`. + +A remote participant can only receive data messages once it reaches `ParticipantState.ACTIVE`; +until now JS had no way to observe that transition, so code waiting on `ParticipantConnected` +could send to a participant that was not yet reachable. This brings the Node SDK in line with the +Python SDK's `participant_active` event. diff --git a/packages/livekit-rtc/src/index.ts b/packages/livekit-rtc/src/index.ts index 50728c46..1d62f589 100644 --- a/packages/livekit-rtc/src/index.ts +++ b/packages/livekit-rtc/src/index.ts @@ -19,6 +19,7 @@ export { DisconnectReason, ParticipantKind, ParticipantKindDetail, + ParticipantState, } from '@livekit/rtc-ffi-bindings'; export { ConnectionQuality, diff --git a/packages/livekit-rtc/src/participant.ts b/packages/livekit-rtc/src/participant.ts index af60164a..0b6602af 100644 --- a/packages/livekit-rtc/src/participant.ts +++ b/packages/livekit-rtc/src/participant.ts @@ -8,6 +8,7 @@ import { type ParticipantInfo, ParticipantKind, type ParticipantKindDetail, + ParticipantState, } from '@livekit/rtc-ffi-bindings'; import { type ByteStreamOpenCallback, @@ -141,6 +142,18 @@ export abstract class Participant { return this.info.kindDetails ?? []; } + /** + * The participant's lifecycle state. + * + * A remote participant is only able to receive data messages once it reaches + * {@link ParticipantState.ACTIVE}. Between {@link RoomEvent.ParticipantConnected} and + * {@link RoomEvent.ParticipantActive} the participant is visible in `remoteParticipants` + * but not yet reachable. + */ + get state(): ParticipantState { + return this.info.state ?? ParticipantState.JOINING; + } + get disconnectReason(): DisconnectReason | undefined { if (this.info.disconnectReason === DisconnectReason.UNKNOWN_REASON) { return undefined; diff --git a/packages/livekit-rtc/src/participant_active.test.ts b/packages/livekit-rtc/src/participant_active.test.ts new file mode 100644 index 00000000..3666e8ec --- /dev/null +++ b/packages/livekit-rtc/src/participant_active.test.ts @@ -0,0 +1,134 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { DisconnectReason, ParticipantState } from '@livekit/rtc-ffi-bindings'; +import { describe, expect, it, vi } from 'vitest'; +import type { RemoteParticipant } from './participant.js'; +import { Room, RoomEvent } from './room.js'; + +// Same rationale as audio_stream_room_lifecycle.test.ts: these tests fabricate +// participants with synthetic FFI handle ids, so replace FfiHandle with an inert +// stub to keep the native drop-on-GC path from firing on unallocated handles. +vi.mock('@livekit/rtc-ffi-bindings', async () => { + const actual = await vi.importActual( + '@livekit/rtc-ffi-bindings', + ); + class FakeFfiHandle { + private _handle: bigint; + constructor(handle: bigint) { + this._handle = handle; + } + dispose(): void {} + get handle(): bigint { + return this._handle; + } + } + return { ...actual, FfiHandle: FakeFfiHandle }; +}); + +/** A Room wired up just enough for onFfiEvent to accept roomEvent messages. */ +function makeConnectedRoom(): Room { + const room = new Room(); + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const r = room as any; + r.info = { name: 'test-room' }; + r.ffiHandle = { handle: BigInt(1), dispose: () => {} }; + r.localParticipant = {}; + return room; +} + +/** Push a roomEvent through the private FFI handler the way FfiClient would. */ +async function emitRoomEvent(room: Room, message: unknown): Promise { + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const r = room as any; + await r.onFfiEvent({ + message: { + case: 'roomEvent', + value: { roomHandle: r.ffiHandle.handle, message }, + }, + }); +} + +async function connectParticipant(room: Room, identity: string): Promise { + await emitRoomEvent(room, { + case: 'participantConnected', + value: { + info: { + info: { identity, state: ParticipantState.JOINED }, + handle: { id: BigInt(0) }, + }, + }, + }); +} + +describe('participant active', () => { + it('reports the state carried by participantConnected', async () => { + const room = makeConnectedRoom(); + await connectParticipant(room, 'alice'); + + expect(room.remoteParticipants.get('alice')!.state).toBe(ParticipantState.JOINED); + }); + + it('emits ParticipantActive and flips state on the FFI event', async () => { + const room = makeConnectedRoom(); + const active: RemoteParticipant[] = []; + room.on(RoomEvent.ParticipantActive, (p) => active.push(p)); + + await connectParticipant(room, 'alice'); + expect(active).toHaveLength(0); + + await emitRoomEvent(room, { + case: 'participantActive', + value: { participantIdentity: 'alice' }, + }); + + expect(active).toHaveLength(1); + expect(active[0]!.identity).toBe('alice'); + expect(room.remoteParticipants.get('alice')!.state).toBe(ParticipantState.ACTIVE); + }); + + it('ignores ParticipantActive for an unknown participant', async () => { + const room = makeConnectedRoom(); + const active: RemoteParticipant[] = []; + room.on(RoomEvent.ParticipantActive, (p) => active.push(p)); + + await emitRoomEvent(room, { + case: 'participantActive', + value: { participantIdentity: 'nobody' }, + }); + + expect(active).toHaveLength(0); + }); + + it('marks a departing participant as disconnected', async () => { + const room = makeConnectedRoom(); + await connectParticipant(room, 'alice'); + + const departed: RemoteParticipant[] = []; + room.on(RoomEvent.ParticipantDisconnected, (p) => departed.push(p)); + + await emitRoomEvent(room, { + case: 'participantDisconnected', + value: { + participantIdentity: 'alice', + disconnectReason: DisconnectReason.CLIENT_INITIATED, + }, + }); + + expect(departed).toHaveLength(1); + expect(departed[0]!.state).toBe(ParticipantState.DISCONNECTED); + expect(room.remoteParticipants.has('alice')).toBe(false); + }); + + it('defaults to JOINING when the FFI omits a state', async () => { + const room = makeConnectedRoom(); + await emitRoomEvent(room, { + case: 'participantConnected', + value: { + info: { info: { identity: 'bob' }, handle: { id: BigInt(0) } }, + }, + }); + + expect(room.remoteParticipants.get('bob')!.state).toBe(ParticipantState.JOINING); + }); +}); diff --git a/packages/livekit-rtc/src/room.ts b/packages/livekit-rtc/src/room.ts index 4ed47938..33a3ea5a 100644 --- a/packages/livekit-rtc/src/room.ts +++ b/packages/livekit-rtc/src/room.ts @@ -8,7 +8,11 @@ import type { GetSessionStatsCallback, GetSessionStatsResponse, } from '@livekit/rtc-ffi-bindings'; -import { DisconnectReason, type OwnedParticipant } from '@livekit/rtc-ffi-bindings'; +import { + DisconnectReason, + type OwnedParticipant, + ParticipantState, +} from '@livekit/rtc-ffi-bindings'; import { type DisconnectCallback, type TrackPublicationInfo } from '@livekit/rtc-ffi-bindings'; import { ByteStreamReaderReadIncrementalRequest, @@ -602,11 +606,20 @@ export class Room extends (EventEmitter as new () => TypedEmitter const participant = this.createRemoteParticipant(ev.value.info!); this.remoteParticipants.set(participant.identity!, participant); this.emit(RoomEvent.ParticipantConnected, participant); + } else if (ev.case == 'participantActive') { + const participant = this.remoteParticipants.get(ev.value.participantIdentity!); + if (participant) { + participant.info.state = ParticipantState.ACTIVE; + this.emit(RoomEvent.ParticipantActive, participant); + } else { + log.warn(`RoomEvent.ParticipantActive: Could not find participant`); + } } else if (ev.case == 'participantDisconnected') { const participant = this.remoteParticipants.get(ev.value.participantIdentity!); if (participant) { this.remoteParticipants.delete(participant.identity); participant.info.disconnectReason = ev.value.disconnectReason; + participant.info.state = ParticipantState.DISCONNECTED; this.emit(RoomEvent.ParticipantDisconnected, participant); } else { log.warn(`RoomEvent.ParticipantDisconnected: Could not find participant`); @@ -1138,6 +1151,11 @@ export class ConnectError extends Error { export type RoomCallbacks = { participantConnected: (participant: RemoteParticipant) => void; + /** + * Fired when a remote participant becomes active and is able to receive data + * messages. Always follows `participantConnected` for the same participant. + */ + participantActive: (participant: RemoteParticipant) => void; participantDisconnected: (participant: RemoteParticipant) => void; localTrackPublished: (publication: LocalTrackPublication, participant: LocalParticipant) => void; localTrackUnpublished: ( @@ -1210,6 +1228,7 @@ export type RoomCallbacks = { export enum RoomEvent { ParticipantConnected = 'participantConnected', + ParticipantActive = 'participantActive', ParticipantDisconnected = 'participantDisconnected', LocalTrackPublished = 'localTrackPublished', LocalTrackUnpublished = 'localTrackUnpublished', From 52bc5d9ee90b036a80d9196bc8b78b7458d614eb Mon Sep 17 00:00:00 2001 From: tinalenguyen Date: Tue, 1 Sep 2026 12:50:16 -0400 Subject: [PATCH 2/3] fix(rtc): mark participants disconnected when the room disconnects MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `Participant.state` reads `ParticipantInfo.state`, which only ever moved on per-participant events. A room-level disconnect — explicit `disconnect()` or an FFI-driven one — is not reported as each participant departing, so the local participant and every still-listed remote participant kept reporting ACTIVE. That state outlives the disconnect in practice: the participant maps are never cleared, and `Disconnected` handlers routinely capture participants. A retained participant claiming to be ACTIVE reads as reachable when it no longer is. Transition the local participant and all retained remote participants to DISCONNECTED in cleanupOnDisconnect, before `ConnectionStateChanged` and `Disconnected` fire, so a handler that inspects `state` sees the truth. Two remote-participant stubs in audio_stream_room_lifecycle.test.ts gain the `info` that real participants always carry (createRemoteParticipant sets it unconditionally), since cleanup now writes through it. Renames participant_active.test.ts to participant_state.test.ts — it now covers the full state lifecycle rather than just the active transition. --- .changeset/participant-active-event.md | 4 + .../src/audio_stream_room_lifecycle.test.ts | 12 +- .../src/participant_active.test.ts | 134 --------- .../livekit-rtc/src/participant_state.test.ts | 256 ++++++++++++++++++ packages/livekit-rtc/src/room.ts | 13 + 5 files changed, 283 insertions(+), 136 deletions(-) delete mode 100644 packages/livekit-rtc/src/participant_active.test.ts create mode 100644 packages/livekit-rtc/src/participant_state.test.ts diff --git a/.changeset/participant-active-event.md b/.changeset/participant-active-event.md index 62dd93df..1c075c5a 100644 --- a/.changeset/participant-active-event.md +++ b/.changeset/participant-active-event.md @@ -8,3 +8,7 @@ A remote participant can only receive data messages once it reaches `Participant until now JS had no way to observe that transition, so code waiting on `ParticipantConnected` could send to a participant that was not yet reachable. This brings the Node SDK in line with the Python SDK's `participant_active` event. + +`Participant.state` also now reports `DISCONNECTED` once the participant is gone — both when it +departs individually and when the room itself disconnects, since a room-level disconnect is not +reported as each participant departing. diff --git a/packages/livekit-rtc/src/audio_stream_room_lifecycle.test.ts b/packages/livekit-rtc/src/audio_stream_room_lifecycle.test.ts index fb680874..e5d99f74 100644 --- a/packages/livekit-rtc/src/audio_stream_room_lifecycle.test.ts +++ b/packages/livekit-rtc/src/audio_stream_room_lifecycle.test.ts @@ -2,7 +2,7 @@ // // SPDX-License-Identifier: Apache-2.0 import type { OwnedTrack } from '@livekit/rtc-ffi-bindings'; -import { TrackPublishOptions } from '@livekit/rtc-ffi-bindings'; +import { ParticipantState, TrackPublishOptions } from '@livekit/rtc-ffi-bindings'; import { describe, expect, it, vi } from 'vitest'; import type { AudioFrame } from './audio_frame.js'; import type { AudioStreamSource } from './audio_stream.js'; @@ -87,6 +87,9 @@ function makeRoom(opts: { name: string; token?: string; serverUrl?: string }): R interface StubParticipant { identity: string; + // Real participants always carry an `info` (createRemoteParticipant sets it + // unconditionally); cleanupOnDisconnect writes the disconnected state into it. + info: { identity: string; state: ParticipantState }; trackPublications: Map; } @@ -99,7 +102,11 @@ function attachRemoteParticipant( for (const pub of publications) { map.set(pub.trackSid, { sid: pub.publicationSid }); } - const participant: StubParticipant = { identity, trackPublications: map }; + const participant: StubParticipant = { + identity, + info: { identity, state: ParticipantState.ACTIVE }, + trackPublications: map, + }; // eslint-disable-next-line @typescript-eslint/no-explicit-any room.remoteParticipants.set(identity, participant as any); } @@ -738,6 +745,7 @@ describe('AudioStream room lifecycle', () => { const remoteTrack = makeTrack('TR_REMOTE'); const remoteParticipant = { identity: 'bob', + info: { identity: 'bob', state: ParticipantState.ACTIVE }, trackPublications: new Map([['TR_REMOTE', { sid: 'PUB_REMOTE', track: remoteTrack }]]), }; // eslint-disable-next-line @typescript-eslint/no-explicit-any diff --git a/packages/livekit-rtc/src/participant_active.test.ts b/packages/livekit-rtc/src/participant_active.test.ts deleted file mode 100644 index 3666e8ec..00000000 --- a/packages/livekit-rtc/src/participant_active.test.ts +++ /dev/null @@ -1,134 +0,0 @@ -// SPDX-FileCopyrightText: 2026 LiveKit, Inc. -// -// SPDX-License-Identifier: Apache-2.0 -import { DisconnectReason, ParticipantState } from '@livekit/rtc-ffi-bindings'; -import { describe, expect, it, vi } from 'vitest'; -import type { RemoteParticipant } from './participant.js'; -import { Room, RoomEvent } from './room.js'; - -// Same rationale as audio_stream_room_lifecycle.test.ts: these tests fabricate -// participants with synthetic FFI handle ids, so replace FfiHandle with an inert -// stub to keep the native drop-on-GC path from firing on unallocated handles. -vi.mock('@livekit/rtc-ffi-bindings', async () => { - const actual = await vi.importActual( - '@livekit/rtc-ffi-bindings', - ); - class FakeFfiHandle { - private _handle: bigint; - constructor(handle: bigint) { - this._handle = handle; - } - dispose(): void {} - get handle(): bigint { - return this._handle; - } - } - return { ...actual, FfiHandle: FakeFfiHandle }; -}); - -/** A Room wired up just enough for onFfiEvent to accept roomEvent messages. */ -function makeConnectedRoom(): Room { - const room = new Room(); - // eslint-disable-next-line @typescript-eslint/no-explicit-any - const r = room as any; - r.info = { name: 'test-room' }; - r.ffiHandle = { handle: BigInt(1), dispose: () => {} }; - r.localParticipant = {}; - return room; -} - -/** Push a roomEvent through the private FFI handler the way FfiClient would. */ -async function emitRoomEvent(room: Room, message: unknown): Promise { - // eslint-disable-next-line @typescript-eslint/no-explicit-any - const r = room as any; - await r.onFfiEvent({ - message: { - case: 'roomEvent', - value: { roomHandle: r.ffiHandle.handle, message }, - }, - }); -} - -async function connectParticipant(room: Room, identity: string): Promise { - await emitRoomEvent(room, { - case: 'participantConnected', - value: { - info: { - info: { identity, state: ParticipantState.JOINED }, - handle: { id: BigInt(0) }, - }, - }, - }); -} - -describe('participant active', () => { - it('reports the state carried by participantConnected', async () => { - const room = makeConnectedRoom(); - await connectParticipant(room, 'alice'); - - expect(room.remoteParticipants.get('alice')!.state).toBe(ParticipantState.JOINED); - }); - - it('emits ParticipantActive and flips state on the FFI event', async () => { - const room = makeConnectedRoom(); - const active: RemoteParticipant[] = []; - room.on(RoomEvent.ParticipantActive, (p) => active.push(p)); - - await connectParticipant(room, 'alice'); - expect(active).toHaveLength(0); - - await emitRoomEvent(room, { - case: 'participantActive', - value: { participantIdentity: 'alice' }, - }); - - expect(active).toHaveLength(1); - expect(active[0]!.identity).toBe('alice'); - expect(room.remoteParticipants.get('alice')!.state).toBe(ParticipantState.ACTIVE); - }); - - it('ignores ParticipantActive for an unknown participant', async () => { - const room = makeConnectedRoom(); - const active: RemoteParticipant[] = []; - room.on(RoomEvent.ParticipantActive, (p) => active.push(p)); - - await emitRoomEvent(room, { - case: 'participantActive', - value: { participantIdentity: 'nobody' }, - }); - - expect(active).toHaveLength(0); - }); - - it('marks a departing participant as disconnected', async () => { - const room = makeConnectedRoom(); - await connectParticipant(room, 'alice'); - - const departed: RemoteParticipant[] = []; - room.on(RoomEvent.ParticipantDisconnected, (p) => departed.push(p)); - - await emitRoomEvent(room, { - case: 'participantDisconnected', - value: { - participantIdentity: 'alice', - disconnectReason: DisconnectReason.CLIENT_INITIATED, - }, - }); - - expect(departed).toHaveLength(1); - expect(departed[0]!.state).toBe(ParticipantState.DISCONNECTED); - expect(room.remoteParticipants.has('alice')).toBe(false); - }); - - it('defaults to JOINING when the FFI omits a state', async () => { - const room = makeConnectedRoom(); - await emitRoomEvent(room, { - case: 'participantConnected', - value: { - info: { info: { identity: 'bob' }, handle: { id: BigInt(0) } }, - }, - }); - - expect(room.remoteParticipants.get('bob')!.state).toBe(ParticipantState.JOINING); - }); -}); diff --git a/packages/livekit-rtc/src/participant_state.test.ts b/packages/livekit-rtc/src/participant_state.test.ts new file mode 100644 index 00000000..a07a9c5e --- /dev/null +++ b/packages/livekit-rtc/src/participant_state.test.ts @@ -0,0 +1,256 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { Mutex } from '@livekit/mutex'; +import type { OwnedParticipant } from '@livekit/rtc-ffi-bindings'; +import { ConnectionState, DisconnectReason, ParticipantState } from '@livekit/rtc-ffi-bindings'; +import { describe, expect, it, vi } from 'vitest'; +import { FfiClient } from './ffi_client.js'; +import type { RemoteParticipant } from './participant.js'; +import { LocalParticipant } from './participant.js'; +import { Room, RoomEvent } from './room.js'; + +// Same rationale as audio_stream_room_lifecycle.test.ts: these tests fabricate +// participants with synthetic FFI handle ids, so replace FfiHandle with an inert +// stub to keep the native drop-on-GC path from firing on unallocated handles. +vi.mock('@livekit/rtc-ffi-bindings', async () => { + const actual = await vi.importActual( + '@livekit/rtc-ffi-bindings', + ); + class FakeFfiHandle { + private _handle: bigint; + constructor(handle: bigint) { + this._handle = handle; + } + dispose(): void {} + get handle(): bigint { + return this._handle; + } + } + return { ...actual, FfiHandle: FakeFfiHandle }; +}); + +function makeLocalParticipant(identity: string): LocalParticipant { + const owned = { + info: { identity, state: ParticipantState.ACTIVE }, + handle: { id: BigInt(0) }, + } as unknown as OwnedParticipant; + return new LocalParticipant(owned, new Mutex(), new AbortController().signal); +} + +/** A Room wired up just enough for onFfiEvent to accept roomEvent messages. */ +function makeConnectedRoom(): Room { + const room = new Room(); + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const r = room as any; + r.info = { name: 'test-room' }; + r.ffiHandle = { handle: BigInt(1), dispose: () => {} }; + r.localParticipant = makeLocalParticipant('local'); + r._connectionState = ConnectionState.CONN_CONNECTED; + return room; +} + +/** Push a roomEvent through the private FFI handler the way FfiClient would. */ +async function emitRoomEvent(room: Room, message: unknown): Promise { + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const r = room as any; + await r.onFfiEvent({ + message: { + case: 'roomEvent', + value: { roomHandle: r.ffiHandle.handle, message }, + }, + }); +} + +async function connectParticipant(room: Room, identity: string): Promise { + await emitRoomEvent(room, { + case: 'participantConnected', + value: { + info: { + info: { identity, state: ParticipantState.JOINED }, + handle: { id: BigInt(0) }, + }, + }, + }); +} + +describe('participant active', () => { + it('reports the state carried by participantConnected', async () => { + const room = makeConnectedRoom(); + await connectParticipant(room, 'alice'); + + expect(room.remoteParticipants.get('alice')!.state).toBe(ParticipantState.JOINED); + }); + + it('emits ParticipantActive and flips state on the FFI event', async () => { + const room = makeConnectedRoom(); + const active: RemoteParticipant[] = []; + room.on(RoomEvent.ParticipantActive, (p) => active.push(p)); + + await connectParticipant(room, 'alice'); + expect(active).toHaveLength(0); + + await emitRoomEvent(room, { + case: 'participantActive', + value: { participantIdentity: 'alice' }, + }); + + expect(active).toHaveLength(1); + expect(active[0]!.identity).toBe('alice'); + expect(room.remoteParticipants.get('alice')!.state).toBe(ParticipantState.ACTIVE); + }); + + it('ignores ParticipantActive for an unknown participant', async () => { + const room = makeConnectedRoom(); + const active: RemoteParticipant[] = []; + room.on(RoomEvent.ParticipantActive, (p) => active.push(p)); + + await emitRoomEvent(room, { + case: 'participantActive', + value: { participantIdentity: 'nobody' }, + }); + + expect(active).toHaveLength(0); + }); + + it('marks a departing participant as disconnected', async () => { + const room = makeConnectedRoom(); + await connectParticipant(room, 'alice'); + + const departed: RemoteParticipant[] = []; + room.on(RoomEvent.ParticipantDisconnected, (p) => departed.push(p)); + + await emitRoomEvent(room, { + case: 'participantDisconnected', + value: { + participantIdentity: 'alice', + disconnectReason: DisconnectReason.CLIENT_INITIATED, + }, + }); + + expect(departed).toHaveLength(1); + expect(departed[0]!.state).toBe(ParticipantState.DISCONNECTED); + expect(room.remoteParticipants.has('alice')).toBe(false); + }); + + it('defaults to JOINING when the FFI omits a state', async () => { + const room = makeConnectedRoom(); + await emitRoomEvent(room, { + case: 'participantConnected', + value: { + info: { info: { identity: 'bob' }, handle: { id: BigInt(0) } }, + }, + }); + + expect(room.remoteParticipants.get('bob')!.state).toBe(ParticipantState.JOINING); + }); +}); + +describe('participant state on room disconnect', () => { + /** + * A room-level disconnect is not reported as each participant departing, so + * these assertions are about references a caller still holds afterwards — the + * participant maps are never cleared, and `Disconnected` handlers routinely + * capture participants. + */ + + it('marks retained participants disconnected on an explicit disconnect()', async () => { + const room = makeConnectedRoom(); + await connectParticipant(room, 'alice'); + const alice = room.remoteParticipants.get('alice')!; + const local = room.localParticipant!; + await emitRoomEvent(room, { + case: 'participantActive', + value: { participantIdentity: 'alice' }, + }); + expect(alice.state).toBe(ParticipantState.ACTIVE); + expect(local.state).toBe(ParticipantState.ACTIVE); + + // Mock the FFI round-trip so disconnect() resolves without a real server. + const requestSpy = vi + .spyOn(FfiClient.instance, 'request') + .mockReturnValue({ asyncId: BigInt(1) } as never); + const waitForSpy = vi + .spyOn(FfiClient.instance, 'waitFor') + .mockResolvedValue({ error: undefined } as never); + + try { + await room.disconnect(); + } finally { + requestSpy.mockRestore(); + waitForSpy.mockRestore(); + } + + expect(alice.state).toBe(ParticipantState.DISCONNECTED); + expect(local.state).toBe(ParticipantState.DISCONNECTED); + }); + + it('marks retained participants disconnected on an FFI-driven disconnect', async () => { + const room = makeConnectedRoom(); + await connectParticipant(room, 'alice'); + await connectParticipant(room, 'bob'); + const alice = room.remoteParticipants.get('alice')!; + const bob = room.remoteParticipants.get('bob')!; + const local = room.localParticipant!; + await emitRoomEvent(room, { + case: 'participantActive', + value: { participantIdentity: 'alice' }, + }); + + await emitRoomEvent(room, { + case: 'disconnected', + value: { reason: DisconnectReason.SERVER_SHUTDOWN }, + }); + + // alice was ACTIVE and bob only JOINED; a room-level disconnect ends both. + expect(alice.state).toBe(ParticipantState.DISCONNECTED); + expect(bob.state).toBe(ParticipantState.DISCONNECTED); + expect(local.state).toBe(ParticipantState.DISCONNECTED); + }); + + it('has already transitioned participants by the time Disconnected fires', async () => { + const room = makeConnectedRoom(); + await connectParticipant(room, 'alice'); + await emitRoomEvent(room, { + case: 'participantActive', + value: { participantIdentity: 'alice' }, + }); + + // The states a handler actually observes — the whole point of transitioning + // before the event rather than after it. + const observed: Array = []; + room.on(RoomEvent.ConnectionStateChanged, () => { + observed.push(room.remoteParticipants.get('alice')?.state); + }); + room.on(RoomEvent.Disconnected, () => { + observed.push(room.remoteParticipants.get('alice')?.state); + observed.push(room.localParticipant?.state); + }); + + await emitRoomEvent(room, { + case: 'disconnected', + value: { reason: DisconnectReason.SERVER_SHUTDOWN }, + }); + + expect(observed).toEqual([ + ParticipantState.DISCONNECTED, + ParticipantState.DISCONNECTED, + ParticipantState.DISCONNECTED, + ]); + }); + + it('leaves participants untouched while merely reconnecting', async () => { + const room = makeConnectedRoom(); + await connectParticipant(room, 'alice'); + const alice = room.remoteParticipants.get('alice')!; + await emitRoomEvent(room, { + case: 'participantActive', + value: { participantIdentity: 'alice' }, + }); + + await emitRoomEvent(room, { case: 'reconnecting', value: {} }); + + expect(alice.state).toBe(ParticipantState.ACTIVE); + expect(room.localParticipant!.state).toBe(ParticipantState.ACTIVE); + }); +}); diff --git a/packages/livekit-rtc/src/room.ts b/packages/livekit-rtc/src/room.ts index 33a3ea5a..c0832b48 100644 --- a/packages/livekit-rtc/src/room.ts +++ b/packages/livekit-rtc/src/room.ts @@ -504,6 +504,19 @@ export class Room extends (EventEmitter as new () => TypedEmitter } } + // A room-level disconnect isn't reported as each participant departing, so + // nothing else moves them off ACTIVE. Callers hold on to participant objects + // past the disconnect (the maps aren't cleared, and `Disconnected` listeners + // routinely capture them), and a retained participant claiming to be ACTIVE + // reads as reachable when it no longer is. Transition them here, before any + // disconnect event fires, so a handler observing `state` sees the truth. + if (this.localParticipant) { + this.localParticipant.info.state = ParticipantState.DISCONNECTED; + } + for (const participant of this.remoteParticipants.values()) { + participant.info.state = ParticipantState.DISCONNECTED; + } + // Clear sidPromise before removing listeners so that a reconnect // doesn't return a stale, permanently-pending promise. this.sidPromise = undefined; From 09105e4605cb14334c30670fb68f5a27d228a862 Mon Sep 17 00:00:00 2001 From: tinalenguyen Date: Tue, 1 Sep 2026 14:44:31 -0400 Subject: [PATCH 3/3] fix(rtc): drain in-flight FFI events before explicit disconnect cleanup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit disconnect() ran cleanupOnDisconnect() outside ffiEventLock and only removed the FfiClient listener afterwards. onFfiEvent is dispatched synchronously but awaits the lock before doing anything, so callbacks delivered before removeListener were still pending when cleanup ran, and went on to process participantActive or participantsUpdated — both of which write participant info — overwriting the DISCONNECTED state cleanup had just set. Remove the listener first so no new callbacks are created, then acquire ffiEventLock before cleanup. The mutex is FIFO, so already-queued callbacks run to completion first and cleanup lands last. This is the JS analogue of the Python SDK's disconnect(), which unsubscribes its queue and awaits the listen task before flipping connection state. cleanupOnDisconnect stays lock-free: the FFI-driven disconnected path reaches it from inside onFfiEvent, which already holds the lock, and Mutex is not reentrant. --- .../livekit-rtc/src/participant_state.test.ts | 95 +++++++++++++++++++ packages/livekit-rtc/src/room.ts | 16 +++- 2 files changed, 110 insertions(+), 1 deletion(-) diff --git a/packages/livekit-rtc/src/participant_state.test.ts b/packages/livekit-rtc/src/participant_state.test.ts index a07a9c5e..53e88c97 100644 --- a/packages/livekit-rtc/src/participant_state.test.ts +++ b/packages/livekit-rtc/src/participant_state.test.ts @@ -254,3 +254,98 @@ describe('participant state on room disconnect', () => { expect(room.localParticipant!.state).toBe(ParticipantState.ACTIVE); }); }); + +describe('participant state vs. queued FFI events', () => { + /** + * onFfiEvent is dispatched synchronously by FfiClient but immediately awaits + * ffiEventLock, so events delivered before disconnect() removes the listener + * are still pending when cleanup runs. These drive that interleaving directly: + * the callback is invoked (entering the lock queue) but not awaited before + * disconnect() is called. + */ + + function mockDisconnectRoundTrip() { + const requestSpy = vi + .spyOn(FfiClient.instance, 'request') + .mockReturnValue({ asyncId: BigInt(1) } as never); + const waitForSpy = vi + .spyOn(FfiClient.instance, 'waitFor') + .mockResolvedValue({ error: undefined } as never); + return () => { + requestSpy.mockRestore(); + waitForSpy.mockRestore(); + }; + } + + it('does not let a queued participantActive resurrect state after disconnect()', async () => { + const room = makeConnectedRoom(); + await connectParticipant(room, 'alice'); + const alice = room.remoteParticipants.get('alice')!; + + // Queued but deliberately not awaited: it is now waiting on ffiEventLock. + const queued = emitRoomEvent(room, { + case: 'participantActive', + value: { participantIdentity: 'alice' }, + }); + + const restore = mockDisconnectRoundTrip(); + try { + await room.disconnect(); + } finally { + restore(); + } + await queued; + + expect(alice.state).toBe(ParticipantState.DISCONNECTED); + expect(room.localParticipant!.state).toBe(ParticipantState.DISCONNECTED); + }); + + it('does not let a queued participantsUpdated resurrect state after disconnect()', async () => { + const room = makeConnectedRoom(); + await connectParticipant(room, 'alice'); + const alice = room.remoteParticipants.get('alice')!; + + // participantsUpdated replaces `info` wholesale, so it overwrites state too. + const queued = emitRoomEvent(room, { + case: 'participantsUpdated', + value: { + participants: [{ identity: 'alice', state: ParticipantState.ACTIVE }], + }, + }); + + const restore = mockDisconnectRoundTrip(); + try { + await room.disconnect(); + } finally { + restore(); + } + await queued; + + expect(alice.state).toBe(ParticipantState.DISCONNECTED); + }); + + it('drains queued events before cleanup rather than dropping them', async () => { + const room = makeConnectedRoom(); + await connectParticipant(room, 'alice'); + + const seen: string[] = []; + room.on(RoomEvent.ParticipantActive, () => seen.push('active')); + room.on(RoomEvent.Disconnected, () => seen.push('disconnected')); + + const queued = emitRoomEvent(room, { + case: 'participantActive', + value: { participantIdentity: 'alice' }, + }); + + const restore = mockDisconnectRoundTrip(); + try { + await room.disconnect(); + } finally { + restore(); + } + await queued; + + // The queued event still ran; it simply ran first. + expect(seen).toEqual(['active', 'disconnected']); + }); +}); diff --git a/packages/livekit-rtc/src/room.ts b/packages/livekit-rtc/src/room.ts index c0832b48..4753b826 100644 --- a/packages/livekit-rtc/src/room.ts +++ b/packages/livekit-rtc/src/room.ts @@ -416,9 +416,23 @@ export class Room extends (EventEmitter as new () => TypedEmitter return ev.message.case == 'disconnect' && ev.message.value.asyncId == res.asyncId; }); - this.cleanupOnDisconnect(DisconnectReason.CLIENT_INITIATED); + // Stop accepting new events, then queue behind the ones already in flight. + // An onFfiEvent callback that fired before removeListener is already waiting + // on ffiEventLock, and processing it after cleanup would resurrect state the + // cleanup just tore down (participantActive and participantsUpdated both + // write participant info). Taking the lock here drains those first — the FIFO + // mirror of the Python SDK awaiting its listen task before flipping state. + // cleanupOnDisconnect must not take the lock itself: the FFI-driven path + // reaches it from inside onFfiEvent, which already holds it. FfiClient.instance.removeListener(FfiClientEvent.FfiEvent, this.onFfiEvent); + const unlock = await this.ffiEventLock.lock(); + try { + this.cleanupOnDisconnect(DisconnectReason.CLIENT_INITIATED); + } finally { + unlock(); + } + this.removeAllListeners(); }