fix: harden presentation sync lifecycle

web/apps/server/src/index.ts

@@ -15,6 +15,7 @@ import {
 } from "@lda/workflow-rpc";
 import { createApp, type RunOperation } from "./app.js";
 import { createPresentationRoomService } from "./presentation-sync/rooms.js";
+import { shutdownServer } from "./shutdown.js";
 import { WebSocketServer } from "ws";
 
 const port = Number(process.env.WEB_PORT ?? "8787");
@@ -75,12 +76,13 @@ expirySweep.unref();
 
 console.log(`workflow console server listening on http://${hostname}:${port}`);
 
+let shuttingDown = false;
 const shutdown = (signal: NodeJS.Signals) => {
+  if (shuttingDown) return;
+  shuttingDown = true;
   console.log(`received ${signal}, stopping workflow console server`);
   clearInterval(expirySweep);
-  wss.close();
-  server.close(() => process.exit(0));
-  setTimeout(() => process.exit(1), 5_000).unref();
+  shutdownServer({ server, wss, exit: (code) => process.exit(code) });
 };
 
 process.once("SIGINT", shutdown);

web/apps/server/src/presentation-sync/rooms.test.ts

@@ -4,6 +4,7 @@ import type { PresentationPeer, PresentationRoomService } from "./rooms.js";
 import {
   createPresentationRoomService,
   EMPTY_ROOM_GRACE_MS,
+  PresentationRoomJoinError,
   ROOM_INACTIVITY_TTL_MS,
 } from "./rooms.js";
 
@@ -78,6 +79,27 @@ describe("createPresentationRoomService", () => {
     expect(joined.connectionToken).not.toBe(created.connectionToken);
   });
 
