feat: F5.3a the simulation rooms on server F

One room per project for everyone in it, held by a few worker_threads, each room with a replica of the document that Hocuspocus' direct connection keeps open: changes go in, what the room writes comes back. /sim?project=... signs in as /sync does and says refusals and partings as the socket's closing; only a write grant changes the room. The state names who started the simulation. After the last person the room stays 60 s, then stops, writes back and ends. Freezing stops a running simulation and refuses to start one, deleting ends the room without writing back, revoking lets the person go from the room too; a failed thread ends its rooms, stopping the service stops them all. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-04 14:29:11 +00:00
parent 5ded84cb33
commit 8cfc518f0c
17 changed files with 1033 additions and 43 deletions
@@ -0,0 +1,339 @@
+// The simulation rooms on F (plan §6): one room for a project, shared by
+// everyone in it, held by one of a few threads (worker_threads; the one
+// with the fewest rooms takes a new one). A room keeps its project's
+// document open on F (a direct connection: what the simulation leaves goes
+// into the document even when no one has it open) and gets every change of
+// it; what the room writes comes back as the room's own change.
+//
+// Who may do what: everyone in the project watches; only with a write grant
+// does one start, stop, set the speed, hold "Pakete verwerfen", clear the
+// capture or act in a window (WRITING_KINDS; else `readOnly`). A frozen
+// project's simulation does not start (`frozen`); freezing stops a running
+// one, what it left written back first.
+//
+// When the last person leaves, the room stays LINGER_MILLISECONDS (a tab
+// changed, a tablet locked, a short drop of the network: plan §6.2), then
+// stops — what the simulation left written back — and ends. A thread that
+// fails ends its rooms: their people are let go (SIM_ROOM_FAILED_CODE) and
+// find a new room, in design mode, when they come back.
+
+import { availableParallelism } from 'node:os';
+import { Worker } from 'node:worker_threads';
+import type { DirectConnection, Hocuspocus } from '@hocuspocus/server';
+import * as Y from 'yjs';
+import { SIM_CLOSED_CODE, SIM_ROOM_FAILED_CODE, WRITING_KINDS, type ClientMessageKind, type HostMessage, type SyncClosing } from '@mdg-filius/protocol';
+import type WebSocket from 'ws';
+import { errorText, type Log } from './log.js';
+import type { FromRoomThread, ToRoomThread } from './simChannel.js';
+
+export const LINGER_MILLISECONDS = 60_000;
+
+/** Who is in a room. */
+export interface SimPerson {
+  readonly user: string;
+  readonly login: string;
+  readonly name: string;
+  /** A write grant: may change the room. */
+  readonly write: boolean;
+}
+
+export interface SimRoomsOptions {
+  readonly hocuspocus: Hocuspocus;
+  readonly log: Log;
+  /** The bundled room thread (dist/simWorker.mjs). */
+  readonly workerUrl: URL;
+  /** How many threads hold rooms; all cores but one by default. */
+  readonly threads?: number;
+  readonly lingerMilliseconds?: number;
+}
+
+interface Thread {
+  worker: Worker;
+  readonly rooms: Set<Room>;
+}
+
+interface Room {
+  readonly project: string;
+  readonly thread: Thread;
+  readonly direct: DirectConnection;
+  readonly doc: Y.Doc;
+  readonly clients: Map<number, { readonly socket: WebSocket; readonly person: SimPerson }>;
+  readonly forward: (update: Uint8Array, origin: unknown) => void;
+  /** What waits for the thread's answer to `stop` and `close`. */
+  readonly waiting: { stopped: (() => void)[]; closed: (() => void)[] };
+  frozen: boolean;
+  linger: ReturnType<typeof setTimeout> | undefined;
+  ended: boolean;
+}
+
+type RawData = Buffer | ArrayBuffer | Buffer[];
+
+/** What a socket sent while F looked the person up, held until the room hears it. */
+export interface HeldMessages {
+  readonly messages: readonly RawData[];
+  /** Stops holding: the room's listener takes over. */
+  release(): void;
+}
+
+/** Holds a socket's messages from now on. */
+export function holdMessages(socket: WebSocket): HeldMessages {
+  const messages: RawData[] = [];
+  const hold = (data: RawData) => messages.push(data);
+  socket.on('message', hold);
+  return { messages, release: () => socket.off('message', hold) };
+}
+
+const text = (data: RawData) => (Array.isArray(data) ? Buffer.concat(data) : Buffer.from(data as ArrayBuffer)).toString('utf8');
+
+export class SimRooms {
+  private readonly threads: Thread[] = [];
+  private readonly rooms = new Map<string, Promise<Room>>();
+  private readonly ending = new Map<string, Promise<void>>();
+  private nextClient = 1;
+  private closing = false;
+
+  constructor(private readonly options: SimRoomsOptions) {
+    const count = options.threads ?? Math.max(1, availableParallelism() - 1);
+    for (let index = 0; index < count; index++) this.threads.push(this.spawn());
+  }
+
+  private spawn(): Thread {
+    const thread: Thread = { worker: new Worker(this.options.workerUrl), rooms: new Set() };
+    thread.worker.on('message', (message: FromRoomThread) => this.fromThread(thread, message));
+    thread.worker.on('error', (error) => this.options.log.error('sim.threadError', { error: errorText(error) }));
+    thread.worker.on('exit', (code) => {
+      if (!this.closing) this.threadFailed(thread, code);
+    });
+    return thread;
+  }
+
+  private threadFailed(thread: Thread, code: number): void {
+    this.options.log.error('sim.threadFailed', { code, rooms: thread.rooms.size });
+    for (const room of thread.rooms) {
+      for (const { socket } of room.clients.values()) socket.close(SIM_ROOM_FAILED_CODE, 'roomFailed');
+      this.forget(room);
+      void room.direct.disconnect();
+    }
+    this.threads[this.threads.indexOf(thread)] = this.spawn();
+  }
+
+  private post(room: Room, message: ToRoomThread): void {
+    room.thread.worker.postMessage(message);
+  }
+
+  private fromThread(thread: Thread, message: FromRoomThread): void {
+    const room = [...thread.rooms].find((each) => each.project === message.room);
+    if (room === undefined) return;
+    switch (message.type) {
+      case 'doc':
+        Y.applyUpdate(room.doc, message.update, room);
+        return;
+      case 'host':
+        room.clients.get(message.client)?.socket.send(JSON.stringify(message.message));
+        return;
+      case 'stopped':
+      case 'closed':
+        room.waiting[message.type].shift()?.();
+        return;
+    }
+  }
+
+  /** The room of a project, made when there is none (its document loaded; refused if F does not know the project). */
+  private async roomOf(project: string, frozen: boolean): Promise<Room> {
+    // A room ending is let go first: its document's direct connection must not meet Hocuspocus unloading it.
+    await this.ending.get(project);
+    let made = this.rooms.get(project);
+    if (made === undefined) {
+      made = this.open(project, frozen);
+      this.rooms.set(project, made);
+      made.catch(() => this.rooms.delete(project));
+    }
+    return made;
+  }
+
+  private async open(project: string, frozen: boolean): Promise<Room> {
+    const direct = await this.options.hocuspocus.openDirectConnection(project, { room: true });
+    const thread = this.threads.reduce((fewest, each) => (each.rooms.size < fewest.rooms.size ? each : fewest));
+    const doc = direct.document!;
+    const room: Room = {
+      project,
+      thread,
+      direct,
+      doc,
+      clients: new Map(),
+      forward: (update, origin) => {
+        if (origin !== room) this.post(room, { type: 'doc', room: project, update });
+      },
+      waiting: { stopped: [], closed: [] },
+      frozen,
+      linger: undefined,
+      ended: false,
+    };
+    thread.rooms.add(room);
+    this.post(room, { type: 'open', room: project, state: Y.encodeStateAsUpdate(doc) });
+    doc.on('update', room.forward);
+    this.options.log.info('sim.roomOpened', { project });
+    return room;
+  }
+
+  /**
+   * A person in a project's room: the socket's messages go to the room, the
+   * room's to the socket, with what it sent before (`held`, while F looked
+   * the person up). Rejected if F does not know the project.
+   */
+  async join(socket: WebSocket, project: string, person: SimPerson, frozen: boolean, held: HeldMessages = holdMessages(socket)): Promise<void> {
+    const room = await this.roomOf(project, frozen);
+    // Ended while it was made (its project deleted, its thread failed): the next room.
+    if (room.ended) return this.join(socket, project, person, frozen, held);
+    // Gone while the room was made: a room no one came to ends as one everyone left.
+    if (socket.readyState !== socket.OPEN) {
+      held.release();
+      if (room.clients.size === 0 && room.linger === undefined && !room.ended) this.lingerOn(room);
+      return;
+    }
+    const client = this.nextClient++;
+    room.clients.set(client, { socket, person });
+    if (room.linger !== undefined) clearTimeout(room.linger);
+    room.linger = undefined;
+    this.post(room, { type: 'connect', room: project, client, name: person.name });
+    this.options.log.info('sim.joined', { project, user: person.login, write: person.write });
+    const receive = (data: RawData) => this.receive(room, client, person, data);
+    held.release();
+    for (const data of held.messages) receive(data);
+    socket.on('message', receive);
+    socket.on('close', () => this.leave(room, client));
+  }
+
+  private receive(room: Room, client: number, person: SimPerson, data: RawData): void {
+    const answer = (message: Omit<Extract<HostMessage, { kind: 'error' }>, 'v'>) => room.clients.get(client)?.socket.send(JSON.stringify({ v: 1, ...message }));
+    let message: unknown;
+    try {
+      message = JSON.parse(text(data));
+    } catch {
+      answer({ kind: 'error', code: 'badMessage', detail: 'not JSON' });
+      return;
+    }
+    const { kind, reqId } = (typeof message === 'object' && message !== null ? message : {}) as { kind?: unknown; reqId?: unknown };
+    const request = typeof reqId === 'string' ? { reqId } : {};
+    if (WRITING_KINDS.includes(kind as ClientMessageKind)) {
+      if (!person.write) {
+        answer({ kind: 'error', code: 'readOnly', ...request });
+        return;
+      }
+      if (kind === 'start' && room.frozen) {
+        answer({ kind: 'error', code: 'frozen', ...request });
+        return;
+      }
+      if (kind === 'start' || kind === 'stop') this.options.log.info(`sim.${kind}`, { project: room.project, user: person.login });
+    }
+    this.post(room, { type: 'client', room: room.project, client, message });
+  }
+
+  private leave(room: Room, client: number): void {
+    if (!room.clients.delete(client) || room.ended) return;
+    this.post(room, { type: 'disconnect', room: room.project, client });
+    if (room.clients.size === 0) this.lingerOn(room);
+  }
+
+  private lingerOn(room: Room): void {
+    room.linger = setTimeout(() => void this.end(room, true), this.options.lingerMilliseconds ?? LINGER_MILLISECONDS);
+  }
+
+  /** The room no longer reachable: a new one is made for whoever comes. */
+  private forget(room: Room): void {
+    room.ended = true;
+    if (room.linger !== undefined) clearTimeout(room.linger);
+    room.doc.off('update', room.forward);
+    room.thread.rooms.delete(room);
+    void this.rooms.get(room.project)?.then(
+      (current) => {
+        if (current === room) this.rooms.delete(room.project);
+      },
+      () => {},
+    );
+    for (const resolve of [...room.waiting.stopped, ...room.waiting.closed]) resolve();
+  }
+
+  private ask(room: Room, message: Extract<ToRoomThread, { type: 'stop' | 'close' }>): Promise<void> {
+    return new Promise((resolve) => {
+      room.waiting[message.type === 'stop' ? 'stopped' : 'closed'].push(resolve);
+      this.post(room, message);
+    });
+  }
+
+  /** Ends a room: stopped first with `writeBack` (what it left in the document), else as it is; its document let go. */
+  private end(room: Room, writeBack: boolean): Promise<void> {
+    if (room.ended) return this.ending.get(room.project) ?? Promise.resolve();
+    const done = (async () => {
+      this.rooms.delete(room.project);
+      await this.ask(room, { type: 'close', room: room.project, writeBack });
+      this.forget(room);
+      await room.direct.disconnect();
+      this.options.log.info('sim.roomClosed', { project: room.project, writeBack });
+    })();
+    room.ended = true;
+    this.ending.set(room.project, done);
+    void done.finally(() => {
+      if (this.ending.get(room.project) === done) this.ending.delete(room.project);
+    });
+    return done;
+  }
+
+  private async existing(project: string): Promise<Room | undefined> {
+    const room = await this.rooms.get(project)?.catch(() => undefined);
+    return room === undefined || room.ended ? undefined : room;
+  }
+
+  /** A project frozen (its running simulation stopped, what it left written back) or thawed. */
+  async freeze(project: string, frozen: boolean): Promise<void> {
+    const room = await this.existing(project);
+    if (room === undefined) return;
+    room.frozen = frozen;
+    if (frozen) await this.ask(room, { type: 'stop', room: project });
+  }
+
+  /** A project deleted: its room ends without writing back, its people are let go; how many. */
+  async drop(project: string): Promise<number> {
+    const room = await this.existing(project);
+    if (room === undefined) return 0;
+    const sockets = [...room.clients.values()].map(({ socket }) => socket);
+    await this.end(room, false);
+    return this.letGo(sockets, 'deleted');
+  }
+
+  /** A person's grant ended, to one project or to all: their sockets let go; how many. */
+  disconnect(user: string, project?: string): number {
+    const sockets: WebSocket[] = [];
+    for (const thread of this.threads)
+      for (const room of thread.rooms) {
+        if (project !== undefined && room.project !== project) continue;
+        for (const { socket, person } of room.clients.values()) if (person.user === user) sockets.push(socket);
+      }
+    return this.letGo(sockets, 'revoked');
+  }
+
+  private letGo(sockets: readonly WebSocket[], reason: SyncClosing): number {
+    for (const socket of sockets) socket.close(SIM_CLOSED_CODE, reason);
+    return sockets.length;
+  }
+
+  /** How many rooms, people in them, and rooms each thread holds. */
+  stats(): { readonly rooms: number; readonly people: number; readonly threads: readonly number[] } {
+    const rooms = this.threads.flatMap((thread) => [...thread.rooms]);
+    return { rooms: rooms.length, people: rooms.reduce((sum, room) => sum + room.clients.size, 0), threads: this.threads.map((thread) => thread.rooms.size) };
+  }
+
+  /** The threads' workers (for the stats' tests: one may be ended to see the rooms go). */
+  get workers(): readonly Worker[] {
+    return this.threads.map((thread) => thread.worker);
+  }
+
+  /** Every room stopped and ended (what it left written back), every person let go, the threads ended. */
+  async close(): Promise<void> {
+    const rooms = this.threads.flatMap((thread) => [...thread.rooms]);
+    await Promise.all(rooms.map((room) => this.end(room, true)));
+    this.closing = true;
+    for (const room of rooms) for (const { socket } of room.clients.values()) socket.close(1001, 'service stopping');
+    await Promise.all(this.threads.map((thread) => thread.worker.terminate()));
+  }
+}