julian.kiedaischandClaude Opus 5.5
12b26f4abe
Who started the simulation, "Pakete verwerfen" and the dropped packets move from the toolbar into a properties sheet beside the surface. New in it: restart the simulation, after asking. The room learns a restart message: stopped as by stop, every capture, dropped frame and program window emptied, started anew and named after who restarted; refused for readers and frozen projects, logged, and waiting for asked blobs like a start. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
272 lines
17 KiB
TypeScript
272 lines
17 KiB
TypeScript
// The large binary files on F (F5.5): kept by their SHA-256 through POST
|
||
// /blobs from the editor's origin with a write grant, given back through GET
|
||
// /blobs/<sha> to whoever has a grant, with CORS for the editor's origins;
|
||
// refused from another origin, to a reader, in a frozen project, without a
|
||
// session, beyond the size. A room gets the blobs its document names before
|
||
// its simulation starts: a picture copied in a command line is the same
|
||
// blob, and a file it did not touch stays one.
|
||
|
||
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 { attachBlobs, BLOB_THRESHOLD_BYTES, blobSha, changeFiles, memoryBlobSide, nodeFiles, ROOTS, writeProject } from '@mdg-filius/doc';
|
||
import { openProjectArchive } from '@mdg-filius/fls';
|
||
import type { HostMessage } from '@mdg-filius/protocol';
|
||
import { BLOB_MAX_BYTES, PostgresBlobs, PostgresSessions, PostgresStore, shaOf } from '../src/index.js';
|
||
import { BLOB_ASK_AGAIN_MILLISECONDS, BLOB_ASKS } from '../src/simChannel.js';
|
||
import { testDatabase, type TestDatabase } from './database.js';
|
||
import { EDITOR_ORIGIN, newProject, SERVICE_KEY, signIn, startTestServer, until, type TestServer } from './harness.js';
|
||
|
||
let database: TestDatabase;
|
||
let store: PostgresStore;
|
||
let sessions: PostgresSessions;
|
||
let blobs: PostgresBlobs;
|
||
|
||
beforeAll(async () => {
|
||
database = await testDatabase();
|
||
store = PostgresStore.connect(database.url);
|
||
sessions = new PostgresSessions(store.pool);
|
||
blobs = new PostgresBlobs(store.pool);
|
||
await store.migrate();
|
||
await sessions.migrate();
|
||
await blobs.migrate();
|
||
}, 60_000);
|
||
|
||
afterAll(async () => {
|
||
await store?.close();
|
||
await database?.stop();
|
||
});
|
||
|
||
const picture = (length = BLOB_THRESHOLD_BYTES * 2, seed = 7) => Uint8Array.from({ length }, (_, index) => (index * 13 + seed) % 256);
|
||
|
||
function post(server: TestServer, project: string, body: Uint8Array, headers: Record<string, string>): Promise<Response> {
|
||
return fetch(`${server.base}/blobs?project=${project}`, { method: 'POST', headers: { 'content-type': 'application/octet-stream', ...headers }, body });
|
||
}
|
||
const get = (server: TestServer, project: string, sha: string, headers: Record<string, string>) => fetch(`${server.base}/blobs/${sha}?project=${project}`, { headers });
|
||
|
||
/** F's store, slower by so many milliseconds in reading. */
|
||
function slowly(inner: PostgresBlobs, { get }: { readonly get: number }): PostgresBlobs {
|
||
return Object.assign(Object.create(inner) as PostgresBlobs, {
|
||
get: async (shas: readonly string[]) => (await new Promise((resolve) => setTimeout(resolve, get)), inner.get(shas)),
|
||
});
|
||
}
|
||
|
||
/** FILIUS' example with a picture as a blob on the notebook (its bytes not in F's store). */
|
||
async function projectWithPicture(seed: number) {
|
||
const project = randomUUID();
|
||
const imported = openProjectArchive(new Uint8Array(readFileSync(new URL('../../../corpus/java-examples/internetwork_komplett.fls', import.meta.url))), () => randomUUID());
|
||
const notebook = imported.graph.nodes.find((each) => each.displayName === 'Notebook 1.10')!.id;
|
||
const doc = new Y.Doc();
|
||
attachBlobs(doc, memoryBlobSide());
|
||
writeProject(doc, imported);
|
||
const bytes = picture(BLOB_THRESHOLD_BYTES * 2, seed);
|
||
changeFiles(doc, notebook, 'person', (files) => files.writeImageFile('/spaet.png', bytes, 'image/png'));
|
||
await store.create(project, Y.encodeStateAsUpdate(doc));
|
||
return { project, notebook, bytes };
|
||
}
|
||
|
||
/** In a project's room by WebSocket, signed in with a write grant. */
|
||
async function enterRoom(server: TestServer, project: string) {
|
||
const cookie = await signIn(server, project);
|
||
const socket = new WebSocket(`ws://127.0.0.1:${server.port}/sim?project=${project}`, { headers: { cookie } });
|
||
const got: HostMessage[] = [];
|
||
socket.on('message', (data) => got.push(JSON.parse(data.toString()) as HostMessage));
|
||
socket.on('error', () => {});
|
||
await new Promise((resolve) => socket.on('open', resolve));
|
||
return { got, send: (message: object) => socket.send(JSON.stringify({ v: 1, ...message })), close: () => socket.close() };
|
||
}
|
||
|
||
/** A computer's files as F's store holds them. */
|
||
async function storedFiles(project: string, node: string): Promise<Y.Map<Y.Map<unknown>>> {
|
||
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);
|
||
return doc.getMap<Y.Map<unknown>>(ROOTS.nodes).get(node)!.get('files') as Y.Map<Y.Map<unknown>>;
|
||
}
|
||
|
||
describe('the blobs on F', () => {
|
||
it('keeps bytes under the name F gives them, and gives them back to whoever has a grant', async () => {
|
||
const server = await startTestServer(store, sessions);
|
||
const project = await newProject(store);
|
||
const cookie = await signIn(server, project, { login: 'ab12' });
|
||
const bytes = picture();
|
||
const kept = await post(server, project, bytes, { cookie, origin: EDITOR_ORIGIN });
|
||
expect([kept.status, await kept.json()]).toEqual([201, { sha: blobSha(bytes) }]);
|
||
expect(kept.headers.get('access-control-allow-origin')).toBe(EDITOR_ORIGIN);
|
||
expect(kept.headers.get('access-control-allow-credentials')).toBe('true');
|
||
expect(shaOf(bytes)).toBe(blobSha(bytes));
|
||
// Again: the same name, kept once.
|
||
expect(await (await post(server, project, bytes, { cookie, origin: EDITOR_ORIGIN })).json()).toEqual({ sha: blobSha(bytes) });
|
||
expect((await store.pool.query('SELECT count(*)::int AS n FROM filius_blob WHERE sha = $1', [blobSha(bytes)])).rows[0].n).toBe(1);
|
||
const reader = await signIn(server, project, { level: 'read' });
|
||
const back = await get(server, project, blobSha(bytes), { cookie: reader, origin: EDITOR_ORIGIN });
|
||
expect(back.status).toBe(200);
|
||
expect(new Uint8Array(await back.arrayBuffer())).toEqual(bytes);
|
||
expect([back.headers.get('content-type'), back.headers.get('cache-control'), back.headers.get('access-control-allow-origin')]).toEqual([
|
||
'application/octet-stream',
|
||
'private, max-age=31536000, immutable',
|
||
EDITOR_ORIGIN,
|
||
]);
|
||
// Without an origin (the same origin), no CORS.
|
||
expect((await get(server, project, blobSha(bytes), { cookie: reader })).headers.get('access-control-allow-origin')).toBeNull();
|
||
expect(server.log.lines.find((line) => line.event === 'blob.kept')).toMatchObject({ project, user: 'ab12', sha: blobSha(bytes), size: bytes.length });
|
||
await server.close();
|
||
});
|
||
|
||
it('answers the preflight of the editor’s origin, and no other', async () => {
|
||
const server = await startTestServer(store, sessions);
|
||
const ask = (origin: string) => fetch(`${server.base}/blobs?project=x`, { method: 'OPTIONS', headers: { origin, 'access-control-request-method': 'POST' } });
|
||
const ours = await ask(EDITOR_ORIGIN);
|
||
expect([ours.status, ours.headers.get('access-control-allow-methods'), ours.headers.get('access-control-allow-headers')]).toEqual([204, 'GET, POST', 'content-type']);
|
||
expect((await ask('https://elsewhere.test')).status).toBe(403);
|
||
await server.close();
|
||
});
|
||
|
||
it('refuses another origin, a reader, a frozen project, no session, no project and too many bytes', async () => {
|
||
const server = await startTestServer(store, sessions);
|
||
const project = await newProject(store);
|
||
const cookie = await signIn(server, project);
|
||
const reader = await signIn(server, project, { level: 'read' });
|
||
const bytes = picture(100);
|
||
const answer = async (response: Response) => [response.status, await response.json()];
|
||
expect(await answer(await post(server, project, bytes, { cookie }))).toEqual([403, { error: 'foreignOrigin' }]);
|
||
expect(await answer(await post(server, project, bytes, { cookie, origin: 'https://elsewhere.test' }))).toEqual([403, { error: 'foreignOrigin' }]);
|
||
expect(await answer(await post(server, project, bytes, { cookie: reader, origin: EDITOR_ORIGIN }))).toEqual([403, { error: 'readOnly' }]);
|
||
expect(await answer(await post(server, project, bytes, { origin: EDITOR_ORIGIN }))).toEqual([403, { error: 'notSignedIn' }]);
|
||
expect(await answer(await post(server, 'nothing', bytes, { cookie, origin: EDITOR_ORIGIN }))).toEqual([400, { error: 'notAProject' }]);
|
||
expect(await answer(await post(server, randomUUID(), bytes, { cookie, origin: EDITOR_ORIGIN }))).toEqual([403, { error: 'unknownProject' }]);
|
||
expect(await answer(await post(server, project, new Uint8Array(BLOB_MAX_BYTES + 1), { cookie, origin: EDITOR_ORIGIN }))).toEqual([413, { error: 'tooLarge' }]);
|
||
expect(server.log.lines.filter((line) => line.event === 'blob.foreignOrigin')).toHaveLength(2);
|
||
await fetch(`${server.base}/svc/projects/${project}/freeze`, { method: 'POST', headers: { 'x-filius-service-key': SERVICE_KEY }, body: JSON.stringify({ frozen: true }) });
|
||
expect(await answer(await post(server, project, bytes, { cookie, origin: EDITOR_ORIGIN }))).toEqual([403, { error: 'frozen' }]);
|
||
expect(await answer(await get(server, project, 'f'.repeat(64), { cookie }))).toEqual([404, { error: 'unknownBlob' }]);
|
||
expect(await answer(await get(server, project, 'not-a-sha', { cookie }))).toEqual([404, { error: 'unknownBlob' }]);
|
||
expect(await answer(await get(server, randomUUID(), blobSha(bytes), { cookie }))).toEqual([403, { error: 'unknownProject' }]);
|
||
expect(await answer(await fetch(`${server.base}/blobs/x/y?project=${project}`, { headers: { cookie } }))).toEqual([404, { error: 'notFound' }]);
|
||
expect(BLOB_MAX_BYTES).toBe(16 * 1024 * 1024);
|
||
await server.close();
|
||
});
|
||
|
||
it('gives a room the blobs of its document before the simulation starts', async () => {
|
||
const server = await startTestServer(store, sessions);
|
||
// FILIUS' example, with a picture as a blob on the notebook.
|
||
const project = randomUUID();
|
||
const imported = openProjectArchive(new Uint8Array(readFileSync(new URL('../../../corpus/java-examples/internetwork_komplett.fls', import.meta.url))), () => randomUUID());
|
||
const notebook = imported.graph.nodes.find((each) => each.displayName === 'Notebook 1.10')!.id;
|
||
const doc = new Y.Doc();
|
||
const side = memoryBlobSide();
|
||
attachBlobs(doc, side);
|
||
writeProject(doc, imported);
|
||
const bytes = picture();
|
||
changeFiles(doc, notebook, 'person', (files) => files.writeImageFile('/bild.png', bytes, 'image/png'));
|
||
await blobs.put(bytes);
|
||
await store.create(project, Y.encodeStateAsUpdate(doc));
|
||
const cookie = await signIn(server, project);
|
||
const socket = new WebSocket(`ws://127.0.0.1:${server.port}/sim?project=${project}`, { headers: { cookie } });
|
||
const got: HostMessage[] = [];
|
||
socket.on('message', (data) => got.push(JSON.parse(data.toString()) as HostMessage));
|
||
await new Promise((resolve) => socket.on('open', resolve));
|
||
const send = (message: object) => socket.send(JSON.stringify({ v: 1, ...message }));
|
||
send({ kind: 'start' });
|
||
await until(() => got.some((message) => message.kind === 'state' && message.mode === 'simulation'));
|
||
send({ kind: 'input', node: notebook, app: 'terminal', action: 'enter', args: { line: 'cp bild.png kopie.png' }, reqId: 'r' });
|
||
await until(() => got.some((message) => message.kind === 'ack'));
|
||
send({ kind: 'stop' });
|
||
await until(() => got.filter((message) => message.kind === 'state' && message.mode === 'design').length === 2);
|
||
socket.close();
|
||
await server.close();
|
||
const stored = await store.load(project);
|
||
const after = new Y.Doc();
|
||
Y.applyUpdate(after, stored!.state);
|
||
for (const update of stored!.updates) Y.applyUpdate(after, update);
|
||
const files = after.getMap<Y.Map<unknown>>(ROOTS.nodes).get(notebook)!.get('files') as Y.Map<Y.Map<unknown>>;
|
||
// The copy is the same blob: the room had the bytes; the original is untouched.
|
||
expect([files.get('/kopie.png')?.get('blob'), files.get('/bild.png')?.get('blob')]).toEqual([blobSha(bytes), blobSha(bytes)]);
|
||
expect(nodeFiles(after, notebook)!.entry('/kopie.png')!.content).toMatchObject({ kind: 'image', mediaType: 'image/png' });
|
||
}, 30_000);
|
||
|
||
it('asks again for a blob F gets only after the start, and the running simulation reads it', async () => {
|
||
const server = await startTestServer(store, sessions);
|
||
const project = randomUUID();
|
||
const imported = openProjectArchive(new Uint8Array(readFileSync(new URL('../../../corpus/java-examples/internetwork_komplett.fls', import.meta.url))), () => randomUUID());
|
||
const notebook = imported.graph.nodes.find((each) => each.displayName === 'Notebook 1.10')!.id;
|
||
const doc = new Y.Doc();
|
||
attachBlobs(doc, memoryBlobSide());
|
||
writeProject(doc, imported);
|
||
const bytes = picture(BLOB_THRESHOLD_BYTES * 2, 42);
|
||
changeFiles(doc, notebook, 'person', (files) => files.writeImageFile('/spaet.png', bytes, 'image/png'));
|
||
await store.create(project, Y.encodeStateAsUpdate(doc));
|
||
const cookie = await signIn(server, project);
|
||
const socket = new WebSocket(`ws://127.0.0.1:${server.port}/sim?project=${project}`, { headers: { cookie } });
|
||
const got: HostMessage[] = [];
|
||
socket.on('message', (data) => got.push(JSON.parse(data.toString()) as HostMessage));
|
||
await new Promise((resolve) => socket.on('open', resolve));
|
||
socket.send(JSON.stringify({ v: 1, kind: 'start' }));
|
||
await until(() => got.some((message) => message.kind === 'state' && message.mode === 'simulation'));
|
||
// Taken to F only now (an editor still uploading it); the thread asks again a second later.
|
||
await blobs.put(bytes);
|
||
await new Promise((resolve) => setTimeout(resolve, BLOB_ASK_AGAIN_MILLISECONDS + 500));
|
||
socket.send(JSON.stringify({ v: 1, kind: 'input', node: notebook, app: 'terminal', action: 'enter', args: { line: 'cp spaet.png kopie.png' }, reqId: 'r' }));
|
||
await until(() => got.some((message) => message.kind === 'ack'));
|
||
socket.close();
|
||
await server.close();
|
||
const stored = await store.load(project);
|
||
const after = new Y.Doc();
|
||
Y.applyUpdate(after, stored!.state);
|
||
for (const update of stored!.updates) Y.applyUpdate(after, update);
|
||
const files = after.getMap<Y.Map<unknown>>(ROOTS.nodes).get(notebook)!.get('files') as Y.Map<Y.Map<unknown>>;
|
||
expect(files.get('/kopie.png')?.get('blob')).toBe(blobSha(bytes));
|
||
expect(BLOB_ASKS).toBe(5);
|
||
}, 30_000);
|
||
|
||
it.each([
|
||
['start', 51],
|
||
['restart', 52],
|
||
] as const)("starts a simulation (%s) only once the blobs asked for have come, however slow F's store is", async (kind, seed) => {
|
||
const { project, notebook, bytes } = await projectWithPicture(seed);
|
||
await blobs.put(bytes);
|
||
const server = await startTestServer(store, sessions, { blobs: slowly(blobs, { get: 800 }) });
|
||
const room = await enterRoom(server, project);
|
||
room.send({ kind });
|
||
await until(() => room.got.some((message) => message.kind === 'state' && message.mode === 'simulation'), 5_000);
|
||
room.send({ kind: 'input', node: notebook, app: 'terminal', action: 'enter', args: { line: 'cp spaet.png kopie.png' }, reqId: 'r' });
|
||
await until(() => room.got.some((message) => message.kind === 'ack'));
|
||
room.close();
|
||
await server.close();
|
||
expect((await storedFiles(project, notebook)).get('/kopie.png')?.get('blob')).toBe(blobSha(bytes));
|
||
}, 30_000);
|
||
|
||
it('starts a room whose blob F has not got, and empties no file by it', async () => {
|
||
const server = await startTestServer(store, sessions);
|
||
const project = randomUUID();
|
||
const imported = openProjectArchive(new Uint8Array(readFileSync(new URL('../../../corpus/java-examples/internetwork_komplett.fls', import.meta.url))), () => randomUUID());
|
||
const notebook = imported.graph.nodes.find((each) => each.displayName === 'Notebook 1.10')!.id;
|
||
const doc = new Y.Doc();
|
||
attachBlobs(doc, memoryBlobSide());
|
||
writeProject(doc, imported);
|
||
const bytes = picture(BLOB_THRESHOLD_BYTES * 3, 99);
|
||
changeFiles(doc, notebook, 'person', (files) => files.writeImageFile('/fehlt.png', bytes, 'image/png'));
|
||
await store.create(project, Y.encodeStateAsUpdate(doc));
|
||
const cookie = await signIn(server, project);
|
||
const socket = new WebSocket(`ws://127.0.0.1:${server.port}/sim?project=${project}`, { headers: { cookie } });
|
||
const got: HostMessage[] = [];
|
||
socket.on('message', (data) => got.push(JSON.parse(data.toString()) as HostMessage));
|
||
await new Promise((resolve) => socket.on('open', resolve));
|
||
socket.send(JSON.stringify({ v: 1, kind: 'start' }));
|
||
await until(() => got.some((message) => message.kind === 'state' && message.mode === 'simulation'));
|
||
socket.send(JSON.stringify({ v: 1, kind: 'input', node: notebook, app: 'terminal', action: 'enter', args: { line: 'touch neu.txt' }, reqId: 'r' }));
|
||
await until(() => got.some((message) => message.kind === 'ack'));
|
||
socket.close();
|
||
await server.close();
|
||
const stored = await store.load(project);
|
||
const after = new Y.Doc();
|
||
Y.applyUpdate(after, stored!.state);
|
||
for (const update of stored!.updates) Y.applyUpdate(after, update);
|
||
const files = after.getMap<Y.Map<unknown>>(ROOTS.nodes).get(notebook)!.get('files') as Y.Map<Y.Map<unknown>>;
|
||
expect([files.get('/fehlt.png')?.get('blob'), files.has('/neu.txt')]).toEqual([blobSha(bytes), true]);
|
||
}, 30_000);
|
||
});
|