+  it("reports missing sessions and invalid roles with stable typed errors", () => {
+    const { service } = makeService();
+    const created = service.create({
+      role: "presenter",
+      initialHash: "#scene/thesis/title",
+    });
+
+    for (const [input, code] of [
+      [{ role: "audience" as const, code: "NOPE00" }, "session_not_found"],
+      [{ role: "presenter" as const, code: created.code }, "invalid_role"],
+    ] as const) {
+      try {
+        service.join(input);
+        throw new Error("expected join to fail");
+      } catch (error) {
+        expect(error).toBeInstanceOf(PresentationRoomJoinError);
+        expect((error as PresentationRoomJoinError).code).toBe(code);
+      }
+    }
+  });
+
   it("allocates unique room codes", () => {
     const { service } = makeService();
 
@@ -145,7 +167,7 @@ describe("createPresentationRoomService", () => {
 
     service.disconnect(created.connectionToken, oldPeer);
 
-    expect(service.publish(created.connectionToken, {
+    expect(service.publish(created.connectionToken, newPeer, {
       type: "location.publish",
       hash: "#scene/replacement-survives",
       baseRevision: 0,
@@ -158,19 +180,61 @@ describe("createPresentationRoomService", () => {
     });
   });
 
+  it("rejects publish and end commands from a replaced presenter peer", () => {
+    const { service } = makeService();
+    const created = service.create({ role: "presenter", initialHash: "#scene/one" });
+    const oldPeer = peer();
+    const replacement = peer();
+    service.connect(created.connectionToken, oldPeer);
+    service.connect(created.connectionToken, replacement);
+    replacement.send.mockClear();
+
+    expect(service.publish(created.connectionToken, oldPeer, {
+      type: "location.publish",
+      hash: "#scene/stale-peer",
+      baseRevision: 0,
+      messageId: "old-1",
+    })).toEqual({ kind: "not_connected" });
+    expect(service.end(created.connectionToken, oldPeer)).toEqual({
+      kind: "not_connected",
+    });
+    expect(replacement.close).not.toHaveBeenCalled();
+
+    expect(service.publish(created.connectionToken, replacement, {
+      type: "location.publish",
+      hash: "#scene/replacement-active",
+      baseRevision: 0,
+      messageId: "new-1",
+    }).kind).toBe("accepted");
+  });
+
+  it("does not let a replaced peer touch room activity", () => {
+    const { service, advance } = makeService();
+    const created = service.create({ role: "presenter", initialHash: "#scene/one" });
+    const oldPeer = peer();
+    const replacement = peer();
+    service.connect(created.connectionToken, oldPeer);
+    service.connect(created.connectionToken, replacement);
+
+    advance(ROOM_INACTIVITY_TTL_MS);
+    service.ping(created.connectionToken, oldPeer);
+
+    expect(service.sweepExpired()).toBe(1);
+  });
+
   it("accepts one publish and rejects a stale competing publish", () => {
     const { service, presenter, audience, presenterToken, audienceToken } =
       connectedRoom();
 
     expect(
-      service.publish(presenterToken, {
+      service.publish(presenterToken, presenter, {
         type: "location.publish",
         hash: "#scene/problem/direct-actions",
         baseRevision: 0,
         messageId: "presenter-1",
       }).kind,
     ).toBe("accepted");
-    expect(service.publish(audienceToken, {
+    expect(service.publish(audienceToken, audience, {
       type: "location.publish",
       hash: "#scene/positioning/landscape",
       baseRevision: 0,
@@ -205,7 +269,7 @@ describe("createPresentationRoomService", () => {
       presence: { presenters: 1, audience: 0 },
     });
     expect(audience.send).not.toHaveBeenCalled();
-    expect(service.publish(presenterToken, {
+    expect(service.publish(presenterToken, presenter, {
       type: "location.publish",
       hash: "#scene/after-disconnect",
       baseRevision: 0,
@@ -270,10 +334,10 @@ describe("createPresentationRoomService", () => {
   });
 
   it("counts ping as room activity", () => {
-    const { service, advance, presenterToken } = connectedRoom();
+    const { service, advance, presenter, presenterToken } = connectedRoom();
 
     advance(ROOM_INACTIVITY_TTL_MS - 1);
-    service.ping(presenterToken);
+    service.ping(presenterToken, presenter);
     advance(ROOM_INACTIVITY_TTL_MS - 1);
     expect(service.sweepExpired()).toBe(0);
     advance(1);
@@ -283,7 +347,7 @@ describe("createPresentationRoomService", () => {
   it("ends a room when the presenter terminates it", () => {
     const { service, presenter, audience, presenterToken } = connectedRoom();
 
-    expect(service.end(presenterToken)).toEqual({ kind: "ended" });
+    expect(service.end(presenterToken, presenter)).toEqual({ kind: "ended" });
     expect(presenter.send).toHaveBeenCalledWith({
       type: "session.ended",
       reason: "presenter_ended",
@@ -300,14 +364,14 @@ describe("createPresentationRoomService", () => {
     const { service, presenter, audience, audienceToken, presenterToken } =
       connectedRoom();
 
-    expect(service.end(audienceToken)).toEqual({ kind: "forbidden" });
+    expect(service.end(audienceToken, audience)).toEqual({ kind: "forbidden" });
     expect(audience.send).toHaveBeenCalledWith({
       type: "protocol.error",
       code: "forbidden",
       message: "only the presenter can end the session",
     });
     expect(presenter.send).not.toHaveBeenCalled();
-    expect(service.publish(presenterToken, {
+    expect(service.publish(presenterToken, presenter, {
       type: "location.publish",
       hash: "#scene/still-active",
       baseRevision: 0,
@@ -320,7 +384,7 @@ describe("createPresentationRoomService", () => {
     const first = connectRoom(service);
     const second = connectRoom(service);
 
-    expect(service.publish(first.presenterToken, {
+    expect(service.publish(first.presenterToken, first.presenter, {
       type: "location.publish",
       hash: "#scene/first-room",
       baseRevision: 0,

web/apps/server/src/presentation-sync/rooms.ts

@@ -12,6 +12,16 @@ import {
 export const EMPTY_ROOM_GRACE_MS = 10 * 60 * 1_000;
 export const ROOM_INACTIVITY_TTL_MS = 2 * 60 * 60 * 1_000;
 
+export class PresentationRoomJoinError extends Error {
+  constructor(
+    readonly code: "session_not_found" | "invalid_role",
+    message: string,
+  ) {
+    super(message);
+    this.name = "PresentationRoomJoinError";
+  }
+}
+
 export type PresentationPeer = {
   readonly send: (message: ServerSyncMessage) => void;
   readonly close: (code: number, reason: string) => void;
@@ -204,10 +214,16 @@ export const createPresentationRoomService = (options: {
       const roomId = roomIdByCode.get(code);
       const room = roomId === undefined ? undefined : roomsById.get(roomId);
       if (room === undefined) {
-        throw new Error("presentation room not found");
+        throw new PresentationRoomJoinError(
+          "session_not_found",
+          "presentation room not found",
+        );
       }
       if (input.role === room.creatorRole) {
-        throw new Error("presentation room requires the opposite role");
+        throw new PresentationRoomJoinError(
+          "invalid_role",
+          "presentation room requires the opposite role",
+        );
       }
 
       const token = nextUnique(
@@ -272,6 +288,7 @@ export const createPresentationRoomService = (options: {
 
     publish(
       token: string,
+      peer: PresentationPeer,
       message: Extract<ClientSyncMessage, { type: "location.publish" }>,
     ): PublishResult {
       const membership = membershipByToken.get(token);
@@ -279,7 +296,7 @@ export const createPresentationRoomService = (options: {
       if (membership === undefined || room === undefined || !roomsById.has(room.id)) {
         return { kind: "not_found" };
       }
-      if (membership.peer === null) return { kind: "not_connected" };
+      if (membership.peer !== peer) return { kind: "not_connected" };
 
       room.lastActivityAt = now();
       if (message.baseRevision !== room.snapshot.revision) {
@@ -304,27 +321,27 @@ export const createPresentationRoomService = (options: {
       return { kind: "accepted", snapshot: room.snapshot };
     },
 
-    ping(token: string): void {
+    ping(token: string, peer: PresentationPeer): void {
       const membership = membershipByToken.get(token);
       const room = membership?.room;
       if (
         membership === undefined ||
         room === undefined ||
         !roomsById.has(room.id) ||
-        membership.peer === null
+        membership.peer !== peer
       ) {
         return;
       }
       room.lastActivityAt = now();
     },
 
-    end(token: string): EndResult {
+    end(token: string, peer: PresentationPeer): EndResult {
       const membership = membershipByToken.get(token);
       const room = membership?.room;
       if (membership === undefined || room === undefined || !roomsById.has(room.id)) {
         return { kind: "not_found" };
       }
-      if (membership.peer === null) return { kind: "not_connected" };
+      if (membership.peer !== peer) return { kind: "not_connected" };
       if (membership.role !== "presenter") {
         room.lastActivityAt = now();
         membership.peer.send({

web/apps/server/src/presentation-sync/routes.test.ts

@@ -164,6 +164,54 @@ describe("presentation synchronization routes", () => {
     });
     expect(missing.status).toBe(404);
     expect(await missing.json()).toMatchObject({ error: { code: "session_not_found" } });
+
+    const invalidRole = await postJson(origin, "/api/presentation-sync/sessions/join", {
+      role: "presenter",
+      code: created.code,
+    });
+    expect(invalidRole.status).toBe(400);
+    expect(await invalidRole.json()).toEqual({
+      error: {
+        code: "invalid_role",
+        message: "presentation room requires the opposite role",
+      },
+    });
+  });
+
+  it("ignores commands from a replaced presenter socket", async () => {
+    const { origin } = await startServer();
+    const presenterGrant = await grant(origin, "/api/presentation-sync/sessions", {
+      role: "presenter",
+      initialHash: "#scene/original",
+    });
+    const oldPresenter = await trackedConnect(origin, presenterGrant.connectionToken);
+    await expectMessage(oldPresenter, (message) => message.type === "location.snapshot");
+
+    const oldClose = once(oldPresenter, "close");
+    const replacement = await trackedConnect(origin, presenterGrant.connectionToken);
+    expect((await oldClose)[0]).toBe(4001);
+    oldPresenter.send(JSON.stringify({
+      type: "location.publish",
+      hash: "#scene/old-must-not-win",
+      baseRevision: 0,
+      messageId: "old-publish",
+    }));
+    oldPresenter.send(JSON.stringify({ type: "session.end" }));
+
+    replacement.send(JSON.stringify({
+      type: "location.publish",
+      hash: "#scene/replacement-wins",
+      baseRevision: 0,
+      messageId: "replacement-publish",
+    }));
+    await expectMessage(
+      replacement,
+      (message) =>
+        message.type === "location.snapshot" &&
+        message.snapshot.hash === "#scene/replacement-wins" &&
+        message.snapshot.revision === 1,
+    );
+    expect(replacement.readyState).toBe(WebSocket.OPEN);
   });
 
   it("synchronizes two real clients through reconnect and presenter shutdown", async () => {

web/apps/server/src/presentation-sync/routes.ts

@@ -8,7 +8,11 @@ import {
   type SessionGrant,
 } from "@lda/presentation-sync";
 import type { Hono } from "hono";
-import type { PresentationPeer, PresentationRoomService } from "./rooms.js";
+import {
+  PresentationRoomJoinError,
+  type PresentationPeer,
+  type PresentationRoomService,
+} from "./rooms.js";
 
 type PresentationSyncDependencies = {
   readonly rooms: PresentationRoomService;
@@ -52,14 +56,10 @@ export const addPresentationSyncRoutes = (
       const grant: SessionGrant = rooms.join(decoded.value);
       return c.json(grant, 200);
     } catch (error) {
-      const message = error instanceof Error ? error.message : "presentation room not found";
-      if (message.includes("opposite role")) {
-        return c.json({ error: { code: "invalid_role", message } }, 400);
-      }
-      return c.json(
-        { error: { code: "session_not_found", message: "presentation room not found" } },
-        404,
-      );
+      if (!(error instanceof PresentationRoomJoinError)) throw error;
+      return error.code === "invalid_role"
+        ? c.json({ error: { code: error.code, message: error.message } }, 400)
+        : c.json({ error: { code: error.code, message: error.message } }, 404);
     }
   });
 
@@ -103,13 +103,13 @@ export const addPresentationSyncRoutes = (
 
           switch (decoded.value.type) {
             case "location.publish":
-              rooms.publish(token, decoded.value);
+              rooms.publish(token, peer, decoded.value);
               break;
             case "ping":
-              rooms.ping(token);
+              rooms.ping(token, peer);
               break;
             case "session.end":
-              rooms.end(token);
+              rooms.end(token, peer);
               break;
           }
         },

web/apps/server/src/shutdown.test.ts

@@ -0,0 +1,105 @@
+import { EventEmitter, once } from "node:events";
+import type { AddressInfo } from "node:net";
+import {
+  serve,
+  upgradeWebSocket,
+  type WebSocketServerLike,
+} from "@hono/node-server";
+import { Hono } from "hono";
+import { describe, expect, it, vi } from "vitest";
+import WebSocket, { WebSocketServer } from "ws";
+import { shutdownServer } from "./shutdown.js";
+
+const closable = () => ({
+  close: vi.fn<(code: number, reason: string) => void>(),
+  terminate: vi.fn<() => void>(),
+});
+
+describe("shutdownServer", () => {
+  it("completes cleanly with a real active WebSocket client", async () => {
+    const app = new Hono();
+    app.get("/ws", upgradeWebSocket(() => ({})));
+    const wss = new WebSocketServer({ noServer: true });
+    const server = serve({
+      fetch: app.fetch,
+      websocket: { server: wss as WebSocketServerLike },
+      port: 0,
+    });
+    if (!server.listening) await once(server, "listening");
+    const { port } = server.address() as AddressInfo;
+    const client = new WebSocket(`ws://127.0.0.1:${port}/ws`);
+    await once(client, "open");
+    const clientClosed = once(client, "close");
+    const exited = new Promise<number>((resolve) => {
+      shutdownServer({ server, wss, exit: resolve });
+    });
+
+    expect((await clientClosed)[0]).toBe(1001);
+    await expect(exited).resolves.toBe(0);
+  });
+
+  it("closes active WebSocket clients and exits zero after both servers close", () => {
+    const client = closable();
+    let closeHttp: (() => void) | undefined;
+    let closeWebSockets: (() => void) | undefined;
+    const exit = vi.fn<(code: number) => void>();
+    const clearForceTimeout = vi.fn<(handle: ReturnType<typeof setTimeout>) => void>();
+    const forceHandle = { unref: vi.fn() } as unknown as ReturnType<typeof setTimeout>;
+    const setForceTimeout = vi.fn(() => forceHandle);
+
+    shutdownServer({
+      server: {
+        close: (callback) => {
+          closeHttp = callback;
+          return new EventEmitter() as never;
+        },
+        closeAllConnections: vi.fn(),
+      },
+      wss: {
+        clients: new Set([client]),
+        close: (callback) => {
+          closeWebSockets = callback;
+        },
+      },
+      exit,
+      setForceTimeout,
+      clearForceTimeout,
+    });
+
+    expect(client.close).toHaveBeenCalledWith(1001, "server shutdown");
+    closeWebSockets?.();
+    expect(exit).not.toHaveBeenCalled();
+    closeHttp?.();
+    expect(exit).toHaveBeenCalledExactlyOnceWith(0);
+    expect(clearForceTimeout).toHaveBeenCalledWith(forceHandle);
+    expect(setForceTimeout).toHaveBeenCalledWith(expect.any(Function), 5_000);
+    expect(forceHandle.unref).toHaveBeenCalled();
+  });
+
+  it("force-terminates lingering clients and chooses exit one only on timeout", () => {
+    const client = closable();
+    const exit = vi.fn<(code: number) => void>();
+    const closeAllConnections = vi.fn();
+    let force: (() => void) | undefined;
+
+    shutdownServer({
+      server: {
+        close: vi.fn(() => new EventEmitter() as never),
+        closeAllConnections,
+      },
+      wss: { clients: new Set([client]), close: vi.fn() },
+      exit,
+      setForceTimeout: (callback) => {
+        force = callback;
+        return { unref: vi.fn() } as unknown as ReturnType<typeof setTimeout>;
+      },
+      clearForceTimeout: vi.fn(),
+    });
+
+    expect(exit).not.toHaveBeenCalled();
+    force?.();
+    expect(client.terminate).toHaveBeenCalledOnce();
+    expect(closeAllConnections).toHaveBeenCalledOnce();
+    expect(exit).toHaveBeenCalledExactlyOnceWith(1);
+  });
+});

web/apps/server/src/shutdown.ts

@@ -0,0 +1,79 @@
+type ShutdownClient = {
+  readonly close: (code: number, reason: string) => void;
+  readonly terminate: () => void;
+};
+
+type ShutdownHttpServer = {
+  readonly close: (callback: (error?: Error) => void) => unknown;
+  readonly closeAllConnections?: () => void;
+};
+
+type ShutdownWebSocketServer = {
+  readonly clients: ReadonlySet<ShutdownClient>;
+  readonly close: (callback: (error?: Error) => void) => void;
+};
+
+type ForceTimeout = NodeJS.Timeout;
+
+export const shutdownServer = (dependencies: {
+  readonly server: ShutdownHttpServer;
+  readonly wss: ShutdownWebSocketServer;
+  readonly exit: (code: number) => void;
+  readonly setForceTimeout?: (
+    callback: () => void,
+    milliseconds: number,
+  ) => ForceTimeout;
+  readonly clearForceTimeout?: (timeout: ForceTimeout) => void;
+  readonly forceAfterMs?: number;
+}): void => {
+  const {
+    server,
+    wss,
+    exit,
+    setForceTimeout = (callback, milliseconds) =>
+      setTimeout(callback, milliseconds) as NodeJS.Timeout,
+    clearForceTimeout = (timeout) => clearTimeout(timeout),
+    forceAfterMs = 5_000,
+  } = dependencies;
+  let httpClosed = false;
+  let webSocketsClosed = false;
+  let finished = false;
+
+  const finish = (code: number): void => {
+    if (finished) return;
+    finished = true;
+    clearForceTimeout(forceTimeout);
+    exit(code);
+  };
+  const finishNormallyWhenClosed = (): void => {
+    if (httpClosed && webSocketsClosed) finish(0);
+  };
+
+  // A WebSocket server waits for clients to disconnect before its close callback;
+  // initiate their closing handshakes before waiting on either server boundary.
+  for (const client of wss.clients) client.close(1001, "server shutdown");
+
+  const forceTimeout = setForceTimeout(() => {
+    for (const client of wss.clients) client.terminate();
+    server.closeAllConnections?.();
+    finish(1);
+  }, forceAfterMs);
+  forceTimeout.unref();
+
+  wss.close((error) => {
+    if (error) {
+      finish(1);
+      return;
+    }
+    webSocketsClosed = true;
+    finishNormallyWhenClosed();
+  });
+  server.close((error) => {
+    if (error) {
+      finish(1);
+      return;
+    }
+    httpClosed = true;
+    finishNormallyWhenClosed();
+  });
+};