diff --git a/deploy/livekit/livekit.yaml.example b/deploy/livekit/livekit.yaml.example index 3928838..293fc1a 100644 --- a/deploy/livekit/livekit.yaml.example +++ b/deploy/livekit/livekit.yaml.example @@ -23,11 +23,14 @@ room: # Keep this false so a stale participant token cannot recreate a drained room. auto_create: false # Screener enforces its configured admission limit before issuing a token. - # This server-side ceiling covers the repository's maximum 16 viewers + host. - max_participants: 17 + # This server-side ceiling covers the repository's 20 viewers plus one Host. + max_participants: 21 logging: - level: info + # Info-level join records include raw PublisherOffer SDP in LiveKit 1.13.5. + # Keep production at warn so credentials and media-path details stay out of + # the persistent journal; raise it only in an isolated, short-lived canary. + level: warn pion_level: error # External or embedded TURN, Redis, recording, ingress, egress, and webhooks are diff --git a/docs/deployment.md b/docs/deployment.md index 7e22c4e..25593ce 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -514,8 +514,11 @@ webhook, external-TURN, and embedded-TURN service. It explicitly sets deployment's self-hosted STUN listener so pinned LiveKit cannot inherit its default public Google STUN servers. Do not add Redis for this one-node workload. The service journal is the diagnostic log; keep its retention finite and access -restricted. Never enable debug/Pion packet logging continuously or persist JWTs, -SDP, ICE candidates, API secrets, or full `/rtc` and `/rtc/*` request targets. +restricted. Pinned LiveKit 1.13.5 includes raw PublisherOffer SDP in info-level +join records, so the tracked production baseline uses `logging.level: warn` and +`pion_level: error`. Raise either only in an isolated, short-lived canary and +remove its journal afterward. Never persist JWTs, SDP, ICE candidates, API +secrets, or full `/rtc` and `/rtc/*` request targets. Configure both the host firewall and the provider firewall independently: diff --git a/src/client/sfu/subscriber.ts b/src/client/sfu/subscriber.ts index 30bca00..8350d35 100644 --- a/src/client/sfu/subscriber.ts +++ b/src/client/sfu/subscriber.ts @@ -30,6 +30,11 @@ interface SubscribedTrack { onEnded: () => void; } +interface KnownHostPublication { + publication: RemoteTrackPublication; + participant: RemoteParticipant; +} + type SubscriberState = | "idle" | "connecting" @@ -50,6 +55,10 @@ export class SfuSubscriber { private readonly stream = new MediaStream(); private streamEmitted = false; private readonly desiredTrackSids = new Set(); + private readonly knownHostPublications = new Map< + string, + KnownHostPublication + >(); private state: SubscriberState = "idle"; private statsAccumulator: StatsAccumulator = createStatsAccumulator(); private statsTimer: ReturnType | null = null; @@ -110,12 +119,7 @@ export class SfuSubscriber { const generation = this.generation; this.state = "active"; try { - const host = room.remoteParticipants.get(HOST_IDENTITY); - if (host) { - for (const publication of host.trackPublications.values()) { - this.subscribeIfAllowed(publication, host, this.sdk); - } - } + this.reconcileHostSubscriptions(room, this.sdk, generation); return true; } catch (error) { if (this.owns(room, generation)) { @@ -169,11 +173,24 @@ export class SfuSubscriber { room.on( sdk.RoomEvent.TrackPublished, (publication: RemoteTrackPublication, participant: RemoteParticipant) => { - if (this.state !== "active" || !this.owns(room, generation)) { + if (!this.owns(room, generation)) { return; } try { - this.subscribeIfAllowed(publication, participant, sdk); + this.rememberHostPublication(publication, participant, sdk); + } catch { + void this.failClosed(room, generation); + } + }, + ); + room.on( + sdk.RoomEvent.ParticipantConnected, + (participant: RemoteParticipant) => { + if (!this.owns(room, generation)) { + return; + } + try { + this.rememberHostParticipant(participant, sdk); } catch { void this.failClosed(room, generation); } @@ -220,6 +237,7 @@ export class SfuSubscriber { sdk.RoomEvent.TrackUnpublished, (publication: RemoteTrackPublication, participant: RemoteParticipant) => { if (this.owns(room, generation) && participant.identity === HOST_IDENTITY) { + this.knownHostPublications.delete(publication.trackSid); this.desiredTrackSids.delete(publication.trackSid); this.removeTrack(publication.trackSid); } @@ -229,6 +247,11 @@ export class SfuSubscriber { sdk.RoomEvent.ParticipantDisconnected, (participant: RemoteParticipant) => { if (this.owns(room, generation) && participant.identity === HOST_IDENTITY) { + for (const [sid, known] of this.knownHostPublications) { + if (known.participant === participant) { + this.knownHostPublications.delete(sid); + } + } this.desiredTrackSids.clear(); this.clearMedia(true); } @@ -241,6 +264,7 @@ export class SfuSubscriber { }); room.on(sdk.RoomEvent.Reconnected, () => { if (this.owns(room, generation) && this.state === "active") { + this.reconcileHostSubscriptions(room, sdk, generation); this.events.onState?.("connected"); void this.updateStats(); } @@ -267,6 +291,9 @@ export class SfuSubscriber { ) { return; } + if (this.desiredTrackSids.has(publication.trackSid)) { + return; + } this.desiredTrackSids.add(publication.trackSid); publication.setSubscribed(true); if (publication.source === sdk.Track.Source.ScreenShare) { @@ -276,6 +303,55 @@ export class SfuSubscriber { } } + private rememberHostParticipant( + participant: RemoteParticipant, + sdk: LiveKit, + ): void { + if (participant.identity !== HOST_IDENTITY) { + return; + } + for (const publication of participant.trackPublications.values()) { + this.rememberHostPublication(publication, participant, sdk); + } + } + + private rememberHostPublication( + publication: RemoteTrackPublication, + participant: RemoteParticipant, + sdk: LiveKit, + ): void { + if ( + participant.identity !== HOST_IDENTITY || + !isScreenSource(publication.source, sdk.Track) + ) { + return; + } + this.knownHostPublications.set(publication.trackSid, { + publication, + participant, + }); + if (this.state === "active") { + this.subscribeIfAllowed(publication, participant, sdk); + } + } + + private reconcileHostSubscriptions( + room: Room, + sdk: LiveKit, + generation: number, + ): void { + if (!this.owns(room, generation) || this.state !== "active") { + return; + } + const host = room.remoteParticipants.get(HOST_IDENTITY); + if (host) { + this.rememberHostParticipant(host, sdk); + } + for (const { publication, participant } of this.knownHostPublications.values()) { + this.subscribeIfAllowed(publication, participant, sdk); + } + } + private rollbackSubscriptions(room: Room): void { const host = room.remoteParticipants.get(HOST_IDENTITY); for (const publication of host?.trackPublications.values() ?? []) { @@ -474,6 +550,7 @@ export class SfuSubscriber { this.room = null; this.sdk = null; this.desiredTrackSids.clear(); + this.knownHostPublications.clear(); this.clearMedia(notifyStream); return room; } diff --git a/tests/sfu-client.test.ts b/tests/sfu-client.test.ts index 4909607..24bdd9f 100644 --- a/tests/sfu-client.test.ts +++ b/tests/sfu-client.test.ts @@ -250,6 +250,7 @@ const livekit = vi.hoisted(() => { const RoomEvent = { Disconnected: "disconnected", + ParticipantConnected: "participant-connected", ParticipantDisconnected: "participant-disconnected", Reconnected: "reconnected", Reconnecting: "reconnecting", @@ -1229,6 +1230,48 @@ describe("SfuPublisher", () => { }); describe("SfuSubscriber", () => { + it("reconciles a Host publication announced while connect is pending", async () => { + const gate = deferred(); + livekit.state.connectGate = gate.promise; + const subscriber = new SfuSubscriber({ onStream: vi.fn() }); + const connecting = subscriber.connect(connection); + await vi.waitFor(() => expect(livekit.state.rooms).toHaveLength(1)); + const room = livekit.state.rooms[0]; + const host = new livekit.FakeRemoteParticipant("host"); + const hostVideo = new livekit.FakeRemotePublication( + "host-video-pending", + Track.Source.ScreenShare, + ); + host.add(hostVideo); + + room.emit(RoomEvent.TrackPublished, hostVideo, host); + expect(hostVideo.setSubscribed).not.toHaveBeenCalled(); + gate.resolve(); + await expect(connecting).resolves.toBe(true); + + expect(subscriber.activate()).toBe(true); + expect(hostVideo.setSubscribed).toHaveBeenCalledWith(true); + expect(hostVideo.setVideoQuality).toHaveBeenCalledWith(VideoQuality.HIGH); + }); + + it("reconciles a Host participant that appears after activation", async () => { + const subscriber = new SfuSubscriber({ onStream: vi.fn() }); + await subscriber.connect(connection); + const room = livekit.state.rooms[0]; + expect(subscriber.activate()).toBe(true); + + const host = new livekit.FakeRemoteParticipant("host"); + const hostVideo = new livekit.FakeRemotePublication( + "host-video-late", + Track.Source.ScreenShare, + ); + host.add(hostVideo); + room.emit(RoomEvent.ParticipantConnected, host); + + expect(hostVideo.setSubscribed).toHaveBeenCalledWith(true); + expect(hostVideo.setVideoQuality).toHaveBeenCalledWith(VideoQuality.HIGH); + }); + it("keeps the assigned screen subscription at a HIGH ceiling", async () => { const gate = deferred(); livekit.state.connectGate = gate.promise;