diff --git a/CHANGES.md b/CHANGES.md index 9121d05..1065b0f 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -176,6 +176,7 @@ Swift-Pfade relativ zu `filius-ipad/ios/FiliusPad/TopologyEditor/` (Stand | `src/gnutella.ts` | Java `GUIApplicationPeerToPeerAnwendungWindow` (gemessen mit `GnutellaWindowRun`) | neu (F4.6g4): der Bildschirmstand von Gnutella — die bekannten Teilnehmer, die Ergebnisse wie die Tabelle sie zeigt, die Dateien von `/peer2peer`, die Zahl der Teilnehmer | | `src/viewers.ts`, `src/sim.ts` (`sat`, `nat`, `drop`, `dropped`, `clock`) | Java `SatViewer`, `NatViewer`, `SimulationControl` (gemessen mit `ViewersRun`) | neu (F4.6g5): die Tabellen eines Switches und eines Heimrouters als Bildschirmstände; die Nachricht `drop` (Javas „Pakete verwerfen“ gedrückt oder losgelassen), das Thema `dropped` mit den verworfenen Frames; der Zustand trägt `clock`, die Uhrzeit, für die die Zeit 0 der Simulation steht | | `src/sim.ts` (`clearCapture`) | Java `AggregatedMessageTable` („Tabellen löschen“, `Lauscher.reset`) | neu (F4.7): die Nachricht `clearCapture` | +| `src/sim.ts` (`startedBy`, `readOnly`, `frozen`, `WRITING_KINDS`), `src/sync.ts` (Schließcodes von `/sim`) | Plan §6.2, §6.4 | erweitert (F5.3a): der Zustand nennt, wer startete; was einen Raum ändert (Start, Stopp, Geschwindigkeit, Verwerfen, Tabellen löschen, Eingaben), nur mit Schreibstufe; ein eingefrorenes Projekt startet nicht; `/sim` sagt Absage und Trennung als Schließen des WebSockets (4403 mit dem Absagegrund, 4410 mit dem Trennungsgrund, 4500 wenn der Faden des Raums scheiterte) | | `src/sync.ts` | Plan §8.5 | neu (F5.2): was F einem Editor über seinen Platz im gemeinsamen Dokument sagt — warum eine Verbindung abgewiesen wird (`notAProject`, `notSignedIn`, `noAccess`, `unknownProject`), warum F sie schließt (`revoked`, `deleted`), und die Nachricht `access`, wenn sich der Zugang einer bleibenden Verbindung ändert (eingefroren, aufgetaut) | ## packages/sim @@ -200,6 +201,7 @@ Swift-Pfade relativ zu `filius-ipad/ios/FiliusPad/TopologyEditor/` (Stand | `src/room.ts` (Anzeigen, „Pakete verwerfen“), `test/viewers.test.ts`, `test/room.test.ts` | Java `SatViewer`, `NatViewer`, `SimulationControl`, `Verbindung.setDrop` | erweitert (F4.6g5): der Raum schickt die Tabellen eines Switches und eines Heimrouters (ohne Zeilen) und nimmt ihr `reset`; jedes Kabel verwirft, solange irgendein Client „Pakete verwerfen“ hält (ein Client, der geht oder beim Stopp noch hält, zählt nicht mehr); die verworfenen Frames gehen an, wer sie beobachtet, mit jedem Start neu begonnen wie Javas Liste | | `src/room.ts` („Tabellen löschen“), `test/exchangeWindow.test.ts` | Java `Lauscher.reset`, `GUIContainer.closeDialogs/reopenDialogs` | erweitert (F4.7): `clearCapture` leert im Lauf die Listen aller Karten und die verworfenen Frames, jeder Beobachter bekommt sie neu (ab 1 nummeriert), das Kabel leuchtet schon mit dem ersten neuen Frame; im Entwurf `notRunning`. Der Mitschnitt bleibt über Stopp und Start, wie Javas | | `browser/m9.spec.ts`, `browser/traceDigest.ts`, `browser/*Entry.ts`, `/playwright.config.ts` | Plan §13 M9 | neu (F4.2): M9 in Chromium — 17 Beispiele, je 30 s mit Eingaben, ergeben Zeichen für Zeichen denselben Mitschnitt wie in Node; der Raum in einem echten Worker | +| `src/room.ts` („gestartet von“), `test/room.test.ts` | Plan §6.2 | erweitert (F5.3a): ein Client trägt auf F einen Namen; wer die laufende Simulation gestartet hat, steht im Zustand (`startedBy`), bis sie hält oder der Kern scheitert; ein Raum der eigenen Seite nennt niemanden | ## packages/editor @@ -268,6 +270,8 @@ Swift-Pfade relativ zu `filius-ipad/ios/FiliusPad/TopologyEditor/` (Stand | `src/sessions.ts`, `src/auth.ts` | Scratch-Editor `dashboard_auth_routes.py`, `models/session.py` | neu (F5.2): `/__auth` tauscht die Karte gegen einen Keks (`filius_session`, HttpOnly, SameSite=Lax, über HTTPS `Secure`, 12 h) und schickt den Browser nach `next` — nur an einen Ort von F selbst, aufgelöst wie im Browser: `/\…` und `//…` führen nicht weg (Scratchs Prüfung lässt beide durch, Werkzeug reicht sie roh weiter, der Browser liest `//…` — gemessen), und weil F den aufgelösten Pfad weitergibt, auch nicht `/.//…`, das die Auflösung zu `//…` kürzt; F speichert nur den SHA-256 des Schlüssels; ein zweites Projekt im selben Browser hängt eine Berechtigung an die Sitzung, eine Karte einer anderen Person beendet die alte Sitzung; jede Karte verlängert, abgelaufene werden gelöscht | | `src/sync.ts` (Anmeldung), `src/server.ts` (Wege, Herkunft) | Plan §8.5 | geändert (F5.2): `/sync` entscheidet nach der Berechtigung der Sitzung, die der Keks nennt — keine, Lesen, Schreiben; ein eingefrorenes Projekt lesen alle; jede Absage mit Grund und im Protokoll; ein WebSocket von einer Seite fremder Herkunft wird abgewiesen (eine Seite anderswo unter derselben Domain brächte den Keks mit, SameSite trennt nur Sites); unerwartete Fehler eines Wegs werden 500 und stehen im Protokoll | | `src/service.ts`, `src/store.ts` (`freeze`, `remove`), `src/persistence.ts` (`discard`), `src/main.ts` | Dashboard `backend/modules/scratch/src/service.py`, Scratch-Editor `revoke`, `active_sessions` | neu (F5.2): `/svc/*` mit Dienstschlüssel (in konstanter Zeit verglichen): Projekt anlegen (leeres FILIUS-Projekt), löschen (erst nicht mehr gespeichert, dann weg, dann alle getrennt — wer wiederkommt, findet kein Projekt), einfrieren und auftauen (wer drin ist, liest bzw. schreibt wieder und erfährt es), Berechtigung einer Person an einem Projekt oder alle ihre Sitzungen beenden (wer verbunden ist, wird getrennt und beim Wiederkommen abgewiesen), die Sitzungen auflisten; ohne Geheimnisse startet der Dienst nicht | +| `src/simRooms.ts`, `src/simWorker.ts`, `src/simChannel.ts`, `src/server.ts` (`/sim`), `src/service.ts` (Räume) | Plan §6.1–§6.4, §8.2 | neu (F5.3a): die Simulationsräume auf F — einer je Projekt für alle darin, gehalten von wenigen `worker_threads` (der mit den wenigsten Räumen nimmt den nächsten), jeder Raum mit einer Kopie des Dokuments: Hocuspocus' direkte Verbindung hält das Dokument offen, solange der Raum lebt, jede Änderung geht in den Raum, was der Raum schreibt, kommt als seine Änderung zurück; `/sim?project=…` meldet an wie `/sync` (Keks, Berechtigung, Herkunft), Nachrichten vor der Anmeldung werden gehalten; ändern nur mit Schreibstufe; nach dem Letzten bleibt der Raum 60 s und hält dann, schreibt zurück und endet; Einfrieren hält eine laufende Simulation an (ihr Stand kommt noch ins Projekt), Löschen beendet den Raum ohne Zurückschreiben, Widerruf trennt auch im Raum; ein scheiternder Faden beendet seine Räume; beim Herunterfahren halten alle Räume und schreiben zurück; ein Socket ohne Antwort auf Pings wird getrennt; Nachrichten höchstens 1 MiB; `/healthz` zählt die Räume; der Build bündelt den Faden als `dist/simWorker.mjs` | +| `test/sim.test.ts`, `test/harness.ts` (Faden) | — | neu (F5.3a): echte Sockets, der echte Faden (esbuild bündelt ihn wie der Build), Beispiel internetwork_komplett: ein Start für alle mit Namen, Ping in einer Befehlszeile bei der anderen, Zurückschreiben ins Dokument und eine Datei aus dem Dokument im Raum, Lesestufe, Absagen, Nachlauf, Einfrieren, Löschen, Widerruf, Wahl des Fadens, scheiternder Faden, Herunterfahren, Ping | | `test/ticket.test.ts`, `test/tickets.ts`, `test/tickets.python.json` | `tools/ticket-oracle` | neu (F5.2): jede Karte des Python-Orakels liest Node wie Python (Nutzlast oder derselbe Grund der Ablehnung); in Node signierte Karten tragen dasselbe JSON zum selben Zeitpunkt (Nodes zlib packt anders, ungepackt byte-gleich); die Ansprüche | | `test/access.test.ts`, `test/harness.ts` | — | neu (F5.2): `/__auth` (Keks, `next`, schlechte Karten, unbekanntes Projekt, zweites Projekt, nächste Person, Verlängerung), Absagen an `/sync` mit Grund, Herkunft, Dienstweg (Schlüssel, Anlegen, Fehlerantworten, Löschen, Einfrieren, Widerruf einzeln und ganz, Sitzungsliste, 500) | | `test/sync.test.ts`, `test/database.ts` | — | neu (F5.1), seit F5.2 mit echter Anmeldung (Karte, Keks): echte Clients (Hocuspocus' Provider über WebSocket) gegen eine Wegwerf-Postgres in Docker: Wiederkehr nach Neustart, nie verdichtete Aktualisierungen nach hartem Halt, zwei Personen live, Verdichten (500, Ruhe, nichts ohne Änderung), Schließen mit Personen, die sofort wiederkommen, gelöschtes Projekt, doppeltes Anlegen, Hausmeister beim Laden und nach einer Zusammenführung, unbekanntes Projekt, Lesestufe, `/healthz`, Protokoll | diff --git a/README.md b/README.md index c953cf6..7cbf73d 100644 --- a/README.md +++ b/README.md @@ -31,7 +31,7 @@ die Herkunft je Paket. Lizenz dieses Repos: **GPL-3.0** (`LICENSE`). | `packages/protocol` | die eine Nachrichtenliste zwischen Editor und Simulationsraum (`/sim`) | | `packages/sim` | Simulationsraum (für Server F und den Web Worker) und `LocalSimHost` | | `packages/editor` | Oberfläche im Iframe des Dashboards (React, Vite): Fläche, Palette, Werkzeuge des Entwurfs, Geräte-Blatt, Aktionsmodus (Desktops, Anwendungen, Datenaustausch), Dokumentationsmodus (Text, Gliederungsfläche, Bild exportieren, Bericht als PDF); `npm run dev -w @mdg-filius/editor` öffnet die Entwicklungsseite | -| `packages/server` | Filius-Dienst auf Server F: gemeinsame Dokumente über Hocuspocus (`/sync`), gespeichert in Postgres (Stand und Aktualisierungen, verdichtet), Hausmeister; Anmeldung mit der Eintrittskarte des Dashboards (`/__auth`, Keks, Lese- und Schreibstufe); Dienstweg des Dashboards (`/svc/*`: Projekte anlegen, löschen, einfrieren, Zugänge widerrufen); `/healthz`; `npm run build -w @mdg-filius/server` bündelt ihn nach `dist/main.mjs` *(Simulationsräume folgen)* | +| `packages/server` | Filius-Dienst auf Server F: gemeinsame Dokumente über Hocuspocus (`/sync`), gespeichert in Postgres (Stand und Aktualisierungen, verdichtet), Hausmeister; Anmeldung mit der Eintrittskarte des Dashboards (`/__auth`, Keks, Lese- und Schreibstufe); Dienstweg des Dashboards (`/svc/*`: Projekte anlegen, löschen, einfrieren, Zugänge widerrufen); Simulationsräume auf `/sim` in `worker_threads`, einer je Projekt für alle; `/healthz`; `npm run build -w @mdg-filius/server` bündelt ihn nach `dist/main.mjs` und den Faden der Räume nach `dist/simWorker.mjs` | Plan und Entscheidungen: `knowledge/plans/FILIUS_WEB_PLAN.md` im Dashboard. diff --git a/packages/protocol/src/sim.ts b/packages/protocol/src/sim.ts index fce67ce..99fe845 100644 --- a/packages/protocol/src/sim.ts +++ b/packages/protocol/src/sim.ts @@ -71,8 +71,16 @@ export interface DroppedUnit { readonly text: string; } -/** engineFailed: the core threw (a part it has not got yet, such as remote links); the room is back in design mode. */ -export const ERROR_CODES = ['badMessage', 'badVersion', 'unknownKind', 'badTopic', 'notRunning', 'unknownNode', 'unknownApp', 'unknownAction', 'badArgs', 'engineFailed'] as const; +/** + * engineFailed: the core threw (a part it has not got yet, such as remote + * links); the room is back in design mode. readOnly: the person may only + * watch (a read grant on F). frozen: the project is frozen (its deadline + * passed); its simulation does not start. + */ +export const ERROR_CODES = ['badMessage', 'badVersion', 'unknownKind', 'badTopic', 'notRunning', 'unknownNode', 'unknownApp', 'unknownAction', 'badArgs', 'engineFailed', 'readOnly', 'frozen'] as const; + +/** What a client sends that changes the room: only with a write grant (plan §6.4); a reader subscribes. */ +export const WRITING_KINDS: readonly ClientMessageKind[] = ['start', 'stop', 'speed', 'drop', 'clearCapture', 'input']; export type ErrorCode = (typeof ERROR_CODES)[number]; export type HostMessage = @@ -80,9 +88,11 @@ export type HostMessage = * The room's mode, speed and time; sent to every client on a change and * when it connects. `clock`: the time of day, in milliseconds since 1970, * that the simulation's time 0 stands for (a viewer writes its times of day - * from it, as Java's write the wall clock's). + * from it, as Java's write the wall clock's). `startedBy`: who started the + * simulation that runs, by name, on F (plan §6.2: „Simulation gestartet von + * Mia“); a room of one's own page names no one. */ - | { readonly v: 1; readonly kind: 'state'; readonly mode: SimMode; readonly speed: number; readonly t: number; readonly clock: number } + | { readonly v: 1; readonly kind: 'state'; readonly mode: SimMode; readonly speed: number; readonly t: number; readonly clock: number; readonly startedBy?: string } /** The cables that carried a frame since the last tick. */ | { readonly v: 1; readonly kind: 'activity'; readonly links: readonly string[] } /** A card's new frames; on subscribing, all it holds (`reset`). */ diff --git a/packages/protocol/src/sync.ts b/packages/protocol/src/sync.ts index 007e6c0..9ec63e5 100644 --- a/packages/protocol/src/sync.ts +++ b/packages/protocol/src/sync.ts @@ -1,7 +1,9 @@ // What F tells an editor about its place in a shared document (/sync, plan -// §5, §8.5) besides the document itself — the one list both sides import: -// why a connection is refused, why F lets it go, and the message that its -// access changed while it is connected (a project frozen or thawed). +// §5, §8.5) and in its simulation room (/sim, §6.4) besides what they carry +// — the one list both sides import: why a connection is refused, why F lets +// it go, and the message that its access changed while it is connected (a +// project frozen or thawed). /sim says both as a WebSocket's closing: the +// code tells which, the reason is the word. /** Why F refuses a connection to a project (Hocuspocus' permission-denied, with this as its reason). */ export const SYNC_REFUSALS = [ @@ -25,6 +27,13 @@ export const SYNC_CLOSINGS = [ ] as const; export type SyncClosing = (typeof SYNC_CLOSINGS)[number]; +/** /sim closed with one of SYNC_REFUSALS as its reason. */ +export const SIM_REFUSED_CODE = 4403; +/** /sim closed with one of SYNC_CLOSINGS as its reason. */ +export const SIM_CLOSED_CODE = 4410; +/** /sim closed because the room's thread failed: whoever comes back finds a new room, in design mode. */ +export const SIM_ROOM_FAILED_CODE = 4500; + export const SYNC_PROTOCOL_VERSION = 1; /** A stateless message from F: the connection's access now (sent when it changes on a connection that stays). */ diff --git a/packages/server/package.json b/packages/server/package.json index a9b5417..5ab0f11 100644 --- a/packages/server/package.json +++ b/packages/server/package.json @@ -8,7 +8,7 @@ ".": "./src/index.ts" }, "scripts": { - "build": "esbuild src/main.ts --bundle --platform=node --format=esm --target=node20 --outfile=dist/main.mjs --external:pg-native --banner:js=\"import { createRequire } from 'node:module'; const require = createRequire(import.meta.url);\"", + "build": "esbuild src/main.ts src/simWorker.ts --bundle --platform=node --format=esm --target=node20 --outdir=dist --out-extension:.js=.mjs --external:pg-native --banner:js=\"import { createRequire } from 'node:module'; const require = createRequire(import.meta.url);\"", "start": "node dist/main.mjs" }, "dependencies": { diff --git a/packages/server/src/index.ts b/packages/server/src/index.ts index d322a86..2bf50f8 100644 --- a/packages/server/src/index.ts +++ b/packages/server/src/index.ts @@ -1,11 +1,13 @@ -// Server F (plan §8): the shared documents, the sign-in and the dashboard's -// service; later the simulation rooms. +// Server F (plan §8): the shared documents, the simulation rooms, the +// sign-in and the dashboard's service. export * from './auth.js'; export * from './log.js'; export * from './persistence.js'; export * from './server.js'; export * from './service.js'; export * from './sessions.js'; +export * from './simChannel.js'; +export * from './simRooms.js'; export * from './store.js'; export * from './sync.js'; export * from './ticket.js'; diff --git a/packages/server/src/main.ts b/packages/server/src/main.ts index cc7360d..3a6c7ec 100644 --- a/packages/server/src/main.ts +++ b/packages/server/src/main.ts @@ -4,7 +4,8 @@ // origins the editor is served from, separated by commas). Without one of // them F does not start: an empty secret would let anyone sign. The cookie // goes over HTTPS only, unless FILIUS_INSECURE_COOKIE=1 (a developer's -// machine without HTTPS). +// machine without HTTPS). The rooms' thread is the bundle beside this one +// (dist/simWorker.mjs). import { jsonLog } from './log.js'; import { startServer } from './server.js'; @@ -33,6 +34,7 @@ const server = await startServer({ serviceKey: config.FILIUS_SERVICE_KEY!, origins: config.FILIUS_ORIGINS!.split(',').map((origin) => origin.trim()), secureCookie, + workerUrl: new URL('./simWorker.mjs', import.meta.url), }); for (const signal of ['SIGTERM', 'SIGINT'] as const) { process.once(signal, () => { diff --git a/packages/server/src/server.ts b/packages/server/src/server.ts index 39bb2b1..4ec5907 100644 --- a/packages/server/src/server.ts +++ b/packages/server/src/server.ts @@ -1,24 +1,37 @@ // Server F's front door (plan §8.2): one HTTP server — /healthz for the // proxy, /__auth for the editor's frame (ticket in, cookie out), /svc/* for -// the dashboard (with the service key), the WebSocket /sync for the shared -// documents; anything else is not found. A WebSocket from a page of another +// the dashboard (with the service key), the WebSockets /sync for the shared +// documents and /sim?project=… for the simulation rooms (the session's grant +// decides there as at /sync; a refusal closes the socket with its reason); +// anything else is not found. A WebSocket from a page of another // origin is refused (a page elsewhere under the same domain would bring the // person's cookie along: SameSite only tells sites apart). Closed, it first // takes no one new (a client lets go and comes back at once, which would -// keep its document open), then lets everyone go, waits until every -// document is stored and unloaded, and stops. +// keep its document open), stops every room (what it left written back), +// then lets everyone go, waits until every document is stored and unloaded, +// and stops. A /sim socket that answers no ping for PING_MILLISECONDS is +// let go (the proxy keeps an idle socket an hour; a tablet that went away +// says nothing). import { createServer, type IncomingMessage, type Server, type ServerResponse } from 'node:http'; import type { AddressInfo, Socket } from 'node:net'; -import { WebSocketServer } from 'ws'; +import { SIM_REFUSED_CODE, type SyncRefusal } from '@mdg-filius/protocol'; +import { WebSocketServer, type WebSocket } from 'ws'; import { exchangeTicket } from './auth.js'; import { errorText, type Log } from './log.js'; import { handleService } from './service.js'; -import { createSync, type Sync, type SyncOptions } from './sync.js'; +import { holdMessages, SimRooms } from './simRooms.js'; +import { cookieValue, createSync, SESSION_COOKIE, type Sync, type SyncOptions } from './sync.js'; +import { isUuid } from './ticket.js'; + +export const PING_MILLISECONDS = 30_000; +/** A message to a room at most. */ +export const SIM_MESSAGE_BYTES = 1024 * 1024; export interface FiliusServer { readonly port: number; readonly sync: Sync; + readonly sims: SimRooms; close(): Promise; } @@ -31,6 +44,11 @@ export interface ServerOptions extends SyncOptions { /** The origins the editor is served from (FILIUS_ORIGINS): a WebSocket from another is refused. */ readonly origins: readonly string[]; readonly secureCookie: boolean; + /** The bundled room thread (dist/simWorker.mjs). */ + readonly workerUrl: URL; + readonly simThreads?: number; + readonly lingerMilliseconds?: number; + readonly pingMilliseconds?: number; } function refuse(socket: Socket, status: string): void { @@ -45,12 +63,48 @@ function json(response: ServerResponse, status: number, value: object): void { export async function startServer(options: ServerOptions): Promise { const sync = createSync(options); const { hocuspocus } = sync; + const sims = new SimRooms({ + hocuspocus, + log: options.log, + workerUrl: options.workerUrl, + ...(options.simThreads === undefined ? {} : { threads: options.simThreads }), + ...(options.lingerMilliseconds === undefined ? {} : { lingerMilliseconds: options.lingerMilliseconds }), + }); const sockets = new WebSocketServer({ noServer: true }); + const simSockets = new WebSocketServer({ noServer: true, maxPayload: SIM_MESSAGE_BYTES }); + const alive = new WeakSet(); + const pings = setInterval(() => { + for (const socket of simSockets.clients) { + if (!alive.delete(socket)) socket.terminate(); + else socket.ping(); + } + }, options.pingMilliseconds ?? PING_MILLISECONDS); let closing = false; + const admit = async (socket: WebSocket, request: IncomingMessage) => { + alive.add(socket); + socket.on('pong', () => alive.add(socket)); + const held = holdMessages(socket); + const refuse = (reason: SyncRefusal, user?: string) => { + options.log.info('sim.refused', { reason, ...(user === undefined ? {} : { user }) }); + held.release(); + socket.close(SIM_REFUSED_CODE, reason); + }; + const project = new URL(request.url ?? '/', 'http://filius').searchParams.get('project'); + if (!isUuid(project)) return refuse('notAProject'); + const key = cookieValue(request.headers.cookie, SESSION_COOKIE); + const access = key === undefined ? ({ kind: 'notSignedIn' } as const) : await options.sessions.access(key, project); + if (access.kind !== 'granted') return refuse(access.kind, access.kind === 'notSignedIn' ? undefined : access.person.login); + try { + await sims.join(socket, project, { user: access.person.user, login: access.person.login, name: access.person.name, write: access.level === 'write' }, access.frozen, held); + } catch (error) { + options.log.warn('sim.openFailed', { project, error: errorText(error) }); + refuse('unknownProject', access.person.login); + } + }; const handle = async (request: IncomingMessage, response: ServerResponse, path: string) => { - if (request.method === 'GET' && path === '/healthz') json(response, 200, { status: 'ok', documents: hocuspocus.getDocumentsCount(), connections: hocuspocus.getConnectionsCount() }); + if (request.method === 'GET' && path === '/healthz') json(response, 200, { status: 'ok', documents: hocuspocus.getDocumentsCount(), connections: hocuspocus.getConnectionsCount(), rooms: sims.stats().rooms }); else if (request.method === 'GET' && path === '/__auth') await exchangeTicket(request, response, options); - else if (path.startsWith('/svc/')) await handleService(request, response, path, { ...options, sync }); + else if (path.startsWith('/svc/')) await handleService(request, response, path, { ...options, sync, sims }); else json(response, 404, { error: 'notFound' }); }; const http: Server = createServer((request, response) => { @@ -63,7 +117,7 @@ export async function startServer(options: ServerOptions): Promise }); http.on('upgrade', (request, socket, head) => { const path = new URL(request.url ?? '/', 'http://filius').pathname; - if (path !== '/sync') { + if (path !== '/sync' && path !== '/sim') { refuse(socket as Socket, '404 Not Found'); return; } @@ -77,7 +131,14 @@ export async function startServer(options: ServerOptions): Promise refuse(socket as Socket, '503 Service Unavailable'); return; } - sockets.handleUpgrade(request, socket, head, (ws) => hocuspocus.handleConnection(ws, request)); + if (path === '/sim') + simSockets.handleUpgrade(request, socket, head, (ws) => + admit(ws, request).catch((error: unknown) => { + options.log.error('sim.admitFailed', { error: errorText(error) }); + ws.close(1011, 'internal'); + }), + ); + else sockets.handleUpgrade(request, socket, head, (ws) => hocuspocus.handleConnection(ws, request)); }); await new Promise((resolve) => http.listen(options.port ?? 0, options.host ?? '0.0.0.0', resolve)); const port = (http.address() as AddressInfo).port; @@ -85,8 +146,12 @@ export async function startServer(options: ServerOptions): Promise return { port, sync, + sims, async close() { closing = true; + clearInterval(pings); + await sims.close(); + simSockets.close(); hocuspocus.closeConnections(); // Hocuspocus' close is a message on the socket, after which a client signs in again on it: the sockets go too. for (const socket of sockets.clients) socket.close(1001, 'service stopping'); diff --git a/packages/server/src/service.ts b/packages/server/src/service.ts index ce2a9c3..56eb6a2 100644 --- a/packages/server/src/service.ts +++ b/packages/server/src/service.ts @@ -1,10 +1,12 @@ // The dashboard's way into F (plan §8.5): /svc/*, with the service key and // without a person — the dashboard has decided who may, F does it. Make a -// project (an empty FILIUS project), delete one (whoever is in it is let -// go), freeze or thaw one (whoever is in it reads from then on, or writes -// again with a write grant), end a person's grant to a project or all their -// sessions (whoever is connected is let go), and list who is signed in to -// what (the dashboard checks it against its shares). Snapshots, restoring, +// project (an empty FILIUS project), delete one (whoever is in it or in its +// room is let go, the room ends without writing back), freeze or thaw one +// (whoever is in it reads from then on, or writes again with a write grant; +// a running simulation stops, what it left written back), end a person's +// grant to a project or all their sessions (whoever is connected, to the +// document or the room, is let go), and list who is signed in to what (the +// dashboard checks it against its shares). Snapshots, restoring, // copying and the .fls come with the commits (F6). import { createHash, timingSafeEqual } from 'node:crypto'; @@ -14,6 +16,7 @@ import { emptyProject, writeProject } from '@mdg-filius/doc'; import type { SyncClosing } from '@mdg-filius/protocol'; import type { Log } from './log.js'; import type { SessionStore } from './sessions.js'; +import type { SimRooms } from './simRooms.js'; import type { DocStore } from './store.js'; import { accessMessage, type ConnectionContext, type Sync } from './sync.js'; import { isUuid } from './ticket.js'; @@ -28,6 +31,7 @@ export interface ServiceOptions { readonly store: DocStore; readonly sessions: SessionStore; readonly sync: Sync; + readonly sims: SimRooms; readonly log: Log; readonly serviceKey: string; } @@ -92,7 +96,7 @@ export async function handleService(request: IncomingMessage, response: ServerRe } } -async function route(request: IncomingMessage, path: string, { store, sessions, sync, log }: ServiceOptions): Promise<[number, object]> { +async function route(request: IncomingMessage, path: string, { store, sessions, sync, sims, log }: ServiceOptions): Promise<[number, object]> { const close = (connections: ReturnType, reason: SyncClosing) => { for (const connection of connections) connection.close({ code: FORBIDDEN, reason }); return connections.length; @@ -111,7 +115,7 @@ async function route(request: IncomingMessage, path: string, { store, sessions, // Not stored any more, then gone (the grants with it), then everyone let go: coming back, they find no project. await sync.discard(id); if (!(await store.remove(id))) throw new HttpError(404, 'unknownProject'); - const disconnected = close(sync.connections(id), 'deleted'); + const disconnected = (await sims.drop(id)) + close(sync.connections(id), 'deleted'); log.info('svc.deleted', { project: id, disconnected }); return [200, { deleted: true, disconnected }]; } @@ -120,6 +124,7 @@ async function route(request: IncomingMessage, path: string, { store, sessions, const frozen = (await body(request)).frozen; if (typeof frozen !== 'boolean') throw new HttpError(400, 'frozenNotABoolean'); if (!(await store.freeze(id, frozen))) throw new HttpError(404, 'unknownProject'); + await sims.freeze(id, frozen); for (const connection of sync.connections(id)) { const readOnly = frozen || contextOf(connection).level === 'read'; if (connection.readOnly === readOnly) continue; @@ -135,10 +140,12 @@ async function route(request: IncomingMessage, path: string, { store, sessions, const only = given.project === undefined ? undefined : uuid(given.project, 'notAProject'); // The grant first: whoever is let go and comes back at once is refused. const revoked = await sessions.revoke(user, only); - const disconnected = close( - sync.connections(only).filter((connection) => contextOf(connection).person.user === user), - 'revoked', - ); + const disconnected = + sims.disconnect(user, only) + + close( + sync.connections(only).filter((connection) => contextOf(connection).person.user === user), + 'revoked', + ); log.info('svc.revoked', { user, ...(only === undefined ? {} : { project: only }), revoked, disconnected }); return [200, { revoked, disconnected }]; } diff --git a/packages/server/src/simChannel.ts b/packages/server/src/simChannel.ts new file mode 100644 index 0000000..3a0a43d --- /dev/null +++ b/packages/server/src/simChannel.ts @@ -0,0 +1,28 @@ +// The messages between server F's main thread and a thread that holds +// simulation rooms (plan §8.2: rooms in worker_threads, so that a large +// network does not hold up the shared documents). A thread holds several +// rooms, each with a replica of its project's document: the document's +// updates go both ways, a person's messages go in, the room's come out, +// each for one person (a room tells its people apart; the main thread +// numbers them). + +import type { HostMessage } from '@mdg-filius/protocol'; + +export type ToRoomThread = + /** A room for a project, its document as it is now. */ + | { readonly type: 'open'; readonly room: string; readonly state: Uint8Array } + | { readonly type: 'doc'; readonly room: string; readonly update: Uint8Array } + /** A person in the room, by name. */ + | { readonly type: 'connect'; readonly room: string; readonly client: number; readonly name: string } + | { readonly type: 'client'; readonly room: string; readonly client: number; readonly message: unknown } + | { readonly type: 'disconnect'; readonly room: string; readonly client: number } + /** The simulation stopped (what it left written back first); answered by `stopped`. */ + | { readonly type: 'stop'; readonly room: string } + /** The room ended — stopped first with `writeBack`, else as it is (its project is gone); answered by `closed`. */ + | { readonly type: 'close'; readonly room: string; readonly writeBack: boolean }; + +export type FromRoomThread = + | { readonly type: 'doc'; readonly room: string; readonly update: Uint8Array } + | { readonly type: 'host'; readonly room: string; readonly client: number; readonly message: HostMessage } + | { readonly type: 'stopped'; readonly room: string } + | { readonly type: 'closed'; readonly room: string }; diff --git a/packages/server/src/simRooms.ts b/packages/server/src/simRooms.ts new file mode 100644 index 0000000..e604fbf --- /dev/null +++ b/packages/server/src/simRooms.ts @@ -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; +} + +interface Room { + readonly project: string; + readonly thread: Thread; + readonly direct: DirectConnection; + readonly doc: Y.Doc; + readonly clients: Map; + 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 | 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>(); + private readonly ending = new Map>(); + 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 { + // 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 { + 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 { + 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, '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): Promise { + 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 { + 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 { + 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 { + 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 { + 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 { + 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())); + } +} diff --git a/packages/server/src/simWorker.ts b/packages/server/src/simWorker.ts new file mode 100644 index 0000000..ed87db1 --- /dev/null +++ b/packages/server/src/simWorker.ts @@ -0,0 +1,73 @@ +// A thread of server F that holds simulation rooms (simChannel.ts): each +// room a replica of its project's document and the room that reads it, all +// ticking together every TICK_MILLISECONDS. Bundled on its own +// (dist/simWorker.mjs), started by the main thread. + +import { parentPort } from 'node:worker_threads'; +import * as Y from 'yjs'; +import { SimulationRoom, TICK_MILLISECONDS } from '@mdg-filius/sim'; +import type { FromRoomThread, ToRoomThread } from './simChannel.js'; + +/** The origin of what came from the main thread: not sent back. */ +const MAIN_ORIGIN = 'main'; + +interface HeldRoom { + readonly doc: Y.Doc; + readonly room: SimulationRoom; + /** The main thread's number of a person → the room's. */ + readonly clients: Map; +} + +const port = parentPort!; +const rooms = new Map(); +const post = (message: FromRoomThread) => port.postMessage(message); + +port.on('message', (message: ToRoomThread) => { + if (message.type === 'open') { + const doc = new Y.Doc(); + Y.applyUpdate(doc, message.state, MAIN_ORIGIN); + doc.on('update', (update: Uint8Array, origin: unknown) => { + if (origin !== MAIN_ORIGIN && rooms.get(message.room)?.doc === doc) post({ type: 'doc', room: message.room, update }); + }); + const room = new SimulationRoom(doc, { now: () => performance.now(), seed: () => Math.floor(Math.random() * 2 ** 31) }); + rooms.set(message.room, { doc, room, clients: new Map() }); + return; + } + const held = rooms.get(message.room); + if (held === undefined) return; + switch (message.type) { + case 'doc': + Y.applyUpdate(held.doc, message.update, MAIN_ORIGIN); + return; + case 'connect': + held.clients.set( + message.client, + held.room.connect((output) => post({ type: 'host', room: message.room, client: message.client, message: output }), message.name), + ); + return; + case 'client': { + const client = held.clients.get(message.client); + if (client !== undefined) held.room.receive(client, message.message); + return; + } + case 'disconnect': { + const client = held.clients.get(message.client); + held.clients.delete(message.client); + if (client !== undefined) held.room.disconnect(client); + return; + } + case 'stop': + held.room.stop(); + post({ type: 'stopped', room: message.room }); + return; + case 'close': + if (message.writeBack) held.room.stop(); + rooms.delete(message.room); + post({ type: 'closed', room: message.room }); + return; + } +}); + +setInterval(() => { + for (const held of rooms.values()) held.room.tick(); +}, TICK_MILLISECONDS); diff --git a/packages/server/test/harness.ts b/packages/server/test/harness.ts index 259cd8b..a2819cf 100644 --- a/packages/server/test/harness.ts +++ b/packages/server/test/harness.ts @@ -2,9 +2,15 @@ // test secrets, a project as the service makes it, a person signed in with // a ticket as the dashboard signs one, and that person in a project with // Hocuspocus' own provider over a real WebSocket (bringing the session's -// cookie, as the browser does). +// cookie, as the browser does). The rooms' thread is the real one, bundled +// by esbuild as the build bundles it. import { randomUUID } from 'node:crypto'; +import { mkdtempSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join as joinPath } from 'node:path'; +import { pathToFileURL } from 'node:url'; +import { build } from 'esbuild'; import { HocuspocusProvider, HocuspocusProviderWebsocket } from '@hocuspocus/provider'; import { expect } from 'vitest'; import WebSocket from 'ws'; @@ -32,6 +38,18 @@ export function memoryLog(): MemoryLog { export type TestServer = FiliusServer & { readonly log: MemoryLog; readonly base: string }; +let roomThread: Promise | undefined; + +/** src/simWorker.ts bundled into a directory of its own, once a test file. */ +export function bundledRoomThread(): Promise { + roomThread ??= (async () => { + const outfile = joinPath(mkdtempSync(joinPath(tmpdir(), 'filius-room-thread-')), 'simWorker.mjs'); + await build({ entryPoints: [new URL('../src/simWorker.ts', import.meta.url).pathname], bundle: true, platform: 'node', format: 'esm', target: 'node20', outfile, logLevel: 'error' }); + return pathToFileURL(outfile); + })(); + return roomThread; +} + export async function startTestServer(store: DocStore, sessions: SessionStore, options: Partial = {}): Promise { const log = memoryLog(); const server = await startServer({ @@ -44,6 +62,8 @@ export async function startTestServer(store: DocStore, sessions: SessionStore, o serviceKey: SERVICE_KEY, origins: [EDITOR_ORIGIN], secureCookie: false, + workerUrl: await bundledRoomThread(), + simThreads: 1, ...options, }); return Object.assign(server, { log, base: `http://127.0.0.1:${server.port}` }); @@ -61,14 +81,16 @@ export async function newProject(store: DocStore): Promise { export interface Person { readonly user?: string; readonly login?: string; + /** The name the dashboard shows. */ + readonly name?: string; readonly level?: 'read' | 'write'; /** The cookie the browser brings to /__auth. */ readonly cookie?: string; } /** The answer of /__auth to a fresh ticket. */ -export function exchange(server: TestServer, project: string, { user = randomUUID(), login = 'ab12', level = 'write', cookie }: Person = {}, next = '/'): Promise { - const ticket = signTicket({ u: user, n: login, d: 'Ada Beispiel', c: '#3366cc', lang: 'de', p: project, lvl: level }, TICKET_SECRET); +export function exchange(server: TestServer, project: string, { user = randomUUID(), login = 'ab12', name = 'Ada Beispiel', level = 'write', cookie }: Person = {}, next = '/'): Promise { + const ticket = signTicket({ u: user, n: login, d: name, c: '#3366cc', lang: 'de', p: project, lvl: level }, TICKET_SECRET); return fetch(`${server.base}/__auth?ticket=${encodeURIComponent(ticket)}&next=${encodeURIComponent(next)}`, { redirect: 'manual', ...(cookie === undefined ? {} : { headers: { cookie } }), diff --git a/packages/server/test/sim.test.ts b/packages/server/test/sim.test.ts new file mode 100644 index 0000000..1a55b13 --- /dev/null +++ b/packages/server/test/sim.test.ts @@ -0,0 +1,389 @@ +// The simulation rooms on F (F5.3a) with real sockets, the real room thread +// and a real Postgres, on FILIUS' example internetwork_komplett: one room +// for everyone in a project — started by one, running for all, named after +// who started it; a ping in one person's command line seen by the other; +// what the simulation leaves going into the shared document; a reader who +// watches and changes nothing; the refusals, with their reasons; the room +// staying a while after the last person left, then stopping and ending; +// freezing, deleting and revoking reaching the room; a failed thread +// ending its rooms; the service stopping with rooms running. + +import { randomUUID } from 'node:crypto'; +import { readFileSync } from 'node:fs'; +import { afterAll, beforeAll, describe, expect, it } from 'vitest'; +import WebSocket from 'ws'; +import * as Y from 'yjs'; +import { changeFiles, nodeFiles, readProject, writeProject } from '@mdg-filius/doc'; +import { openProjectArchive } from '@mdg-filius/fls'; +import { SIM_CLOSED_CODE, SIM_REFUSED_CODE, SIM_ROOM_FAILED_CODE, WRITING_KINDS, type HostMessage } from '@mdg-filius/protocol'; +import { LINGER_MILLISECONDS, PostgresSessions, PostgresStore } from '../src/index.js'; +import { testDatabase, type TestDatabase } from './database.js'; +import { EDITOR_ORIGIN, join, SERVICE_KEY, signIn, startTestServer, until, type TestServer } from './harness.js'; + +let database: TestDatabase; +let store: PostgresStore; +let sessions: PostgresSessions; + +beforeAll(async () => { + database = await testDatabase(); + store = PostgresStore.connect(database.url); + sessions = new PostgresSessions(store.pool); + await store.migrate(); + await sessions.migrate(); +}, 60_000); + +afterAll(async () => { + await store?.close(); + await database?.stop(); +}); + +const EXAMPLE = new URL('../../../corpus/java-examples/internetwork_komplett.fls', import.meta.url); + +/** FILIUS' example as a project of F; the id of a device by its name. */ +async function exampleProject(): Promise<{ readonly project: string; readonly node: (name: string) => string }> { + const project = randomUUID(); + const imported = openProjectArchive(new Uint8Array(readFileSync(EXAMPLE)), () => randomUUID()); + const doc = new Y.Doc(); + writeProject(doc, imported); + expect(await store.create(project, Y.encodeStateAsUpdate(doc))).toBe(true); + return { project, node: (name) => imported.graph.nodes.find((each) => each.displayName === name)!.id }; +} + +const start = (options: { readonly linger?: number } = {}) => startTestServer(store, sessions, options.linger === undefined ? {} : { lingerMilliseconds: options.linger }); + +interface SimClient { + readonly socket: WebSocket; + readonly got: HostMessage[]; + /** The socket's closing: its code and reason. */ + readonly closed: Promise<{ readonly code: number; readonly reason: string }>; + send(message: object): void; + /** The last state the room sent. */ + state(): Extract | undefined; +} + +function enterRoom(server: TestServer, project: string, { cookie, origin }: { readonly cookie?: string; readonly origin?: string } = {}): Promise { + const socket = new WebSocket(`ws://127.0.0.1:${server.port}/sim?project=${project}`, { headers: { ...(cookie === undefined ? {} : { cookie }), ...(origin === undefined ? {} : { origin }) } }); + const got: HostMessage[] = []; + socket.on('message', (data) => got.push(JSON.parse(data.toString()) as HostMessage)); + const closed = new Promise<{ code: number; reason: string }>((resolve) => socket.on('close', (code, reason) => resolve({ code, reason: reason.toString() }))); + return new Promise((resolve, reject) => { + socket.on('open', () => + resolve({ + socket, + got, + closed, + send: (message) => socket.send(JSON.stringify({ v: 1, ...message })), + state: () => got.filter((message) => message.kind === 'state').at(-1) as Extract | undefined, + }), + ); + socket.on('unexpected-response', (_request, response) => reject(new Error(`refused: ${response.statusCode}`))); + socket.on('error', reject); + }); +} + +/** Signed in with a fresh ticket (by name), then in the room. */ +async function person( + server: TestServer, + project: string, + name: string, + level: 'read' | 'write' = 'write', + user = randomUUID(), +): Promise { + const cookie = await signIn(server, project, { user, login: name.toLowerCase(), level, name }); + return Object.assign(await enterRoom(server, project, { cookie }), { cookie, user }); +} + +const appLines = (client: SimClient) => + client.got.filter((message): message is Extract => message.kind === 'app').flatMap((message) => message.lines as { text?: string }[]); +const sleep = (milliseconds: number) => new Promise((resolve) => setTimeout(resolve, milliseconds)); +function service(server: TestServer, method: string, path: string, body?: unknown): Promise { + return fetch(`${server.base}${path}`, { method, headers: { 'x-filius-service-key': SERVICE_KEY }, ...(body === undefined ? {} : { body: JSON.stringify(body) }) }); +} + +describe('a simulation room on F', () => { + it('runs for everyone in the project once one starts it, names who did, and shows a ping in one command line to the other', async () => { + const server = await start(); + const { project, node } = await exampleProject(); + const mia = await person(server, project, 'Mia'); + const ben = await person(server, project, 'Ben'); + await until(() => mia.state() !== undefined && ben.state() !== undefined); + expect(ben.state()).toMatchObject({ mode: 'design' }); + const notebook = node('Notebook 1.10'); + ben.send({ kind: 'sub', topics: [`app:${notebook}/terminal`] }); + mia.send({ kind: 'start' }); + await until(() => ben.state()?.mode === 'simulation'); + expect(ben.state()!.startedBy).toBe('Mia'); + mia.send({ kind: 'input', node: notebook, app: 'terminal', action: 'enter', args: { line: 'ping 141.99.5.10' }, reqId: 'r1' }); + await until(() => mia.got.some((message) => message.kind === 'ack' && message.reqId === 'r1')); + await until(() => appLines(ben).some((line) => /4 Paket\(e\) gesendet, 4 Paket\(e\) empfangen/.test(line.text ?? '')), 20_000); + ben.send({ kind: 'stop' }); + await until(() => mia.state()?.mode === 'design'); + expect(mia.state()).not.toHaveProperty('startedBy'); + expect(server.log.lines.filter((line) => ['sim.start', 'sim.stop'].includes(line.event as string)).map(({ event, user }) => [event, user])).toEqual([ + ['sim.start', 'mia'], + ['sim.stop', 'ben'], + ]); + expect(server.sims.stats()).toEqual({ rooms: 1, people: 2, threads: [1] }); + mia.socket.close(); + ben.socket.close(); + await server.close(); + }, 30_000); + + it('writes what the simulation leaves into the shared document, for whoever has it open', async () => { + const server = await start(); + const { project, node } = await exampleProject(); + const mia = await person(server, project, 'Mia'); + const editor = await join(server, project, { cookie: mia.cookie }); + const notebook = node('Notebook 1.10'); + mia.send({ kind: 'start' }); + await until(() => mia.state()?.mode === 'simulation'); + mia.send({ kind: 'input', node: notebook, app: 'terminal', action: 'enter', args: { line: 'touch probe.txt' }, reqId: 'r1' }); + await until(() => (nodeFiles(editor.doc, notebook)?.allEntries() ?? []).some((entry) => entry.path.endsWith('probe.txt'))); + // And back: a file a person writes into the document is there for the command line. + mia.send({ kind: 'sub', topics: [`app:${notebook}/terminal`] }); + changeFiles(editor.doc, notebook, 'person', (files) => files.writeTextFile('/from-doc.txt', 'x')); + await until(() => editor.provider.unsyncedChanges === 0); + mia.send({ kind: 'input', node: notebook, app: 'terminal', action: 'enter', args: { line: 'ls' }, reqId: 'r2' }); + await until(() => appLines(mia).some((line) => (line.text ?? '').includes('from-doc.txt'))); + mia.socket.close(); + editor.close(); + await server.close(); + }); + + it('lets a reader watch and change nothing', async () => { + const server = await start(); + const { project, node } = await exampleProject(); + const reader = await person(server, project, 'Rita', 'read'); + await until(() => reader.state() !== undefined); + for (const kind of WRITING_KINDS) reader.send({ kind, ...(kind === 'input' ? { node: node('Notebook 1.10'), app: 'terminal', action: 'enter', args: { line: 'ls' }, reqId: 'r' } : {}) }); + await until(() => reader.got.filter((message) => message.kind === 'error').length === WRITING_KINDS.length); + expect(reader.got.filter((message) => message.kind === 'error')).toEqual([ + ...WRITING_KINDS.slice(0, -1).map(() => ({ v: 1, kind: 'error', code: 'readOnly' })), + { v: 1, kind: 'error', code: 'readOnly', reqId: 'r' }, + ]); + expect(reader.state()).toMatchObject({ mode: 'design' }); + reader.send({ kind: 'sub', topics: ['activity'] }); + reader.socket.send('not json'); + await until(() => reader.got.some((message) => message.kind === 'error' && message.code === 'badMessage')); + expect(server.log.lines.find((line) => line.event === 'sim.joined')).toMatchObject({ project, user: 'rita', write: false }); + reader.socket.close(); + await server.close(); + }); + + it('refuses without a session, without a grant, a name that is no project, and a page of another origin', async () => { + const server = await start(); + const { project } = await exampleProject(); + const other = (await exampleProject()).project; + const cookie = await signIn(server, other, { login: 'ab12' }); + expect(await (await enterRoom(server, project)).closed).toEqual({ code: SIM_REFUSED_CODE, reason: 'notSignedIn' }); + expect(await (await enterRoom(server, project, { cookie })).closed).toEqual({ code: SIM_REFUSED_CODE, reason: 'noAccess' }); + expect(await (await enterRoom(server, 'nothing', { cookie })).closed).toEqual({ code: SIM_REFUSED_CODE, reason: 'notAProject' }); + await expect(enterRoom(server, project, { cookie, origin: 'https://elsewhere.test' })).rejects.toThrow('refused: 403'); + expect(server.log.lines.filter((line) => line.event === 'sim.refused').map(({ reason, user }) => [reason, user])).toEqual([ + ['notSignedIn', undefined], + ['noAccess', 'ab12'], + ['notAProject', undefined], + ]); + await server.close(); + }); + + it('refuses a project its store no longer has when the room is made', async () => { + const server = await start(); + const { project } = await exampleProject(); + const mia = await person(server, project, 'Mia'); + await until(() => mia.state() !== undefined); + mia.socket.close(); + await server.close(); + const forgetful = Object.assign(Object.create(store) as PostgresStore, { load: async () => undefined }); + const again = await startTestServer(forgetful, sessions); + expect(await (await enterRoom(again, project, { cookie: mia.cookie })).closed).toEqual({ code: SIM_REFUSED_CODE, reason: 'unknownProject' }); + expect(again.log.lines.find((line) => line.event === 'sim.openFailed')).toMatchObject({ project }); + await again.close(); + }); + + it(`stays ${LINGER_MILLISECONDS / 1000} s after the last person left, running, then stops, writes back and ends`, async () => { + expect(LINGER_MILLISECONDS).toBe(60_000); + const server = await start({ linger: 600 }); + const { project, node } = await exampleProject(); + const mia = await person(server, project, 'Mia'); + mia.send({ kind: 'start' }); + await until(() => mia.state()?.mode === 'simulation'); + mia.socket.close(); + await sleep(200); + // Back within the while: the same room, still running. + const again = await enterRoom(server, project, { cookie: mia.cookie }); + await until(() => again.state() !== undefined); + expect(again.state()).toMatchObject({ mode: 'simulation', startedBy: 'Mia' }); + // Staying past the while it had: coming back called it off. + await sleep(700); + expect(server.sims.stats().rooms).toBe(1); + again.send({ kind: 'input', node: node('Notebook 1.10'), app: 'terminal', action: 'enter', args: { line: 'touch left.txt' }, reqId: 'r' }); + await until(() => again.got.some((message) => message.kind === 'ack')); + again.socket.close(); + await until(() => server.sims.stats().rooms === 0, 5_000); + expect(server.log.lines.filter((line) => line.event === 'sim.roomClosed')).toEqual([{ level: 'info', event: 'sim.roomClosed', project, writeBack: true }]); + // The document let go once the room had written back. + await until(() => server.sync.hocuspocus.getDocumentsCount() === 0); + const stored = await store.load(project); + const doc = new Y.Doc(); + Y.applyUpdate(doc, stored!.state); + for (const update of stored!.updates) Y.applyUpdate(doc, update); + expect( + nodeFiles(doc, node('Notebook 1.10'))! + .allEntries() + .some((entry) => entry.path.endsWith('left.txt')), + ).toBe(true); + // Whoever comes later finds a new room, in design mode. + const later = await enterRoom(server, project, { cookie: mia.cookie }); + await until(() => later.state() !== undefined); + expect(later.state()).toMatchObject({ mode: 'design' }); + later.socket.close(); + await server.close(); + }); + + it('stops a running simulation when the project is frozen, refuses to start it then, and starts it once thawed', async () => { + const server = await start(); + const { project, node } = await exampleProject(); + const mia = await person(server, project, 'Mia'); + mia.send({ kind: 'start' }); + await until(() => mia.state()?.mode === 'simulation'); + mia.send({ kind: 'input', node: node('Notebook 1.10'), app: 'terminal', action: 'enter', args: { line: 'touch before.txt' }, reqId: 'r' }); + await until(() => mia.got.some((message) => message.kind === 'ack')); + expect((await service(server, 'POST', `/svc/projects/${project}/freeze`, { frozen: true })).status).toBe(200); + expect(mia.state()).toMatchObject({ mode: 'design' }); + mia.send({ kind: 'start', reqId: 'x' }); + await until(() => mia.got.some((message) => message.kind === 'error')); + expect(mia.got.filter((message) => message.kind === 'error')).toEqual([{ v: 1, kind: 'error', code: 'frozen', reqId: 'x' }]); + // A reader is told it reads, not that it is frozen. + const late = await person(server, project, 'Ben'); + late.send({ kind: 'start' }); + await until(() => late.got.some((message) => message.kind === 'error')); + expect(late.got.find((message) => message.kind === 'error')).toMatchObject({ code: 'frozen' }); + expect((await service(server, 'POST', `/svc/projects/${project}/freeze`, { frozen: false })).status).toBe(200); + mia.send({ kind: 'start' }); + await until(() => mia.state()?.mode === 'simulation'); + mia.socket.close(); + late.socket.close(); + await server.close(); + }); + + it('freezes a project whose room nobody made, and makes its room frozen', async () => { + const server = await start(); + const { project } = await exampleProject(); + expect((await service(server, 'POST', `/svc/projects/${project}/freeze`, { frozen: true })).status).toBe(200); + const mia = await person(server, project, 'Mia'); + mia.send({ kind: 'start' }); + await until(() => mia.got.some((message) => message.kind === 'error')); + expect(mia.got.find((message) => message.kind === 'error')).toMatchObject({ code: 'frozen' }); + mia.socket.close(); + await server.close(); + }); + + it('lets everyone in the room go when the project is deleted, and writes nothing back', async () => { + const server = await start(); + const { project } = await exampleProject(); + const mia = await person(server, project, 'Mia'); + mia.send({ kind: 'start' }); + await until(() => mia.state()?.mode === 'simulation'); + const response = await service(server, 'DELETE', `/svc/projects/${project}`); + expect(await response.json()).toEqual({ deleted: true, disconnected: 1 }); + expect(await mia.closed).toEqual({ code: SIM_CLOSED_CODE, reason: 'deleted' }); + expect(server.log.lines.find((line) => line.event === 'sim.roomClosed')).toMatchObject({ project, writeBack: false }); + await until(() => server.sync.hocuspocus.getDocumentsCount() === 0); + expect(server.log.lines.filter((line) => line.event === 'store.failed')).toEqual([]); + expect(await store.load(project)).toBeUndefined(); + await server.close(); + }); + + it('lets a person go whose grant ends, to one project or to all, and no one else', async () => { + const server = await start(); + const { project } = await exampleProject(); + const other = (await exampleProject()).project; + const user = randomUUID(); + const mia = await person(server, project, 'Mia', 'write', user); + const miaThere = await person(server, other, 'Mia', 'write', user); + const ben = await person(server, project, 'Ben'); + expect(await (await service(server, 'POST', '/svc/revoke', { user, project })).json()).toEqual({ revoked: 1, disconnected: 1 }); + expect(await mia.closed).toEqual({ code: SIM_CLOSED_CODE, reason: 'revoked' }); + expect(miaThere.socket.readyState).toBe(WebSocket.OPEN); + // Two sessions: one ticket each. + expect(await (await service(server, 'POST', '/svc/revoke', { user })).json()).toEqual({ revoked: 2, disconnected: 1 }); + expect(await miaThere.closed).toEqual({ code: SIM_CLOSED_CODE, reason: 'revoked' }); + expect(ben.socket.readyState).toBe(WebSocket.OPEN); + ben.socket.close(); + await server.close(); + }); + + it('ends the rooms of a thread that fails, and gives whoever comes back a new room', async () => { + const server = await start(); + const { project } = await exampleProject(); + const mia = await person(server, project, 'Mia'); + mia.send({ kind: 'start' }); + await until(() => mia.state()?.mode === 'simulation'); + await server.sims.workers[0]!.terminate(); + expect(await mia.closed).toEqual({ code: SIM_ROOM_FAILED_CODE, reason: 'roomFailed' }); + expect(server.log.lines.find((line) => line.event === 'sim.threadFailed')).toMatchObject({ rooms: 1 }); + const again = await enterRoom(server, project, { cookie: mia.cookie }); + await until(() => again.state() !== undefined); + expect(again.state()).toMatchObject({ mode: 'design' }); + expect(server.sims.stats()).toEqual({ rooms: 1, people: 1, threads: [1] }); + again.socket.close(); + await server.close(); + }); + + it('stops every room when the service stops, writing back what they left, and lets everyone go', async () => { + const server = await start(); + const { project, node } = await exampleProject(); + const mia = await person(server, project, 'Mia'); + mia.send({ kind: 'start' }); + await until(() => mia.state()?.mode === 'simulation'); + mia.send({ kind: 'input', node: node('Notebook 1.10'), app: 'terminal', action: 'enter', args: { line: 'touch last.txt' }, reqId: 'r' }); + await until(() => mia.got.some((message) => message.kind === 'ack')); + await server.close(); + expect(await mia.closed).toEqual({ code: 1001, reason: 'service stopping' }); + const stored = await store.load(project); + const doc = new Y.Doc(); + Y.applyUpdate(doc, stored!.state); + for (const update of stored!.updates) Y.applyUpdate(doc, update); + expect( + nodeFiles(doc, node('Notebook 1.10'))! + .allEntries() + .some((entry) => entry.path.endsWith('last.txt')), + ).toBe(true); + expect(readProject(doc).graph.nodes.length).toBeGreaterThan(0); + }); + + it('lets go a socket that answers no ping', async () => { + const server = await startTestServer(store, sessions, { pingMilliseconds: 100 }); + const { project } = await exampleProject(); + const mia = await person(server, project, 'Mia'); + await until(() => mia.state() !== undefined); + // A socket that went away says nothing: ws answers pings by itself, so its answering is stopped. + (mia.socket as unknown as { _receiver: { removeAllListeners(name: string): void } })._receiver.removeAllListeners('ping'); + expect((await mia.closed).code).toBe(1006); + await server.close(); + }); + + it('gives a new room to the thread with the fewest', async () => { + const server = await startTestServer(store, sessions, { simThreads: 2 }); + const [first, second, third] = [await exampleProject(), await exampleProject(), await exampleProject()]; + const people = [await person(server, first.project, 'Mia'), await person(server, second.project, 'Ben')]; + await until(() => people.every((each) => each.state() !== undefined)); + expect(server.sims.stats().threads).toEqual([1, 1]); + people.push(await person(server, third.project, 'Rita')); + await until(() => people[2]!.state() !== undefined); + expect([...server.sims.stats().threads].sort()).toEqual([1, 2]); + for (const each of people) each.socket.close(); + await server.close(); + }); + + it('opens a room for the editor with its origin', async () => { + const server = await start(); + const { project } = await exampleProject(); + const cookie = await signIn(server, project); + const client = await enterRoom(server, project, { cookie, origin: EDITOR_ORIGIN }); + await until(() => client.state() !== undefined); + client.socket.close(); + await server.close(); + }); +}); diff --git a/packages/server/test/sync.test.ts b/packages/server/test/sync.test.ts index fa2081b..72eb214 100644 --- a/packages/server/test/sync.test.ts +++ b/packages/server/test/sync.test.ts @@ -280,7 +280,7 @@ describe('server F keeps the shared documents', () => { it('answers /healthz, refuses other ways in, and logs JSON', async () => { const server = await start(); const health = await fetch(`http://127.0.0.1:${server.port}/healthz`); - expect([health.status, await health.json()]).toEqual([200, { status: 'ok', documents: 0, connections: 0 }]); + expect([health.status, await health.json()]).toEqual([200, { status: 'ok', documents: 0, connections: 0, rooms: 0 }]); expect((await fetch(`http://127.0.0.1:${server.port}/nothing`)).status).toBe(404); const refused = await new Promise((resolve) => { const socket = new WebSocket(`ws://127.0.0.1:${server.port}/elsewhere`); diff --git a/packages/sim/src/room.ts b/packages/sim/src/room.ts index 6192a9c..bf9eee1 100644 --- a/packages/sim/src/room.ts +++ b/packages/sim/src/room.ts @@ -109,6 +109,8 @@ export interface RoomOptions { interface Client { readonly send: (message: HostMessage) => void; + /** Who it is, on F; none in a room of one's own page. */ + readonly name: string | undefined; readonly topics: Set; /** Per card, the number of the last frame sent. */ readonly traceSent: Map; @@ -149,6 +151,8 @@ export class SimulationRoom { private readonly browserWindows = new Map(); /** The language of the last start (the pages the devices write themselves). */ private language: PageLanguage | undefined; + /** Who started the simulation that runs. */ + private startedBy: string | undefined; /** Each node's file system and its revision when it was last written back. */ private writtenFiles = new Map(); /** The wall clock's time of the last write-back. */ @@ -320,10 +324,10 @@ export class SimulationRoom { } } - /** A new client; it gets the room's state at once. */ - connect(send: (message: HostMessage) => void): number { + /** A new client, by name on F; it gets the room's state at once. */ + connect(send: (message: HostMessage) => void, name?: string): number { const id = this.nextClient++; - this.clients.set(id, { send, topics: new Set(), traceSent: new Map(), droppedSent: undefined, appSent: new Map(), nodeSent: new Map(), viewSent: new Map() }); + this.clients.set(id, { send, name, topics: new Set(), traceSent: new Map(), droppedSent: undefined, appSent: new Map(), nodeSent: new Map(), viewSent: new Map() }); send(this.state()); return id; } @@ -375,7 +379,7 @@ export class SimulationRoom { for (const topic of input.topics) client.topics.delete(topic); return; case 'start': - this.start(input.language); + this.start(input.language, client.name); return; case 'stop': this.stop(); @@ -422,14 +426,14 @@ export class SimulationRoom { private state(): HostMessage { const t = this.engine.state.currentTimeMilliseconds; - return message({ kind: 'state', mode: this.mode, speed: this.speed, t, clock: (this.options.epoch ?? Date.now)() - t }); + return message({ kind: 'state', mode: this.mode, speed: this.speed, t, clock: (this.options.epoch ?? Date.now)() - t, ...(this.startedBy === undefined ? {} : { startedBy: this.startedBy }) }); } private broadcast(output: HostMessage): void { for (const client of this.clients.values()) client.send(output); } - start(language?: PageLanguage): void { + start(language?: PageLanguage, startedBy?: string): void { if (this.mode === 'simulation') return; this.language = language; const snapshot = this.snapshot(); @@ -442,6 +446,7 @@ export class SimulationRoom { this.software = this.documentSoftware(); this.syncWindows(); this.mode = 'simulation'; + this.startedBy = startedBy; this.lastWall = this.options.now(); this.lastWriteBack = this.lastWall; this.broadcast(this.state()); @@ -455,6 +460,7 @@ export class SimulationRoom { this.engine.handle({ type: 'stop' }); this.dropping.clear(); this.mode = 'design'; + this.startedBy = undefined; this.lastWall = undefined; this.broadcast(this.state()); this.flush([...this.clients.values()], false); @@ -487,6 +493,7 @@ export class SimulationRoom { this.broadcast(message({ kind: 'error', code: 'engineFailed', detail: error instanceof Error ? error.message : String(error) })); if (this.engine.state.phase === 'running') this.engine.handle({ type: 'stop' }); this.mode = 'design'; + this.startedBy = undefined; this.lastWall = undefined; this.broadcast(this.state()); this.flush([...this.clients.values()], false); diff --git a/packages/sim/test/room.test.ts b/packages/sim/test/room.test.ts index 18eef8c..ae59d2b 100644 --- a/packages/sim/test/room.test.ts +++ b/packages/sim/test/room.test.ts @@ -79,6 +79,39 @@ describe('a simulation room', () => { } }); + it('names who started the simulation while it runs, and no one in a room of one page', () => { + const { room, got, send, run } = setUp(); + const mia: HostMessage[] = []; + const miaId = room.connect((message) => mia.push(message), 'Mia'); + room.receive(miaId, { v: 1, kind: 'start' }); + run(400); + const late: HostMessage[] = []; + room.connect((message) => late.push(message), 'Ben'); + expect(ofKind(late, 'state').map((state) => state.startedBy)).toEqual(['Mia']); + send({ kind: 'speed', value: 50 }); + send({ kind: 'stop' }); + expect(ofKind(mia, 'state').map(({ mode, startedBy }) => [mode, startedBy])).toEqual([ + ['design', undefined], + ['simulation', 'Mia'], + ['simulation', 'Mia'], + ['design', undefined], + ]); + // Started by the client without a name, the room names no one. + send({ kind: 'start' }); + expect(ofKind(got, 'state').at(-1)).not.toHaveProperty('startedBy'); + }); + + it('forgets who started it when the core fails', () => { + const { room, run, node } = setUp(exampleDoc('java-examples/modem_beispiel.fls')); + const mia: HostMessage[] = []; + const miaId = room.connect((message) => mia.push(message), 'Mia'); + room.receive(miaId, { v: 1, kind: 'start' }); + room.receive(miaId, { v: 1, kind: 'input', node: node('Notebook 0.10').id, app: 'terminal', action: 'enter', args: { line: 'ping 192.168.0.11' }, reqId: 'r' }); + run(10_000); + expect(ofKind(mia, 'state').at(-1)).toMatchObject({ mode: 'design' }); + expect(ofKind(mia, 'state').at(-1)).not.toHaveProperty('startedBy'); + }); + it("sends a card's frames and a terminal's lines as the engine has them", () => { const { got, send, node, run } = setUp(); const notebook = node('Notebook 1.10');