fix: isolate viewer peer connection generations (#8)

docs/project-memory.md

@@ -33,7 +33,7 @@ Last updated: 2026-08-18
 ## Current Implementation
 
 - The repository contains a single npm package using Node.js 24, React, TypeScript, Vite, native WebRTC, `ws`, Zod, Vitest, and a separate coturn deployment. The server defaults to an all-interface listener for LAN development and containers; the bare-metal reverse-proxy baseline explicitly binds loopback. It provides a process-only `/healthz` endpoint and requires STUN plus explicit TURN/UDP and TURN/TCP URLs before production startup. TURN/TLS is accepted but optional; configuration checks do not establish public-network reachability.
-- The Web PoC implements capture-before-room creation, live source replacement without renegotiating healthy peers, one independent peer connection per viewer, stable signaling reconnect identities that also work for LAN viewers on HTTP, explicit ICE restart or peer rebuild, host generation isolation, short-lived TURN credentials, three manual quality profiles, and local WebRTC statistics.
+- The Web PoC implements capture-before-room creation, live source replacement without renegotiating healthy peers, one independent peer connection per viewer, stable signaling reconnect identities that also work for LAN viewers on HTTP, explicit ICE restart or peer rebuild, host session generation isolation, viewer connection-generation guards for asynchronous signaling and stats, short-lived TURN credentials, three manual quality profiles, and local WebRTC statistics.
 - Access is deliberately small: `ACCESS_PASSWORD` is optional and site-wide. When configured, host and viewer routes first establish a 12-hour stateless HMAC HttpOnly `SameSite=Strict` cookie; room creation and WebSocket upgrade accept that cookie and no direct room-creation Bearer bypass. There are no accounts, database, JWTs, server-side access-session map, or logout flow.
 - A room uses a random 12-digit numeric code. `/r/{code}` carries no viewer token or fragment, `/join` accepts only the code, and the 256-bit host token stays internal to the host signaling session. In public mode the code is the sole viewing capability and is not a strong privacy guarantee, so private Internet deployments should configure `ACCESS_PASSWORD`.
 - The Web control plane is deployed at `https://share.bonfire.icu` behind nginx with Node.js 24.19.0, and `turn.bonfire.icu` runs authenticated coturn 4.17.2 on standard UDP/TCP 3478. HTTPS, WSS, room authentication, certificate renewal, public STUN, authenticated TURN/UDP and TURN/TCP allocations, and relay-only bidirectional data paths are verified. TURN/TLS is intentionally not enabled.

docs/status.md

@@ -18,9 +18,10 @@ The first measurable WebRTC Web proof of concept is implemented at `https://shar
 
 ## Verified In This Revision
 
-- `npm run check` passes type checking, 87 Vitest tests, the client production build, and the server production build.
+- `npm run check` passes type checking, 90 Vitest tests, the client production build, and the server production build.
 - Headless Chromium 151 created real local peer connections from animated canvas streams. The viewer reached `connected`, received a live video track, and reported a direct UDP path.
 - Browser fault injection recovered after dropping the first offer, dropping the first answer, and failing the first `createOffer`. A viewer-only signaling reconnect preserved the existing healthy media peer; replacing that viewer tab rebuilt media, released the old peer, and did not enter a reconnect loop.
+- Delayed ViewerPeer answer, candidate flush, ICE event, and stats results are discarded after a connection generation is replaced, so an old peer cannot signal through or overwrite the new peer snapshot.
 - Cancelling a delayed room A and immediately starting room B left B live, stopped only A's capture, and invalidated A's invitation after bounded cleanup.
 - Injecting a live picker-cancellation result preserved the old stream. Replacing synthetic video, adding synthetic audio, and removing it all kept the same connected host/viewer peer objects and did not create another offer; retired capture tracks stopped after each successful change.
 - Chrome 151 at 390 px completed the protected-site gate, login, host, and manual room-code join flow without horizontal overflow. The unauthenticated gate shows no application header or sharing controls.

docs/方案设计.md

@@ -115,7 +115,7 @@ Zod 在服务边界拒绝未知类型、超长字符串、非法 SDP/candidate
 6. 双方持续通过信令服务交换 trickle ICE candidates。
 7. remote description 尚未设置时，客户端仅缓存有界数量的 candidates；旧 `connectionId` 消息直接丢弃。
 
-这种单一 offerer 模式满足当前单向媒体需求，不引入 perfect-negotiation 状态机。同一页面中的不健康连接发送 `rebuild: false`，host 先尝试 `restartIce()`，非稳定信令状态或恢复失败时重建该 peer。新标签页没有旧 `RTCPeerConnection`，或页面 peer 与 `authenticated.connectionId` 代次不一致时，viewer 发送 `rebuild: true`，host 直接换代。健康且代次一致的媒体在单纯信令重连时保持不变；所有异步启动结果以 host 会话 generation 隔离，旧房间不得改写新房间状态。
+这种单一 offerer 模式满足当前单向媒体需求，不引入 perfect-negotiation 状态机。同一页面中的不健康连接发送 `rebuild: false`，host 先尝试 `restartIce()`，非稳定信令状态或恢复失败时重建该 peer。新标签页没有旧 `RTCPeerConnection`，或页面 peer 与 `authenticated.connectionId` 代次不一致时，viewer 发送 `rebuild: true`，host 直接换代。健康且代次一致的媒体在单纯信令重连时保持不变；所有异步启动结果以 host 会话 generation 隔离，viewer 的异步 answer、candidate 和 stats 结果也必须同时匹配当前 connection 对象与 connectionId，旧代次不得改写新状态或发出信令。
 
 信令 WebSocket 暂时中断时，不立即关闭仍健康的 P2P 媒体；客户端指数退避恢复控制通道。明确不可重连的会话替换会释放旧页面的 peer、统计定时器和媒体引用，host 也停止旧捕获。首版只保证同一页面生命周期内恢复，不承诺浏览器进程被系统挂起后的连续播放。
 

src/client/webrtc/viewer-peer.ts

@@ -55,6 +55,8 @@ export class ViewerPeer {
       return;
     }
     this.hostPeerId = hostPeerId;
+    let operationConnection: RTCPeerConnection | null = null;
+    let operationConnectionId: string | null = null;
 
     try {
       if (payload.kind === "description") {
@@ -68,16 +70,31 @@ export class ViewerPeer {
         if (!connection) {
           return;
         }
+        const connectionId = payload.connectionId;
+        operationConnection = connection;
+        operationConnectionId = connectionId;
         await connection.setRemoteDescription(payload.description);
-        await this.flushCandidates(payload.connectionId);
+        if (!this.isCurrentConnection(connection, connectionId)) {
+          return;
+        }
+        await this.flushCandidates(connection, connectionId);
+        if (!this.isCurrentConnection(connection, connectionId)) {
+          return;
+        }
         const answer = await connection.createAnswer();
+        if (!this.isCurrentConnection(connection, connectionId)) {
+          return;
+        }
         await connection.setLocalDescription(answer);
+        if (!this.isCurrentConnection(connection, connectionId)) {
+          return;
+        }
         if (!connection.localDescription) {
           throw new Error("Local description was not created");
         }
         this.events.sendSignal({
           kind: "description",
-          connectionId: payload.connectionId,
+          connectionId,
           description: {
             type: "answer",
             sdp: connection.localDescription.sdp,
@@ -89,11 +106,25 @@ export class ViewerPeer {
         this.connectionId === payload.connectionId &&
         this.connection.remoteDescription
       ) {
-        await this.connection.addIceCandidate(payload.candidate);
+        const connection = this.connection;
+        const connectionId = payload.connectionId;
+        operationConnection = connection;
+        operationConnectionId = connectionId;
+        await connection.addIceCandidate(payload.candidate);
+        if (!this.isCurrentConnection(connection, connectionId)) {
+          return;
+        }
       } else {
         this.queueCandidate(payload.connectionId, payload.candidate);
       }
     } catch (error) {
+      if (
+        !operationConnection ||
+        operationConnectionId === null ||
+        !this.isCurrentConnection(operationConnection, operationConnectionId)
+      ) {
+        return;
+      }
       this.setError(error, "处理分享端信令失败");
     }
   }
@@ -203,10 +234,14 @@ export class ViewerPeer {
       this.handleConnectionState(connection.connectionState);
       this.emit();
     });
-    connection.addEventListener("iceconnectionstatechange", () => this.emit());
+    connection.addEventListener("iceconnectionstatechange", () => {
+      if (this.connection === connection) {
+        this.emit();
+      }
+    });
 
     this.statsTimer = window.setInterval(() => {
-      void this.updateStats(connection);
+      void this.updateStats(connection, connectionId);
     }, 2_000);
     this.emit();
   }
@@ -256,27 +291,40 @@ export class ViewerPeer {
     }
   }
 
-  private async flushCandidates(connectionId: string): Promise<void> {
-    if (!this.connection || this.connectionId !== connectionId) {
+  private async flushCandidates(
+    connection: RTCPeerConnection,
+    connectionId: string,
+  ): Promise<void> {
+    if (!this.isCurrentConnection(connection, connectionId)) {
       return;
     }
     const candidates = this.pendingByConnection.get(connectionId) ?? [];
     this.pendingByConnection.delete(connectionId);
     for (const candidate of candidates) {
-      await this.connection.addIceCandidate(candidate);
+      await connection.addIceCandidate(candidate);
+      if (!this.isCurrentConnection(connection, connectionId)) {
+        return;
+      }
     }
   }
 
-  private async updateStats(connection: RTCPeerConnection): Promise<void> {
-    if (this.disposed || this.connection !== connection) {
+  private async updateStats(
+    connection: RTCPeerConnection,
+    connectionId: string,
+  ): Promise<void> {
+    if (!this.isCurrentConnection(connection, connectionId)) {
       return;
     }
+    const statsAccumulator = this.statsAccumulator;
     try {
       const metrics = await collectConnectionMetrics(
         connection,
         "receive",
-        this.statsAccumulator,
+        statsAccumulator,
       );
+      if (!this.isCurrentConnection(connection, connectionId)) {
+        return;
+      }
       if (this.snapshot) {
         this.snapshot = { ...this.snapshot, metrics };
         this.emit();
@@ -286,6 +334,17 @@ export class ViewerPeer {
     }
   }
 
+  private isCurrentConnection(
+    connection: RTCPeerConnection,
+    connectionId: string,
+  ): boolean {
+    return (
+      !this.disposed &&
+      this.connection === connection &&
+      this.connectionId === connectionId
+    );
+  }
+
   private setError(error: unknown, fallback: string): void {
     if (!this.snapshot) {
       return;

tests/viewer-peer-generation.test.ts

@@ -0,0 +1,275 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+
+import type { SignalPayload } from "../src/shared/protocol.ts";
+import type { PeerSnapshot } from "../src/client/types.ts";
+import { ViewerPeer } from "../src/client/webrtc/viewer-peer.ts";
+
+interface Deferred<T> {
+  promise: Promise<T>;
+  resolve: (value: T) => void;
+}
+
+function createDeferred<T>(): Deferred<T> {
+  let resolve!: (value: T) => void;
+  const promise = new Promise<T>((resolvePromise) => {
+    resolve = resolvePromise;
+  });
+  return { promise, resolve };
+}
+
+interface ConnectionPlan {
+  candidateGates?: Promise<void>[];
+  localDescriptionGate?: Promise<void>;
+}
+
+class FakeMediaStream {
+  private readonly tracks: MediaStreamTrack[] = [];
+
+  getTrackById(id: string): MediaStreamTrack | null {
+    return this.tracks.find((track) => track.id === id) ?? null;
+  }
+
+  addTrack(track: MediaStreamTrack): void {
+    this.tracks.push(track);
+  }
+}
+
+class FakePeerConnection extends EventTarget {
+  static readonly instances: FakePeerConnection[] = [];
+  static readonly plans: ConnectionPlan[] = [];
+
+  connectionState: RTCPeerConnectionState = "new";
+  iceConnectionState: RTCIceConnectionState = "new";
+  remoteDescription: RTCSessionDescription | null = null;
+  localDescription: RTCSessionDescription | null = null;
+  statsGate: Promise<RTCStatsReport> | null = null;
+
+  private readonly candidateGates: Promise<void>[];
+  private readonly localDescriptionGate: Promise<void> | null;
+
+  readonly addIceCandidate = vi.fn(async (_candidate: RTCIceCandidateInit | null) => {
+    const gate = this.candidateGates.shift();
+    if (gate) {
+      await gate;
+    }
+  });
+  readonly getStats = vi.fn(async () => {
+    if (this.statsGate) {
+      return this.statsGate;
+    }
+    return new Map() as unknown as RTCStatsReport;
+  });
+  readonly setLocalDescription = vi.fn(
+    async (description: RTCSessionDescriptionInit) => {
+      if (this.localDescriptionGate) {
+        await this.localDescriptionGate;
+      }
+      this.localDescription = description as RTCSessionDescription;
+    },
+  );
+
+  constructor() {
+    super();
+    const plan = FakePeerConnection.plans.shift() ?? {};
+    this.candidateGates = [...(plan.candidateGates ?? [])];
+    this.localDescriptionGate = plan.localDescriptionGate ?? null;
+    FakePeerConnection.instances.push(this);
+  }
+
+  async setRemoteDescription(
+    description: RTCSessionDescriptionInit,
+  ): Promise<void> {
+    this.remoteDescription = description as RTCSessionDescription;
+  }
+
+  async createAnswer(): Promise<RTCSessionDescriptionInit> {
+    return {
+      type: "answer",
+      sdp: `answer-${FakePeerConnection.instances.indexOf(this)}`,
+    };
+  }
+
+  setConfiguration(): void {}
+
+  close(): void {
+    this.connectionState = "closed";
+  }
+}
+
+const intervalCallbacks = new Map<number, () => void>();
+let nextIntervalId = 1;
+
+function offer(connectionId: string): SignalPayload {
+  return {
+    kind: "description",
+    connectionId,
+    description: { type: "offer", sdp: `offer-${connectionId}` },
+  };
+}
+
+function candidate(connectionId: string, value: string): SignalPayload {
+  return {
+    kind: "candidate",
+    connectionId,
+    candidate: {
+      candidate: value,
+      sdpMid: "0",
+      sdpMLineIndex: 0,
+    },
+  };
+}
+
+function createPeer(
+  signals: SignalPayload[],
+  snapshots: PeerSnapshot[],
+): ViewerPeer {
+  return new ViewerPeer(
+    { iceServers: [], expiresAt: null, relayAvailable: false },
+    {
+      sendSignal: (payload) => {
+        signals.push(payload);
+        return true;
+      },
+      sendRestartRequest: () => true,
+      onStream: () => undefined,
+      onUpdate: (snapshot) => snapshots.push(snapshot),
+    },
+  );
+}
+
+async function flushAsyncWork(): Promise<void> {
+  await new Promise<void>((resolve) => setImmediate(resolve));
+}
+
+beforeEach(() => {
+  FakePeerConnection.instances.length = 0;
+  FakePeerConnection.plans.length = 0;
+  intervalCallbacks.clear();
+  nextIntervalId = 1;
+  vi.stubGlobal("MediaStream", FakeMediaStream);
+  vi.stubGlobal("RTCPeerConnection", FakePeerConnection);
+  vi.stubGlobal("window", {
+    setInterval: (callback: () => void) => {
+      const id = nextIntervalId++;
+      intervalCallbacks.set(id, callback);
+      return id;
+    },
+    clearInterval: (id: number) => intervalCallbacks.delete(id),
+    setTimeout,
+    clearTimeout,
+  });
+});
+
+afterEach(() => {
+  vi.unstubAllGlobals();
+});
+
+describe("ViewerPeer connection generations", () => {
+  it("does not send an old answer or candidate after replacing the connection", async () => {
+    const oldLocalDescription = createDeferred<void>();
+    FakePeerConnection.plans.push({
+      localDescriptionGate: oldLocalDescription.promise,
+    });
+    const signals: SignalPayload[] = [];
+    const snapshots: PeerSnapshot[] = [];
+    const peer = createPeer(signals, snapshots);
+
+    const oldAccept = peer.acceptSignal("host", offer("connection-old"));
+    await vi.waitFor(() => {
+      expect(FakePeerConnection.instances[0]?.setLocalDescription).toHaveBeenCalledOnce();
+    });
+    const oldConnection = FakePeerConnection.instances[0]!;
+
+    await peer.acceptSignal("host", offer("connection-new"));
+    const iceEvent = new Event("icecandidate");
+    Object.defineProperty(iceEvent, "candidate", {
+      value: {
+        candidate: "candidate-from-old-connection",
+        sdpMid: "0",
+        sdpMLineIndex: 0,
+        usernameFragment: null,
+      },
+    });
+    oldConnection.dispatchEvent(iceEvent);
+    oldLocalDescription.resolve();
+    await oldAccept;
+
+    expect(signals).toHaveLength(1);
+    expect(signals[0]).toMatchObject({
+      kind: "description",
+      connectionId: "connection-new",
+      description: { type: "answer" },
+    });
+    expect(snapshots.at(-1)).toMatchObject({
+      connectionId: "connection-new",
+      error: null,
+    });
+  });
+
+  it("stops flushing old candidates when the connection is replaced", async () => {
+    const firstCandidate = createDeferred<void>();
+    FakePeerConnection.plans.push({
+      candidateGates: [firstCandidate.promise],
+    });
+    const peer = createPeer([], []);
+    await peer.acceptSignal("host", candidate("connection-old", "candidate-1"));
+    await peer.acceptSignal("host", candidate("connection-old", "candidate-2"));
+
+    const oldAccept = peer.acceptSignal("host", offer("connection-old"));
+    await vi.waitFor(() => {
+      expect(FakePeerConnection.instances[0]?.addIceCandidate).toHaveBeenCalledOnce();
+    });
+    const oldConnection = FakePeerConnection.instances[0]!;
+
+    await peer.acceptSignal("host", offer("connection-new"));
+    const newConnection = FakePeerConnection.instances[1]!;
+    firstCandidate.resolve();
+    await oldAccept;
+
+    expect(oldConnection.addIceCandidate).toHaveBeenCalledTimes(1);
+    expect(newConnection.addIceCandidate).not.toHaveBeenCalled();
+  });
+
+  it("does not apply stats that resolve after the connection is replaced", async () => {
+    const signals: SignalPayload[] = [];
+    const snapshots: PeerSnapshot[] = [];
+    const peer = createPeer(signals, snapshots);
+    await peer.acceptSignal("host", offer("connection-old"));
+    const oldConnection = FakePeerConnection.instances[0]!;
+    const oldStats = createDeferred<RTCStatsReport>();
+    oldConnection.statsGate = oldStats.promise;
+    const oldStatsCallback = intervalCallbacks.get(1)!;
+
+    oldStatsCallback();
+    await vi.waitFor(() => expect(oldConnection.getStats).toHaveBeenCalledOnce());
+    await peer.acceptSignal("host", offer("connection-new"));
+    const newSnapshotCount = snapshots.filter(
+      ({ connectionId }) => connectionId === "connection-new",
+    ).length;
+
+    oldStats.resolve(
+      new Map([
+        [
+          "inbound-old",
+          {
+            id: "inbound-old",
+            type: "inbound-rtp",
+            timestamp: 2_000,
+            kind: "video",
+            bytesReceived: 1_000,
+            framesDecoded: 60,
+            frameWidth: 1_920,
+            frameHeight: 1_080,
+          },
+        ],
+      ]) as unknown as RTCStatsReport,
+    );
+    await flushAsyncWork();
+
+    const newSnapshots = snapshots.filter(
+      ({ connectionId }) => connectionId === "connection-new",
+    );
+    expect(newSnapshots).toHaveLength(newSnapshotCount);
+    expect(newSnapshots.at(-1)?.metrics.resolution).toBeNull();
+  });
+});