fix(sfu): reconcile late host publications (#213)

Restores SFU subscription reconciliation across real LiveKit event ordering and tightens production logging.

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

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:
 

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<string>();
+  private readonly knownHostPublications = new Map<
+    string,
+    KnownHostPublication
+  >();
   private state: SubscriberState = "idle";
   private statsAccumulator: StatsAccumulator = createStatsAccumulator();
   private statsTimer: ReturnType<typeof setInterval> | 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;
   }

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;