diff --git a/.changeset/participant-active-event.md b/.changeset/participant-active-event.md new file mode 100644 index 00000000..1c075c5a --- /dev/null +++ b/.changeset/participant-active-event.md @@ -0,0 +1,14 @@ +--- +'@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. + +`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/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_state.test.ts b/packages/livekit-rtc/src/participant_state.test.ts new file mode 100644 index 00000000..53e88c97 --- /dev/null +++ b/packages/livekit-rtc/src/participant_state.test.ts @@ -0,0 +1,351 @@ +// 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); + }); +}); + +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 4ed47938..4753b826 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, @@ -412,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(); } @@ -500,6 +518,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; @@ -602,11 +633,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 +1178,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 +1255,7 @@ export type RoomCallbacks = { export enum RoomEvent { ParticipantConnected = 'participantConnected', + ParticipantActive = 'participantActive', ParticipantDisconnected = 'participantDisconnected', LocalTrackPublished = 'localTrackPublished', LocalTrackUnpublished = 'localTrackUnpublished',