julian.kiedaischandClaude Opus 5.5
8cfc518f0c
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>
296 lines
15 KiB
TypeScript
296 lines
15 KiB
TypeScript
// Server F's shared documents (F5.1) with real clients (Hocuspocus' own
|
|
// provider over WebSocket) and a real Postgres: a project comes back as it
|
|
// was after the service restarts, two people see each other's changes at
|
|
// once, the updates are compacted (after 500, a minute after the last
|
|
// change, when the last person leaves), the janitor repairs what a merge
|
|
// leaves — once, as a client of its own —, a project the service does not
|
|
// know is refused and not made, read-only people cannot write, /healthz
|
|
// answers, and the log is JSON.
|
|
|
|
import { randomUUID } from 'node:crypto';
|
|
import pg from 'pg';
|
|
import { afterAll, beforeAll, describe, expect, it } from 'vitest';
|
|
import WebSocket from 'ws';
|
|
import * as Y from 'yjs';
|
|
import { addDevice, connect, emptyProject, readProject, ROOTS, topologyView, writeProject } from '@mdg-filius/doc';
|
|
import { COMPACT_AFTER_IDLE_MILLISECONDS, COMPACT_AFTER_UPDATES, JANITOR_DELAY_MILLISECONDS, PostgresSessions, PostgresStore, type DocStore } from '../src/index.js';
|
|
import { testDatabase, type TestDatabase } from './database.js';
|
|
import { enter, newProject as makeProject, startTestServer, until, type TestServer } from './harness.js';
|
|
|
|
let database: TestDatabase;
|
|
let store: PostgresStore;
|
|
let sessions: PostgresSessions;
|
|
let sql: pg.Pool;
|
|
|
|
beforeAll(async () => {
|
|
database = await testDatabase();
|
|
store = PostgresStore.connect(database.url);
|
|
sessions = new PostgresSessions(store.pool);
|
|
await store.migrate();
|
|
await sessions.migrate();
|
|
sql = new pg.Pool({ connectionString: database.url });
|
|
}, 60_000);
|
|
|
|
afterAll(async () => {
|
|
await sql?.end();
|
|
await store?.close();
|
|
await database?.stop();
|
|
});
|
|
|
|
const start = (options: { readonly idle?: number; readonly store?: DocStore } = {}): Promise<TestServer> =>
|
|
startTestServer(options.store ?? store, sessions, options.idle === undefined ? {} : { compactAfterIdleMilliseconds: options.idle });
|
|
const newProject = () => makeProject(store);
|
|
/** A person signed in to the project with a write grant (a reader with `read`), in it. */
|
|
const join = (server: TestServer, project: string, eager = false, level: 'read' | 'write' = 'write') => enter(server, project, { level }, { eager });
|
|
|
|
const deviceCount = (doc: Y.Doc) => doc.getMap(ROOTS.nodes).size;
|
|
const NAMES = { pc: 'Rechner', networkSwitch: 'Switch' } as const;
|
|
/** A device as the palette puts one down. */
|
|
const place = (doc: Y.Doc, id: string, kind: keyof typeof NAMES, x: number, origin: string) => addDevice(doc, { id, kind, name: NAMES[kind], x, y: 0, newId: randomUUID }, origin);
|
|
|
|
describe('server F keeps the shared documents', () => {
|
|
it('gives a project back as it was after the service restarts, from its compacted state', async () => {
|
|
const project = await newProject();
|
|
let server = await start();
|
|
const first = await join(server, project);
|
|
place(first.doc, 'a', 'pc', 10, 'person');
|
|
place(first.doc, 'b', 'networkSwitch', 200, 'person');
|
|
await until(async () => (await store.pending(project)) >= 2);
|
|
first.close();
|
|
// The last person left: compacted and unloaded.
|
|
await until(() => server.sync.hocuspocus.getDocumentsCount() === 0);
|
|
expect(await store.pending(project)).toBe(0);
|
|
await server.close();
|
|
|
|
server = await start();
|
|
const again = await join(server, project);
|
|
expect(topologyView(again.doc).nodes.map((node) => [node.id, node.kind, node.x])).toEqual([
|
|
['a', 'pc', 10],
|
|
['b', 'networkSwitch', 200],
|
|
]);
|
|
// Loading stores nothing anew.
|
|
expect(await store.pending(project)).toBe(0);
|
|
again.close();
|
|
await server.close();
|
|
});
|
|
|
|
it('takes the updates a hard stop left uncompacted, and counts them towards the next compaction', async () => {
|
|
const project = await newProject();
|
|
// The service stopped hard: three updates stored, never compacted.
|
|
const before = new Y.Doc();
|
|
Y.applyUpdate(before, (await store.load(project))!.state);
|
|
before.on('update', (update: Uint8Array) => void store.append(project, update));
|
|
for (const [id, x] of [
|
|
['a', 0],
|
|
['b', 100],
|
|
['c', 200],
|
|
] as const)
|
|
place(before, id, 'pc', x, 'person');
|
|
await until(async () => (await store.pending(project)) === 3);
|
|
const server = await start();
|
|
const person = await join(server, project);
|
|
expect(topologyView(person.doc).nodes.map((node) => node.id)).toEqual(['a', 'b', 'c']);
|
|
const notes = person.doc.getMap<number>('test');
|
|
for (let index = 0; index < COMPACT_AFTER_UPDATES - 3; index++) notes.set('n', index);
|
|
await until(() => server.log.lines.some((line) => line.event === 'doc.compacted'));
|
|
expect(server.log.lines.find((line) => line.event === 'doc.compacted')!.updates).toBe(COMPACT_AFTER_UPDATES);
|
|
person.close();
|
|
await server.close();
|
|
});
|
|
|
|
it('compacts once after exactly 500 updates, and not again when the document has been quiet since', async () => {
|
|
const project = await newProject();
|
|
const server = await start({ idle: 200 });
|
|
const person = await join(server, project);
|
|
const notes = person.doc.getMap<number>('test');
|
|
for (let index = 0; index < COMPACT_AFTER_UPDATES; index++) notes.set('n', index);
|
|
await until(() => server.log.lines.some((line) => line.event === 'doc.compacted'), 15_000);
|
|
await new Promise((resolve) => setTimeout(resolve, 600));
|
|
expect(server.log.lines.filter((line) => line.event === 'doc.compacted').map((line) => line.updates)).toEqual([COMPACT_AFTER_UPDATES]);
|
|
expect((await sql.query('SELECT rev FROM filius_doc WHERE project_uuid = $1', [project])).rows[0].rev).toBe(1);
|
|
person.close();
|
|
await server.close();
|
|
});
|
|
|
|
it('compacts nothing after a visit that changed nothing', async () => {
|
|
const project = await newProject();
|
|
const rev = async () => (await sql.query('SELECT rev FROM filius_doc WHERE project_uuid = $1', [project])).rows[0].rev as number;
|
|
const server = await start({ idle: 100 });
|
|
const person = await join(server, project);
|
|
await new Promise((resolve) => setTimeout(resolve, 300));
|
|
person.close();
|
|
await until(() => server.sync.hocuspocus.getDocumentsCount() === 0);
|
|
expect(await rev()).toBe(0);
|
|
expect(server.log.lines.some((line) => line.event === 'doc.compacted')).toBe(false);
|
|
await server.close();
|
|
});
|
|
|
|
it('stores everything before it closes, with people still there who come back at once', async () => {
|
|
const project = await newProject();
|
|
const server = await start();
|
|
const person = await join(server, project, true);
|
|
place(person.doc, 'a', 'pc', 0, 'person');
|
|
await until(async () => (await store.pending(project)) === 1);
|
|
await server.close();
|
|
expect(await store.pending(project)).toBe(0);
|
|
const stored = new Y.Doc();
|
|
Y.applyUpdate(stored, (await store.load(project))!.state);
|
|
expect(deviceCount(stored)).toBe(1);
|
|
person.close();
|
|
});
|
|
|
|
it('says so in the log when a project went while it was open, and stores nothing in its place', async () => {
|
|
const project = await newProject();
|
|
const server = await start();
|
|
const person = await join(server, project);
|
|
place(person.doc, 'a', 'pc', 0, 'person');
|
|
await until(async () => (await store.pending(project)) === 1);
|
|
await store.remove(project);
|
|
person.close();
|
|
await until(() => server.sync.hocuspocus.getDocumentsCount() === 0);
|
|
expect(server.log.lines.filter((line) => line.event === 'store.failed').map((line) => line.project)).toEqual([project]);
|
|
expect(await store.load(project)).toBeUndefined();
|
|
await server.close();
|
|
});
|
|
|
|
it('makes a project once: a second one of the same id changes nothing', async () => {
|
|
const project = await newProject();
|
|
const first = (await store.load(project))!.state;
|
|
expect(await store.create(project, new Uint8Array([0, 0]))).toBe(false);
|
|
expect((await store.load(project))!.state).toEqual(first);
|
|
});
|
|
|
|
it("shows each person the other's changes at once", async () => {
|
|
const project = await newProject();
|
|
const server = await start();
|
|
const [mia, ben] = await Promise.all([join(server, project), join(server, project)]);
|
|
place(mia.doc, 'pc', 'pc', 0, 'mia');
|
|
place(ben.doc, 'switch', 'networkSwitch', 100, 'ben');
|
|
await until(() => deviceCount(mia.doc) === 2 && deviceCount(ben.doc) === 2);
|
|
connect(mia.doc, { node: 'pc' }, { node: 'switch' }, 'cable', 'mia');
|
|
await until(() => ben.doc.getMap(ROOTS.links).size === 1);
|
|
expect(readProject(ben.doc)).toEqual(readProject(mia.doc));
|
|
mia.close();
|
|
ben.close();
|
|
await server.close();
|
|
});
|
|
|
|
it(`compacts after ${COMPACT_AFTER_UPDATES} updates, and a minute after the last change`, async () => {
|
|
const project = await newProject();
|
|
const server = await start({ idle: 300 });
|
|
const person = await join(server, project);
|
|
const notes = person.doc.getMap<number>('test');
|
|
for (let index = 0; index < COMPACT_AFTER_UPDATES + 20; index++) notes.set('n', index);
|
|
await until(async () => (await sql.query('SELECT rev FROM filius_doc WHERE project_uuid = $1', [project])).rows[0].rev >= 1, 15_000);
|
|
// The compaction took the first 500; what came after waits — until the document has been quiet for a while.
|
|
await until(async () => (await store.pending(project)) === 0, 10_000);
|
|
expect(server.log.lines.filter((line) => line.event === 'doc.compacted').map((line) => line.updates)).toEqual([COMPACT_AFTER_UPDATES, 20]);
|
|
const stored = await store.load(project);
|
|
const reloaded = new Y.Doc();
|
|
Y.applyUpdate(reloaded, stored!.state);
|
|
expect(reloaded.getMap('test').get('n')).toBe(COMPACT_AFTER_UPDATES + 19);
|
|
person.close();
|
|
await server.close();
|
|
});
|
|
|
|
it('repairs what a merge leaves, once, as a client of its own', async () => {
|
|
const project = await newProject();
|
|
const server = await start();
|
|
const [mia, ben] = await Promise.all([join(server, project), join(server, project)]);
|
|
place(mia.doc, 'pc', 'pc', 0, 'mia');
|
|
place(mia.doc, 'switch', 'networkSwitch', 100, 'mia');
|
|
await until(() => deviceCount(ben.doc) === 2);
|
|
// Both plug a cable into the computer's one card at once; Mia's goes to a port that is gone too.
|
|
const port = topologyView(mia.doc).nodes.find((node) => node.id === 'pc')!.ports[0]!.id;
|
|
const switchPorts = topologyView(mia.doc)
|
|
.nodes.find((node) => node.id === 'switch')!
|
|
.ports.map((each) => each.id);
|
|
const cable = (doc: Y.Doc, id: string, targetPort: string) =>
|
|
doc.transact(() => {
|
|
const link = new Y.Map<unknown>();
|
|
for (const [key, value] of Object.entries({ sourceNode: 'pc', sourcePort: port, targetNode: 'switch', targetPort, order: 0 })) link.set(key, value);
|
|
doc.getMap(ROOTS.links).set(id, link);
|
|
}, 'person');
|
|
cable(mia.doc, 'cable-b', switchPorts[0]!);
|
|
cable(ben.doc, 'cable-a', switchPorts[1]!);
|
|
cable(mia.doc, 'cable-c', 'no-such-port');
|
|
await until(() => [...ben.doc.getMap(ROOTS.links).keys()].join() === 'cable-a' && [...mia.doc.getMap(ROOTS.links).keys()].join() === 'cable-a');
|
|
await new Promise((resolve) => setTimeout(resolve, 600));
|
|
const repairs = server.log.lines.filter((line) => line.event === 'janitor.repaired');
|
|
expect(repairs.reduce((sum, line) => sum + (line.danglingLinks as number) + (line.sharedPortLinks as number), 0)).toBe(2);
|
|
mia.close();
|
|
ben.close();
|
|
await server.close();
|
|
});
|
|
|
|
it('repairs a stored document when it is loaded, before anyone changes it', async () => {
|
|
const project = randomUUID();
|
|
const doc = new Y.Doc();
|
|
writeProject(doc, emptyProject());
|
|
doc.transact(() => {
|
|
const link = new Y.Map<unknown>();
|
|
for (const [key, value] of Object.entries({ sourceNode: 'gone', sourcePort: 'p', targetNode: 'gone too', targetPort: 'q', order: 0 })) link.set(key, value);
|
|
doc.getMap(ROOTS.links).set('stray', link);
|
|
});
|
|
await store.create(project, Y.encodeStateAsUpdate(doc));
|
|
const server = await start();
|
|
const person = await join(server, project);
|
|
await until(() => person.doc.getMap(ROOTS.links).size === 0);
|
|
expect(server.log.lines.find((line) => line.event === 'janitor.repaired')).toMatchObject({ project, danglingLinks: 1 });
|
|
person.close();
|
|
await server.close();
|
|
});
|
|
|
|
it("keeps the plan's measures: compacted after 500 updates or a minute of quiet, repaired 250 ms after a change", () => {
|
|
expect([COMPACT_AFTER_UPDATES, COMPACT_AFTER_IDLE_MILLISECONDS, JANITOR_DELAY_MILLISECONDS]).toEqual([500, 60_000, 250]);
|
|
});
|
|
|
|
it('refuses a project its store does not have when it comes to load it, and makes none', async () => {
|
|
// Deleted between the sign-in and the loading: the store has it no more.
|
|
const project = await newProject();
|
|
const forgetful = Object.assign(Object.create(store) as PostgresStore, { load: async () => undefined });
|
|
const server = await start({ store: forgetful });
|
|
await expect(join(server, project)).rejects.toThrow('refused: unknownProject');
|
|
expect(server.log.lines.some((line) => line.event === 'doc.unknown' && line.project === project)).toBe(true);
|
|
expect(server.sync.hocuspocus.getDocumentsCount()).toBe(0);
|
|
await server.close();
|
|
});
|
|
|
|
it('lets a reader see the document and change nothing', async () => {
|
|
const project = await newProject();
|
|
const writerServer = await start();
|
|
const writer = await join(writerServer, project);
|
|
place(writer.doc, 'pc', 'pc', 0, 'person');
|
|
await until(async () => (await store.pending(project)) >= 1);
|
|
writer.close();
|
|
await writerServer.close();
|
|
const server = await start();
|
|
const reader = await join(server, project, false, 'read');
|
|
expect(deviceCount(reader.doc)).toBe(1);
|
|
place(reader.doc, 'switch', 'networkSwitch', 0, 'reader');
|
|
await new Promise((resolve) => setTimeout(resolve, 400));
|
|
expect(await store.pending(project)).toBe(0);
|
|
reader.close();
|
|
await server.close();
|
|
const stored = new Y.Doc();
|
|
Y.applyUpdate(stored, (await store.load(project))!.state);
|
|
expect(deviceCount(stored)).toBe(1);
|
|
});
|
|
|
|
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, rooms: 0 }]);
|
|
expect((await fetch(`http://127.0.0.1:${server.port}/nothing`)).status).toBe(404);
|
|
const refused = await new Promise<number>((resolve) => {
|
|
const socket = new WebSocket(`ws://127.0.0.1:${server.port}/elsewhere`);
|
|
socket.on('unexpected-response', (_request, response) => resolve(response.statusCode ?? 0));
|
|
socket.on('error', () => resolve(-1));
|
|
});
|
|
expect(refused).toBe(404);
|
|
await server.close();
|
|
expect(server.log.lines.map((line) => line.event)).toEqual(['server.listening', 'server.closed']);
|
|
expect(server.log.lines[0]).toEqual({ level: 'info', event: 'server.listening', port: server.port });
|
|
});
|
|
});
|