From 4b33e365806df5e56e8e5203c108435ec23a09b9 Mon Sep 17 00:00:00 2001 From: Julian Kiedaisch Date: Sun, 4 Oct 2026 16:46:50 +0000 Subject: [PATCH] feat: F5.5 large binary files as blobs on server F From 16 KB a picture's or binary file's entry names its bytes by their SHA-256 instead of holding them; the bytes are on F, content-addressed (GET/POST /blobs with the session's grant, from the editor's origin, at most 16 MiB). A document's blob side says where they are: the editor's fetches and uploads, the room thread's asks F's main thread, and a start waits for the ones asked for; without a side everything stays in the document. Text stays Y.Text. A blob whose bytes are not there is never written back empty. A blob F has not got yet is asked for again a few times, and a running room reads late bytes. Co-Authored-By: Claude Opus 5.5 --- CHANGES.md | 7 +- package-lock.json | 13 + packages/doc/package.json | 1 + packages/doc/src/blobs.ts | 95 +++++++ packages/doc/src/commands.ts | 3 +- packages/doc/src/files.ts | 20 +- packages/doc/src/index.ts | 1 + packages/doc/src/projectDoc.ts | 34 ++- packages/doc/test/blobs.test.ts | 158 +++++++++++ packages/editor/browser/serverF.ts | 6 + packages/editor/browser/shared.spec.ts | 50 ++++ packages/editor/src/desktop/fileExplorer.ts | 4 +- packages/editor/src/i18n/de.json | 1 + packages/editor/src/i18n/en.json | 1 + packages/editor/src/remote/SharedProject.tsx | 40 ++- packages/editor/src/remote/remoteBlobs.ts | 89 ++++++ packages/editor/src/sheet/sheet.css | 16 ++ packages/editor/test/remoteBlobs.test.ts | 135 ++++++++++ packages/server/src/blobRoutes.ts | 72 +++++ packages/server/src/blobs.ts | 52 ++++ packages/server/src/index.ts | 2 + packages/server/src/main.ts | 4 + packages/server/src/server.ts | 8 +- packages/server/src/simChannel.ts | 16 +- packages/server/src/simRooms.ts | 36 ++- packages/server/src/simWorker.ts | 63 ++++- packages/server/test/blobs.test.ts | 268 +++++++++++++++++++ packages/server/test/harness.ts | 10 +- packages/sim/src/room.ts | 9 + 29 files changed, 1185 insertions(+), 29 deletions(-) create mode 100644 packages/doc/src/blobs.ts create mode 100644 packages/doc/test/blobs.test.ts create mode 100644 packages/editor/src/remote/remoteBlobs.ts create mode 100644 packages/editor/test/remoteBlobs.test.ts create mode 100644 packages/server/src/blobRoutes.ts create mode 100644 packages/server/src/blobs.ts create mode 100644 packages/server/test/blobs.test.ts diff --git a/CHANGES.md b/CHANGES.md index 1ff3eb7..854e85d 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -159,6 +159,7 @@ Swift-Pfade relativ zu `filius-ipad/ios/FiliusPad/TopologyEditor/` (Stand | `test/devices.test.ts`, `test/fixtures/new-devices.fls` | — | neu (F4.4a): neue Geräte jeder Art gegen Javas gespeicherte (Zufall maskiert), Kabelregeln, ein selbst gebautes Netz pingt im Kern, Rückgängig nur der eigenen Änderungen über zwei Replikate | | `test/projectDoc.test.ts` | — | neu (F4.1): alle 23 Korpusdateien → Dokument → Yjs-Update → Projekt: dasselbe Projekt, derselbe Schnappschuss, dieselbe `.fls`, dieselbe Reihenfolge je Ordner; Kennungen absichtlich gegen die Dateireihenfolge | | `test/collaboration.test.ts`, `test/values.test.ts` | — | neu (F4.1): zwei Replikate — zwei Felder eines Geräts, zwei neue Zeilen, gemeinsames Tippen; der Hausmeister nach einem Zusammenführen (I1, I2 auf beiden Replikaten gleich, I4 auch in anderer Schreibweise), idempotent ohne Aktualisierung; `long` einer eigenen Anwendung, Binärdateien | +| `src/blobs.ts`, `src/projectDoc.ts` (`fileEntry`, `bytesOf`), `src/files.ts` (`holds`), `src/commands.ts`, `test/blobs.test.ts` | Plan §5.1 | neu (F5.5): große Binärdateien der Rechner außerhalb des Dokuments — ab 16 KB trägt der Eintrag eines Bilds oder einer Binärdatei `blob` (SHA-256 der Bytes, hex; `@noble/hashes`, MIT) und `size` statt `data`; Text bleibt `Y.Text`, kleinere Binärdateien bleiben im Eintrag; wo die Bytes sind, sagt die Blob-Seite des Dokuments (`attachBlobs`), ohne sie bleibt alles im Dokument wie bisher; ein Blob ohne Bytes liest sich als leer, wird so aber nie zurückgeschrieben (sonst wäre die Datei weg); `missingBlobs`, `onBlobsArrive`, `memoryBlobSide` | ## packages/protocol @@ -202,6 +203,7 @@ Swift-Pfade relativ zu `filius-ipad/ios/FiliusPad/TopologyEditor/` (Stand | `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 | +| `src/room.ts` (`blobsArrived`) | — | erweitert (F5.5): kamen Bytes nach, liest die laufende Simulation die Dateien neu | ## packages/editor @@ -264,7 +266,9 @@ Swift-Pfade relativ zu `filius-ipad/ios/FiliusPad/TopologyEditor/` (Stand | `src/Editor.tsx`, `src/canvas/Canvas.tsx` (`locked`), `src/sheet/SheetLock.tsx`, `src/sheet/DeviceSheet.tsx`, `src/sheet/RouterSheets.tsx`, `src/sheet/FirewallPanel.tsx`, `src/sheet/ServicePanels.tsx`, `src/design/DocumentationSidebar.tsx`, `src/simulation/ModeBar.tsx`, `src/simulation/useSimulation.ts`, `src/i18n/*.json` | Plan §6.2, §6.4, §9.3 | geändert (F5.3b): wer nur liest, sieht alles und ändert nichts — keine Palette, kein Rückgängig, nichts bewegt sich oder hat ein Menü, die Felder des Blatts sind gesperrt (die Reiter bleiben), keine neuen Doku-Elemente, die Simulation wird von ihm weder gestartet noch angehalten; Desktops und Datenaustausch öffnen zum Zuschauen; ein „nein“ des Raums wird gesagt; „Simulation gestartet von …“ und „Nur ansehen“ bzw. „Eingefroren – nur ansehen“ neben den Modusknöpfen | | `src/remote/presence.ts`, `src/remote/usePresence.ts`, `src/remote/PresenceMarks.tsx`, `src/remote/sharedDoc.ts` (`you`, Awareness), `src/remote/SharedProject.tsx` (`onPeople`), `src/Editor.tsx`, `src/canvas/Canvas.tsx` (`chosenBy`), `src/desktop/DesktopWindow.tsx`, `src/simulation/ViewerWindow.tsx`, `src/simulation/CaptureWindow.tsx`, `src/bridge/dashboardBridge.ts` (`people`), `src/main.tsx`, `src/i18n/*.json` | Plan §7.4 | neu (F5.4): die anderen im Projekt über Yjs-Awareness — jeder Editor sagt, wer er ist (von F), welches Gerät er gewählt und welche Fenster er offen hat, und nichts sonst (kein Mauszeiger, nichts, das sich ohne Anlass ändert; gesendet nur bei einer Änderung); fremde Zustände werden geprüft gelesen; ein Rahmen in der Farbe der Person um ein Gerät, das sie gewählt hat, ein Punkt an einem Fenster, das sie auch offen hat, ein Kreis mit den Initialen je Person neben den Modusknöpfen (eine Person mit zwei Tabs einmal); eine Person ohne Farbe des Dashboards bekommt eine aus ihrer Kennung; die Brücke meldet dem Dashboard die Personen (`people`). Name und Farbe sagt der Editor selbst — wer seinen Editor verändert, könnte den anderen im Projekt einen anderen Namen zeigen | | `test/presence.test.ts` | — | neu (F5.4): Lesen fremder Zustände, die anderen ohne sich selbst, eine Person einmal, Rahmen und Punkte, Farben, Initialen, Senden nur bei Änderung | -| `test/remoteSimHost.test.ts`, `browser/shared.spec.ts`, `browser/serverF.ts` | — | neu (F5.3b): `RemoteSimHost` auf nachgebauten Sockets mit gestellter Uhr; drei Personen in drei Browsern an einem echten F (aus dem Bündel gestartet): verschieben, nur lesen, eine Simulation für alle mit Namen, ein Ping in einer Befehlszeile bei der anderen, Eintippen als Leser abgewiesen, eingefroren und aufgetaut, Zugang beendet; Absagen ohne Sitzung und ohne Projekt | +| `src/remote/remoteBlobs.ts`, `src/remote/SharedProject.tsx` (Blob-Seite), `src/desktop/fileExplorer.ts` (Bildbetrachter), `src/i18n/*.json` | Plan §5.1 | neu (F5.5): die Blob-Seite des Editors auf F — was das Dokument nennt, wird geholt (mit dem Keks; was F noch nicht hat, ein paar Mal im Abstand neu gefragt), was hier entsteht (eine importierte Datei), sofort gehalten und zu F gebracht (einmal wiederholt, dann gesagt); der Bildbetrachter zeichnet neu, wenn die Bytes kommen | +| `test/remoteBlobs.test.ts` | — | neu (F5.5): Hochladen, Wiederholen, Scheitern (auch bei falschem Namen), Holen je Blob einmal, nachfragen bei 404, Netzverlust | +| `test/remoteSimHost.test.ts`, `browser/shared.spec.ts`, `browser/serverF.ts` | — | neu (F5.3b), erweitert (F5.5: ein großes Bild, von Mia importiert, bei Ben im Bildbetrachter, in F's Speicher unter seinem SHA-256): `RemoteSimHost` auf nachgebauten Sockets mit gestellter Uhr; drei Personen in drei Browsern an einem echten F (aus dem Bündel gestartet): verschieben, nur lesen, eine Simulation für alle mit Namen, ein Ping in einer Befehlszeile bei der anderen, Eintippen als Leser abgewiesen, eingefroren und aufgetaut, Zugang beendet; Absagen ohne Sitzung und ohne Projekt | ## packages/server @@ -278,6 +282,7 @@ Swift-Pfade relativ zu `filius-ipad/ios/FiliusPad/TopologyEditor/` (Stand | `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` | | `src/sync.ts` (Zugang gleich zu Beginn, stille Absage) | Plan §8.5 | geändert (F5.3b): jede Verbindung erfährt ihren Zugang sofort nach der Anmeldung (Nachricht `access`), damit der Editor weiß, ob er liest und warum; eine Absage trägt keinen Meldungstext — Hocuspocus schreibt sonst die Meldung eines gescheiterten Hooks auf die Konsole, außerhalb des JSON-Protokolls | | `src/sync.ts` (`you`), `protocol/src/sync.ts` (`AccessMessage.you`) | Plan §7.4 | erweitert (F5.4): die Zugangsnachricht beim Verbinden nennt die Person (Kennung, Name, Farbe aus der Karte), damit der Editor sich den anderen zeigen kann — der Keks ist für die Seite nicht lesbar | +| `src/blobs.ts`, `src/blobRoutes.ts`, `src/simWorker.ts` (Blobs), `src/simRooms.ts` (Blobs), `src/simChannel.ts`, `src/server.ts` (`/blobs`), `src/main.ts`, `test/blobs.test.ts` | Plan §5.1 | neu (F5.5): die Blobs auf F, inhaltsadressiert (`filius_blob`, F rechnet den Namen selbst aus); `GET /blobs/?project=` mit Berechtigung am Projekt, `POST /blobs?project=` mit Schreibstufe, nicht eingefroren, nur von der Herkunft des Editors, höchstens 16 MiB; CORS für die Herkünfte des Editors; der Raumfaden fragt nach den Blobs seines Dokuments, ein Start wartet auf die gefragten, was F noch nicht hat, wird ein paar Mal neu gefragt und kommt in die laufende Simulation; ein Blob, den die Simulation macht, wird gespeichert, bevor die Änderung, die ihn nennt, angewandt wird. Noch nicht: Blobs vergessen, die kein Projekt mehr nennt | | `test/editorConnection.test.ts` | — | neu (F5.3b), erweitert (F5.4: Identität von F, die anderen über die Awareness, auch eine lesende Person, weg mit dem Schließen): `connectSharedDoc` des Editors gegen einen echten F — bereit mit Schreib- und Lesestufe, eingefroren und aufgetaut (neues Dokument, die verirrte Änderung ist weg, Schreiben geht wieder), eingefroren von Beginn an, Absage, Widerruf, Löschen, Loslassen | | `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 | diff --git a/package-lock.json b/package-lock.json index dab6c27..44840a3 100644 --- a/package-lock.json +++ b/package-lock.json @@ -895,6 +895,18 @@ "node": "^22.20 || ^24.12 || >=25" } }, + "node_modules/@noble/hashes": { + "version": "2.4.0", + "resolved": "https://registry.npmjs.org/@noble/hashes/-/hashes-2.4.0.tgz", + "integrity": "sha512-X5XaVWZIBCT7HHZGm5I7ZQXDwLG+bGXuSrMQAW+7Zvl87h1kmc1ZB1VSRJcpUfoUrGQp4Fkoxm5kZ+Ms+aW+eA==", + "license": "MIT", + "engines": { + "node": ">= 20.19.0" + }, + "funding": { + "url": "https://paulmillr.com/funding/" + } + }, "node_modules/@pdf-lib/standard-fonts": { "version": "1.0.0", "resolved": "https://registry.npmjs.org/@pdf-lib/standard-fonts/-/standard-fonts-1.0.0.tgz", @@ -2843,6 +2855,7 @@ "dependencies": { "@mdg-filius/core": "0.0.0", "@mdg-filius/fls": "0.0.0", + "@noble/hashes": "2.4.0", "yjs": "^13.6.33" } }, diff --git a/packages/doc/package.json b/packages/doc/package.json index 9691bce..ceb7ca9 100644 --- a/packages/doc/package.json +++ b/packages/doc/package.json @@ -10,6 +10,7 @@ "dependencies": { "@mdg-filius/core": "0.0.0", "@mdg-filius/fls": "0.0.0", + "@noble/hashes": "2.4.0", "yjs": "^13.6.33" } } diff --git a/packages/doc/src/blobs.ts b/packages/doc/src/blobs.ts new file mode 100644 index 0000000..4ca9082 --- /dev/null +++ b/packages/doc/src/blobs.ts @@ -0,0 +1,95 @@ +// A computer's large binary files outside the document (plan §5.1: no +// binary in the CRDT): from BLOB_THRESHOLD_BYTES on, a picture's or other +// binary file's entry holds `blob` — the SHA-256 of its bytes, in hex — and +// `size` instead of `data`; the bytes are on F, content-addressed. Text +// stays Y.Text whatever its size: two type into it together. Below the +// threshold a binary stays in the entry, which spares a small icon a fetch. +// +// Where the bytes are is a document's blob side (attachBlobs): the editor's +// on F (it fetches and uploads), the room thread's (it asks F's main +// thread). A document without one — the development page, the tests, the +// room in a worker of the page — keeps every binary in its entry. Reading a +// blob whose bytes have not come yet gives no bytes; whoever needs them all +// asks for the missing ones first (missingBlobs) and hears them come +// (onBlobsArrive). + +import { sha256 } from '@noble/hashes/sha2.js'; +import type * as Y from 'yjs'; +import { ROOTS } from './projectDoc.js'; + +export const BLOB_THRESHOLD_BYTES = 16 * 1024; + +export interface BlobSide { + /** A blob's bytes, if this side has them. */ + bytes(sha: string): Uint8Array | undefined; + /** Bytes become a blob: kept here and taken to F. */ + keep(sha: string, bytes: Uint8Array): void; + /** Hears blobs come; returns how to stop. */ + onArrive(listener: () => void): () => void; +} + +const sides = new WeakMap(); + +/** Where a document's blobs are, from now on. */ +export function attachBlobs(doc: Y.Doc, side: BlobSide): void { + sides.set(doc, side); +} + +export function blobSideOf(doc: Y.Doc | null | undefined): BlobSide | undefined { + return doc === null || doc === undefined ? undefined : sides.get(doc); +} + +const hex = (bytes: Uint8Array) => Array.from(bytes, (byte) => byte.toString(16).padStart(2, '0')).join(''); + +/** A blob's name: the SHA-256 of its bytes, in hex. */ +export function blobSha(bytes: Uint8Array): string { + return hex(sha256(bytes)); +} + +export const isBlobSha = (value: unknown): value is string => typeof value === 'string' && /^[0-9a-f]{64}$/.test(value); + +/** Every blob the document's files name. */ +export function blobsOf(doc: Y.Doc): Set { + const found = new Set(); + for (const node of doc.getMap>(ROOTS.nodes).values()) { + const files = node.get('files') as Y.Map> | undefined; + for (const entry of files?.values() ?? []) { + const sha = entry.get('blob'); + if (isBlobSha(sha)) found.add(sha); + } + } + return found; +} + +/** The blobs the document names whose bytes its side has not got (all, without a side). */ +export function missingBlobs(doc: Y.Doc): string[] { + const side = blobSideOf(doc); + return [...blobsOf(doc)].filter((sha) => side?.bytes(sha) === undefined); +} + +/** Hears the document's blobs come (nothing without a side); returns how to stop. */ +export function onBlobsArrive(doc: Y.Doc, listener: () => void): () => void { + return blobSideOf(doc)?.onArrive(listener) ?? (() => {}); +} + +/** A side that keeps blobs in memory and tells when some come (`arrived`); `taken` hears each kept one. */ +export function memoryBlobSide(taken: (sha: string, bytes: Uint8Array) => void = () => {}): BlobSide & { arrived(blobs: ReadonlyMap): void } { + const held = new Map(); + const listeners = new Set<() => void>(); + return { + bytes: (sha) => held.get(sha), + keep(sha, bytes) { + if (held.has(sha)) return; + held.set(sha, bytes); + taken(sha, bytes); + }, + onArrive(listener) { + listeners.add(listener); + return () => listeners.delete(listener); + }, + arrived(blobs) { + for (const [sha, bytes] of blobs) held.set(sha, bytes); + if (blobs.size > 0) for (const listener of listeners) listener(); + }, + }; +} diff --git a/packages/doc/src/commands.ts b/packages/doc/src/commands.ts index 0a08bd3..a361a87 100644 --- a/packages/doc/src/commands.ts +++ b/packages/doc/src/commands.ts @@ -29,6 +29,7 @@ import { } from '@mdg-filius/core'; import { DEFAULT_MODEM_PORT, modemPair, type ImportedProject } from '@mdg-filius/fls'; import { emptyProject, inOrder, nodeEntry, ROOTS } from './projectDoc.js'; +import { blobSideOf } from './blobs.js'; /** The kinds of the palette, in Java's order (GUIDesignSidebar.addComponentItemsToSidebar; the cable comes first, a tool). */ export const PALETTE: readonly Exclude[] = ['pc', 'notebook', 'networkSwitch', 'gateway', 'router', 'remoteLink']; @@ -120,7 +121,7 @@ export function addDevice(doc: Y.Doc, device: NewDevice, origin: unknown): void const project = newDeviceProject(device); doc.transact(() => { const order = Math.max(-1, ...[...nodes(doc).values()].map((node) => node.get('order') as number)) + 1; - nodes(doc).set(device.id, nodeEntry(project.graph.nodes[0]!, order, project)); + nodes(doc).set(device.id, nodeEntry(project.graph.nodes[0]!, order, project, blobSideOf(doc))); }, origin); } diff --git a/packages/doc/src/files.ts b/packages/doc/src/files.ts index bde3107..f794492 100644 --- a/packages/doc/src/files.ts +++ b/packages/doc/src/files.ts @@ -11,6 +11,7 @@ import * as Y from 'yjs'; import type { Id, VirtualFileContent, VirtualFileSystem } from '@mdg-filius/core'; import { VirtualFileSystem as FileSystem } from '@mdg-filius/core'; +import { blobSha, blobSideOf } from './blobs.js'; import { fileEntry, fileSystemOf, inOrder, ROOTS } from './projectDoc.js'; export function sameBytes(left: Uint8Array, right: Uint8Array): boolean { @@ -29,11 +30,21 @@ export function replaceText(text: Y.Text, next: string): void { if (next.length - start - end > 0) text.insert(start, next.slice(start, next.length - end)); } -/** Whether a document entry holds the content. */ +/** + * Whether a document entry holds the content (a blob by its bytes where they + * are here, else by its name). + */ function holds(entry: Y.Map, content: VirtualFileContent): boolean { if (entry.get('kind') !== content.kind) return false; if (content.kind === 'directory' || content.kind === 'text') return true; - return sameBytes(entry.get('data') as Uint8Array, content.data) && entry.get('mediaType') === content.mediaType; + if (entry.get('mediaType') !== content.mediaType) return false; + const blob = entry.get('blob'); + if (typeof blob !== 'string') return sameBytes(entry.get('data') as Uint8Array, content.data); + const held = blobSideOf(entry.doc)?.bytes(blob); + // A blob whose bytes are not here was read as no bytes: written back so, that is no change (it would empty the file). + if (held === undefined && content.data.length === 0) return true; + if (entry.get('size') !== content.data.length) return false; + return held === undefined ? blobSha(content.data) === blob : sameBytes(held, content.data); } /** @@ -42,6 +53,7 @@ function holds(entry: Y.Map, content: VirtualFileContent): boolean { * longer the file system's (a renamed entry), numbered anew. */ export function writeFiles(node: Y.Map, fileSystem: VirtualFileSystem): void { + const blobs = blobSideOf(node.doc); let files = node.get('files') as Y.Map> | undefined; const wanted: [string, VirtualFileContent][] = []; const walk = (folder: string): void => { @@ -62,9 +74,9 @@ export function writeFiles(node: Y.Map, fileSystem: VirtualFileSystem): for (const [path, content] of wanted) { const entry = files.get(path); if (entry === undefined) { - files.set(path, fileEntry(order++, content)); + files.set(path, fileEntry(order++, content, blobs)); } else if (!holds(entry, content)) { - files.set(path, fileEntry(entry.get('order') as number, content)); + files.set(path, fileEntry(entry.get('order') as number, content, blobs)); } else if (content.kind === 'text') { replaceText(entry.get('text') as Y.Text, content.text); } diff --git a/packages/doc/src/index.ts b/packages/doc/src/index.ts index 52c9097..0be74bc 100644 --- a/packages/doc/src/index.ts +++ b/packages/doc/src/index.ts @@ -1,4 +1,5 @@ // Public surface of the shared document. +export * from './blobs.js'; export * from './commands.js'; export * from './documentation.js'; export * from './fields.js'; diff --git a/packages/doc/src/projectDoc.ts b/packages/doc/src/projectDoc.ts index 6f1c5ab..58705b5 100644 --- a/packages/doc/src/projectDoc.ts +++ b/packages/doc/src/projectDoc.ts @@ -35,6 +35,7 @@ import { } from '@mdg-filius/core'; import type { FlsOwnApplication, ImportedProject } from '@mdg-filius/fls'; import { fromY, toY, type PlainValue } from './values.js'; +import { BLOB_THRESHOLD_BYTES, blobSha, blobSideOf, type BlobSide } from './blobs.js'; export const SCHEMA_VERSION = 1; @@ -98,23 +99,29 @@ function ownApplications(text: string): FlsOwnApplication[] { ) as FlsOwnApplication[]; } -export function fileEntry(order: number, content: VirtualFileContent): Y.Map { +/** A file's entry; a large binary as a blob when the document has a blob side (blobs.ts). */ +export function fileEntry(order: number, content: VirtualFileContent, blobs?: BlobSide): Y.Map { const entry = new Y.Map(); entry.set('order', order); entry.set('kind', content.kind); if (content.kind === 'text') entry.set('text', new Y.Text(content.text)); - if (content.kind === 'binary' || content.kind === 'image') entry.set('data', content.data); + if ((content.kind === 'binary' || content.kind === 'image') && blobs !== undefined && content.data.length >= BLOB_THRESHOLD_BYTES) { + const sha = blobSha(content.data); + blobs.keep(sha, content.data); + entry.set('blob', sha); + entry.set('size', content.data.length); + } else if (content.kind === 'binary' || content.kind === 'image') entry.set('data', content.data); if ((content.kind === 'binary' || content.kind === 'image') && content.mediaType !== undefined) entry.set('mediaType', content.mediaType); return entry; } /** A device's files, each folder's entries in the order they were made (parents before their entries). */ -function filesMap(files: VirtualFileSystem): Y.Map> { +function filesMap(files: VirtualFileSystem, blobs: BlobSide | undefined): Y.Map> { const map = new Y.Map>(); let order = 0; const walk = (folder: string): void => { for (const entry of files.entriesInCreationOrder(folder)) { - map.set(entry.path, fileEntry(order++, entry.content)); + map.set(entry.path, fileEntry(order++, entry.content, blobs)); if (entry.content.kind === 'directory') walk(entry.path); } }; @@ -123,7 +130,7 @@ function filesMap(files: VirtualFileSystem): Y.Map> { } /** A device's entry in the document, with what the project holds for it. */ -export function nodeEntry(node: TopologyNode, order: number, project: ImportedProject): Y.Map { +export function nodeEntry(node: TopologyNode, order: number, project: ImportedProject, blobs?: BlobSide): Y.Map { const map = new Y.Map(); map.set('order', order); map.set('kind', node.kind); @@ -159,7 +166,7 @@ export function nodeEntry(node: TopologyNode, order: number, project: ImportedPr } map.set('programs', programs); const files = project.virtualFileSystems.get(node.id); - if (files !== undefined) map.set('files', filesMap(files)); + if (files !== undefined) map.set('files', filesMap(files, blobs)); const own = project.ownApplicationsByNodeId.get(node.id); if (own !== undefined) map.set('own', ownText(own)); return map; @@ -202,7 +209,7 @@ export function writeProject(doc: Y.Doc, project: ImportedProject, origin: unkno meta.set('schema', SCHEMA_VERSION); meta.set('versionLine', project.versionLine); const nodes = doc.getMap>(ROOTS.nodes); - project.graph.nodes.forEach((node, index) => nodes.set(node.id, nodeEntry(node, index, project))); + project.graph.nodes.forEach((node, index) => nodes.set(node.id, nodeEntry(node, index, project, blobSideOf(doc)))); const links = doc.getMap>(ROOTS.links); project.graph.links.forEach((link, index) => links.set(link.id, linkMap(link, index))); const docs = doc.getMap>(ROOTS.docs); @@ -225,6 +232,15 @@ export function inOrder(map: Y.Map>): [string, Y.Map][] * system refuses (its folder missing before the janitor ran, or a name two * people gave at once in different case) is left out, not the project. */ +const NO_BYTES = new Uint8Array(0); + +/** A binary entry's bytes: its own, or its blob's (none while they have not come). */ +export function bytesOf(entry: Y.Map): Uint8Array { + const blob = entry.get('blob'); + if (typeof blob === 'string') return blobSideOf(entry.doc)?.bytes(blob) ?? NO_BYTES; + return entry.get('data') as Uint8Array; +} + export function fileSystemOf(files: Y.Map>): VirtualFileSystem { const fileSystem = new VirtualFileSystem(); for (const [path, entry] of inOrder(files)) { @@ -233,8 +249,8 @@ export function fileSystemOf(files: Y.Map>): VirtualFileSystem { try { if (kind === 'directory') fileSystem.createDirectory(path); else if (kind === 'text') fileSystem.writeTextFile(path, (entry.get('text') as Y.Text).toString()); - else if (kind === 'image') fileSystem.writeImageFile(path, entry.get('data') as Uint8Array, mediaType!); - else fileSystem.writeBinaryFile(path, entry.get('data') as Uint8Array, mediaType); + else if (kind === 'image') fileSystem.writeImageFile(path, bytesOf(entry), mediaType!); + else fileSystem.writeBinaryFile(path, bytesOf(entry), mediaType); } catch { // Left out (see above). } diff --git a/packages/doc/test/blobs.test.ts b/packages/doc/test/blobs.test.ts new file mode 100644 index 0000000..915904e --- /dev/null +++ b/packages/doc/test/blobs.test.ts @@ -0,0 +1,158 @@ +// A computer's large binary files as blobs (plan §5.1): with a blob side, a +// picture or binary from 16 KB on is named by its SHA-256 in its entry and +// kept by the side, a smaller one stays in its entry, a text stays Y.Text +// whatever its size; without a side everything stays in the document as +// before. Read back with the side's bytes; a copy of the document whose +// side has not got them yet reads no bytes, names them as missing, and +// hears them come. Written again unchanged, nothing changes. + +import { readFileSync } from 'node:fs'; +import { describe, expect, it } from 'vitest'; +import * as Y from 'yjs'; +import { VirtualFileSystem } from '@mdg-filius/core'; +import { openProjectArchive } from '@mdg-filius/fls'; +import { + attachBlobs, + BLOB_THRESHOLD_BYTES, + blobsOf, + blobSha, + changeFiles, + emptyProject, + isBlobSha, + memoryBlobSide, + missingBlobs, + nodeFiles, + onBlobsArrive, + writeFiles, + writeProject, + ROOTS, +} from '../src/index.js'; + +const bytes = (length: number, seed = 1) => Uint8Array.from({ length }, (_, index) => (index * 31 + seed) % 251); + +/** A document with one computer, its files as given, and a side (when given) attached before it is written. */ +function docWith(files: (fs: VirtualFileSystem) => void, side?: ReturnType): { doc: Y.Doc; node: string } { + const doc = new Y.Doc(); + if (side !== undefined) attachBlobs(doc, side); + writeProject(doc, emptyProject()); + const node = 'pc'; + doc.transact(() => { + const entry = new Y.Map(); + doc.getMap>(ROOTS.nodes).set(node, entry); + const fs = new VirtualFileSystem(); + files(fs); + writeFiles(entry, fs); + }); + return { doc, node }; +} + +const entryOf = (doc: Y.Doc, node: string, path: string) => (doc.getMap>(ROOTS.nodes).get(node)!.get('files') as Y.Map>).get(path)!; + +describe('blobs', () => { + it('names a blob by the SHA-256 of its bytes, in hex', () => { + expect(blobSha(new TextEncoder().encode('abc'))).toBe('ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad'); + expect([isBlobSha(blobSha(bytes(3))), isBlobSha('ABC'), isBlobSha(7)]).toEqual([true, false, false]); + expect(BLOB_THRESHOLD_BYTES).toBe(16 * 1024); + }); + + it('keeps a large picture and a large binary as blobs, a small one and every text in the document', () => { + const taken: string[] = []; + const side = memoryBlobSide((sha) => taken.push(sha)); + const big = bytes(BLOB_THRESHOLD_BYTES); + const { doc, node } = docWith((fs) => { + fs.writeImageFile('/big.png', big, 'image/png'); + fs.writeBinaryFile('/big.bin', bytes(BLOB_THRESHOLD_BYTES + 5, 2), 'application/zip'); + fs.writeBinaryFile('/small.bin', bytes(BLOB_THRESHOLD_BYTES - 1, 3)); + fs.writeTextFile('/long.txt', 'x'.repeat(BLOB_THRESHOLD_BYTES * 2)); + }, side); + const picture = entryOf(doc, node, '/big.png'); + expect([picture.get('blob'), picture.get('size'), picture.has('data'), picture.get('mediaType')]).toEqual([blobSha(big), big.length, false, 'image/png']); + expect(entryOf(doc, node, '/big.bin').get('blob')).toBe(blobSha(bytes(BLOB_THRESHOLD_BYTES + 5, 2))); + expect(entryOf(doc, node, '/small.bin').get('data')).toEqual(bytes(BLOB_THRESHOLD_BYTES - 1, 3)); + expect(entryOf(doc, node, '/long.txt').get('text')).toBeInstanceOf(Y.Text); + expect(taken.sort()).toEqual([blobSha(big), blobSha(bytes(BLOB_THRESHOLD_BYTES + 5, 2))].sort()); + // Read back with the side: the bytes. + const read = nodeFiles(doc, node)!; + expect(read.entry('/big.png')!.content).toEqual({ kind: 'image', data: big, mediaType: 'image/png' }); + expect(blobsOf(doc)).toEqual(new Set(taken)); + expect(missingBlobs(doc)).toEqual([]); + }); + + it('keeps the large pictures of a whole project as blobs when it is written with a side', () => { + const big = bytes(BLOB_THRESHOLD_BYTES * 2, 4); + const project = openProjectArchive(new Uint8Array(readFileSync(new URL('../../../corpus/java-examples/zwei_rechnernetze_komplett.fls', import.meta.url))), () => crypto.randomUUID()); + const [node, files] = [...project.virtualFileSystems][0]!; + files.writeImageFile('/big.png', big, 'image/png'); + const taken: string[] = []; + const doc = new Y.Doc(); + attachBlobs( + doc, + memoryBlobSide((sha) => taken.push(sha)), + ); + writeProject(doc, project); + expect(entryOf(doc, node, '/big.png').get('blob')).toBe(blobSha(big)); + expect(taken).toEqual([blobSha(big)]); + }); + + it('keeps everything in the document without a side', () => { + const big = bytes(BLOB_THRESHOLD_BYTES * 2); + const { doc, node } = docWith((fs) => fs.writeImageFile('/big.png', big, 'image/png')); + expect(entryOf(doc, node, '/big.png').get('data')).toEqual(big); + expect(entryOf(doc, node, '/big.png').has('blob')).toBe(false); + expect(nodeFiles(doc, node)!.entry('/big.png')!.content).toEqual({ kind: 'image', data: big, mediaType: 'image/png' }); + expect(missingBlobs(doc)).toEqual([]); + expect(onBlobsArrive(doc, () => undefined)).toBeTypeOf('function'); + }); + + it('reads no bytes in a copy whose side has not got them, names them missing, and hears them come', () => { + const big = bytes(BLOB_THRESHOLD_BYTES * 2); + const { doc, node } = docWith((fs) => fs.writeImageFile('/big.png', big, 'image/png'), memoryBlobSide()); + const copy = new Y.Doc(); + const side = memoryBlobSide(); + attachBlobs(copy, side); + Y.applyUpdate(copy, Y.encodeStateAsUpdate(doc)); + expect(nodeFiles(copy, node)!.entry('/big.png')!.content).toEqual({ kind: 'image', data: new Uint8Array(0), mediaType: 'image/png' }); + expect(missingBlobs(copy)).toEqual([blobSha(big)]); + let heard = 0; + const stop = onBlobsArrive(copy, () => (heard += 1)); + side.arrived(new Map()); + side.arrived(new Map([[blobSha(big), big]])); + stop(); + side.arrived(new Map([[blobSha(big), big]])); + expect(heard).toBe(1); + expect(missingBlobs(copy)).toEqual([]); + expect(nodeFiles(copy, node)!.entry('/big.png')!.content).toMatchObject({ data: big }); + // A copy without a side names every blob missing. + const bare = new Y.Doc(); + Y.applyUpdate(bare, Y.encodeStateAsUpdate(doc)); + expect(missingBlobs(bare)).toEqual([blobSha(big)]); + }); + + it('changes nothing when the same files are written again, with the bytes here or only their name', () => { + const big = bytes(BLOB_THRESHOLD_BYTES * 2); + const side = memoryBlobSide(); + const { doc, node } = docWith((fs) => fs.writeImageFile('/big.png', big, 'image/png'), side); + const copy = new Y.Doc(); + attachBlobs(copy, memoryBlobSide()); + Y.applyUpdate(copy, Y.encodeStateAsUpdate(doc)); + for (const each of [doc, copy]) { + let updates = 0; + each.on('update', () => (updates += 1)); + changeFiles(each, node, 'person', (fs) => fs.writeImageFile('/big.png', big, 'image/png')); + expect(updates).toBe(0); + } + // Other bytes of the same size, another type, another size: a new entry each. + let updates = 0; + doc.on('update', () => (updates += 1)); + changeFiles(doc, node, 'person', (fs) => fs.writeImageFile('/big.png', bytes(big.length, 9), 'image/png')); + expect(entryOf(doc, node, '/big.png').get('blob')).toBe(blobSha(bytes(big.length, 9))); + changeFiles(doc, node, 'person', (fs) => fs.writeImageFile('/big.png', bytes(big.length, 9), 'image/gif')); + expect(entryOf(doc, node, '/big.png').get('mediaType')).toBe('image/gif'); + changeFiles(doc, node, 'person', (fs) => fs.writeImageFile('/big.png', bytes(big.length + 1, 9), 'image/gif')); + expect(entryOf(doc, node, '/big.png').get('size')).toBe(big.length + 1); + expect(updates).toBe(3); + // A copy without the bytes compares by the name. + changeFiles(copy, node, 'person', (fs) => fs.writeImageFile('/big.png', bytes(big.length, 5), 'image/png')); + expect(entryOf(copy, node, '/big.png').get('blob')).toBe(blobSha(bytes(big.length, 5))); + }); +}); diff --git a/packages/editor/browser/serverF.ts b/packages/editor/browser/serverF.ts index 427eea3..5bbb240 100644 --- a/packages/editor/browser/serverF.ts +++ b/packages/editor/browser/serverF.ts @@ -29,6 +29,8 @@ export interface ServerF { /** Signs a person in, in this browser context. */ signIn(context: BrowserContext, project: string, person: { readonly name: string; readonly level?: 'read' | 'write'; readonly user?: string }): Promise; service(method: string, path: string, body?: unknown): Promise; + /** The size of a blob in F's store, none without it. */ + blobSize(sha: string): Promise; close(): Promise; } @@ -88,6 +90,10 @@ export async function startServerF(editorBase: string): Promise { if (response?.status() !== 200) throw new Error(`not signed in: ${response?.status()}`); await page.close(); }, + async blobSize(sha) { + const result = await sql.query<{ size: number }>('SELECT size FROM filius_blob WHERE sha = $1', [sha]); + return result.rows[0]?.size; + }, service: (method, path, body) => fetch(`${base}${path}`, { method, headers: { 'x-filius-service-key': SERVICE_KEY }, ...(body === undefined ? {} : { body: JSON.stringify(body) }) }), async close() { const ended = new Promise((resolve) => child.once('exit', resolve)); diff --git a/packages/editor/browser/shared.spec.ts b/packages/editor/browser/shared.spec.ts index d700ef8..26ec877 100644 --- a/packages/editor/browser/shared.spec.ts +++ b/packages/editor/browser/shared.spec.ts @@ -9,7 +9,12 @@ // told, thawed he writes again; his grant ended, he is told and the editor // goes. +import { createHash } from 'node:crypto'; +import { mkdtempSync, readFileSync, statSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; import { expect, test, type Browser, type Page } from '@playwright/test'; +import { BLOB_THRESHOLD_BYTES } from '@mdg-filius/doc'; import { drag, onDevice, serveEditor } from './editorPage.js'; import { colorOf } from '../src/remote/presence.js'; import { startServerF, type ServerF } from './serverF.js'; @@ -158,6 +163,51 @@ test('three people in one project of F: moving, reading only, one simulation for await expect(canvas(mia)).toBeVisible(); }); +test('takes a large picture one person imports to F as a blob, and shows it to another', async ({ browser }) => { + test.setTimeout(120_000); + const project = await f.project('java-examples/internetwork_komplett.fls'); + const notebook = project.node('Notebook 1.10'); + const mia = await person(browser, project.id, 'Mia'); + const ben = await person(browser, project.id, 'Ben'); + // A picture far over 16 KB: a screenshot of the page. + const file = join(mkdtempSync(join(tmpdir(), 'filius-blob-')), 'bild.png'); + await mia.screenshot({ path: file }); + const size = statSync(file).size; + expect(size).toBeGreaterThan(BLOB_THRESHOLD_BYTES); + await mia.locator('button[name="simulationMode"]').click(); + for (const page of [mia, ben]) await expect(canvas(page)).toHaveAttribute('data-mode', 'simulation'); + const open = async (page: Page) => { + const at = await onDevice(page, notebook); + await page.mouse.click(at.x, at.y); + return desktop(page, notebook); + }; + const miaDesktop = await open(mia); + await miaDesktop.locator('[data-program="installation"]').click(); + const installation = miaDesktop.locator('[data-program-window="installation"]'); + for (const program of ['Datei-Explorer', 'Bildbetrachter']) await installation.locator('[data-list="available"] [role="option"]').filter({ hasText: program }).click(); + await installation.getByRole('button', { name: 'Installieren' }).click(); + await installation.getByRole('button', { name: 'Änderungen annehmen' }).click(); + await miaDesktop.locator('[data-program="fileExplorer"]').click(); + const explorer = miaDesktop.locator('[data-program-window="fileExplorer"]'); + await explorer.getByRole('toolbar', { name: 'Aktionen' }).getByRole('button', { name: 'Importieren' }).click(); + const importing = explorer.getByRole('dialog', { name: 'Datei importieren' }); + await importing.locator('input[type="file"]').setInputFiles(file); + await importing.getByRole('button', { name: 'Datei importieren' }).click(); + await expect(importing.getByRole('status')).toHaveText('Datei erfolgreich importiert!'); + // Ben opens it in his viewer: the picture, as large as the page was. + const benDesktop = await open(ben); + await benDesktop.locator('[data-program="imageViewer"]').click(); + const viewer = benDesktop.locator('[data-program-window="imageViewer"]'); + await viewer.getByRole('button', { name: 'Öffnen' }).click(); + const dialog = viewer.getByRole('dialog', { name: 'Datei Öffnen' }); + await dialog.locator('.file-dialog-entries button', { hasText: 'bild.png' }).click(); + await dialog.getByRole('button', { name: 'Öffnen' }).click(); + await expect.poll(() => viewer.getByRole('img', { name: 'Bild' }).evaluate((image: HTMLImageElement) => image.naturalWidth)).toBe(mia.viewportSize()!.width); + // In F's store under its SHA-256. + const sha = createHash('sha256').update(readFileSync(file)).digest('hex'); + expect(await f.blobSize(sha)).toBe(size); +}); + test('tells the dashboard over the bridge who else is in the project, each time it changes', async ({ browser }) => { const project = await f.project('java-examples/zwei_rechnernetze_komplett.fls'); const context = await browser.newContext(); diff --git a/packages/editor/src/desktop/fileExplorer.ts b/packages/editor/src/desktop/fileExplorer.ts index c3323ca..60ffb39 100644 --- a/packages/editor/src/desktop/fileExplorer.ts +++ b/packages/editor/src/desktop/fileExplorer.ts @@ -41,7 +41,7 @@ import { type FolderEntry, type Id, } from '@mdg-filius/core'; -import { changeFiles, nodeFiles } from '@mdg-filius/doc'; +import { changeFiles, nodeFiles, onBlobsArrive } from '@mdg-filius/doc'; import { WindowState, windowOf } from './windowState.js'; /** A row of the explorer's tree: every folder and file, each folder's entries after it in the order they were made. */ @@ -214,6 +214,8 @@ export class ImageViewerWindow extends WindowState { ) { super(); doc.on('update', () => this.pathValue !== null && this.changed()); + // A picture that is a blob shows once its bytes have come. + onBlobsArrive(doc, () => this.pathValue !== null && this.changed()); } private files(): VirtualFileSystem { diff --git a/packages/editor/src/i18n/de.json b/packages/editor/src/i18n/de.json index 1b9ee03..52e548d 100644 --- a/packages/editor/src/i18n/de.json +++ b/packages/editor/src/i18n/de.json @@ -367,6 +367,7 @@ "remote.refused": "Der Filius-Dienst hat die Verbindung abgelehnt ({reason}).", "remote.revoked": "Ihr Zugang zu diesem Projekt wurde beendet.", "remote.unknownProject": "Dieses Projekt gibt es nicht mehr.", + "remote.uploadFailed": "Eine Datei konnte nicht zum Filius-Dienst gebracht werden; andere sehen sie nicht. Laden Sie die Seite neu und versuchen Sie es noch einmal.", "report.apps": "Anwendungen", "report.baseConfig": "Basiskonfiguration", "report.components": "Komponenten", diff --git a/packages/editor/src/i18n/en.json b/packages/editor/src/i18n/en.json index ecf0a73..c2370e7 100644 --- a/packages/editor/src/i18n/en.json +++ b/packages/editor/src/i18n/en.json @@ -367,6 +367,7 @@ "remote.refused": "The Filius service refused the connection ({reason}).", "remote.revoked": "Your access to this project has ended.", "remote.unknownProject": "This project no longer exists.", + "remote.uploadFailed": "A file could not be taken to the Filius service; others do not see it. Reload the page and try again.", "report.apps": "Applications", "report.baseConfig": "Base Configuration", "report.components": "Components", diff --git a/packages/editor/src/remote/SharedProject.tsx b/packages/editor/src/remote/SharedProject.tsx index 88ad6dd..b7c4dc8 100644 --- a/packages/editor/src/remote/SharedProject.tsx +++ b/packages/editor/src/remote/SharedProject.tsx @@ -4,9 +4,14 @@ // refuses or lets go — of the document or the room — a word why, instead of // the editor. Who only reads gets the editor locked (Editor's readOnly). // The others in the project are shown in the editor and told to whoever -// embeds it (`onPeople`: the dashboard's header, over the bridge). +// embeds it (`onPeople`: the dashboard's header, over the bridge). The +// document's large binary files come from F and go to it (remoteBlobs): +// fetched as the document names them; one that could not be taken to F is +// said. import { useEffect, useMemo, useState } from 'react'; +import { attachBlobs } from '@mdg-filius/doc'; +import { remoteBlobs, type RemoteBlobs } from './remoteBlobs.js'; import { peopleOf, type PresenceUser } from './presence.js'; import { usePresence } from './usePresence.js'; import { Editor } from '../Editor.js'; @@ -18,7 +23,25 @@ import { connectSharedDoc, type SharedDocState } from './sharedDoc.js'; type Ready = Extract; /** The editor on the ready document, with the others in the project. */ -function ReadyEditor({ state, createSimHost, onPeople }: { readonly state: Ready; readonly createSimHost: SimHostFactory; readonly onPeople: ((people: readonly PresenceUser[]) => void) | undefined }) { +function ReadyEditor({ + state, + createSimHost, + onPeople, + blobs, +}: { + readonly state: Ready; + readonly createSimHost: SimHostFactory; + readonly onPeople: ((people: readonly PresenceUser[]) => void) | undefined; + readonly blobs: RemoteBlobs; +}) { + // The blob side before anything reads or writes a file of the document. + useMemo(() => attachBlobs(state.doc, blobs), [state.doc, blobs]); + useEffect(() => { + const fetchMissing = () => void blobs.fetchMissing(state.doc); + fetchMissing(); + state.doc.on('update', fetchMissing); + return () => state.doc.off('update', fetchMissing); + }, [state.doc, blobs]); const presence = usePresence(state.awareness, state.you); const key = JSON.stringify(peopleOf(presence.peers, state.you.id)); useEffect(() => onPeople?.(JSON.parse(key) as PresenceUser[]), [key, onPeople]); @@ -39,6 +62,8 @@ export function SharedProject({ server, project, onPeople }: SharedProjectProps) const [state, setState] = useState({ kind: 'connecting' }); /** F ended the room for this page: the editor goes as when the document is let go. */ const [roomEnd, setRoomEnd] = useState(null); + const [uploadFailed, setUploadFailed] = useState(false); + const blobs = useMemo(() => remoteBlobs({ server, project, onFailure: () => setUploadFailed(true) }), [server, project]); useEffect(() => connectSharedDoc({ server, project, onState: setState }), [server, project]); const createSimHost = useMemo( () => () => { @@ -63,5 +88,14 @@ export function SharedProject({ server, project, onPeople }: SharedProjectProps) {t('remote.connecting')}

); - return ; + return ( + <> + {uploadFailed ? ( +

+ {t('remote.uploadFailed')} +

+ ) : null} + + + ); } diff --git a/packages/editor/src/remote/remoteBlobs.ts b/packages/editor/src/remote/remoteBlobs.ts new file mode 100644 index 0000000..26f7ff6 --- /dev/null +++ b/packages/editor/src/remote/remoteBlobs.ts @@ -0,0 +1,89 @@ +// The editor's blob side on F (packages/doc's blobs.ts): the blobs the +// document names are fetched from F as they come into it (each once, with +// the cookie; one F has not got — yet: the editor that made it may still be +// taking it there — is asked for again a few times, a while apart; one the +// network lost is asked for with the next change). A blob made here (a file +// put into a computer) is kept at once and taken to F — tried again once +// after a while, then told as failed. +// What arrives is told to the document's listeners, so that a picture +// shows once its bytes are here. + +import type * as Y from 'yjs'; +import { memoryBlobSide, missingBlobs, type BlobSide } from '@mdg-filius/doc'; + +/** The wait before an upload is tried again. */ +export const UPLOAD_RETRY_MILLISECONDS = 2_000; +/** How often, and how long apart, a blob F has not got is asked for again. */ +export const BLOB_ASKS = 5; +export const BLOB_ASK_AGAIN_MILLISECONDS = 2_000; + +export interface RemoteBlobsOptions { + /** F's address. */ + readonly server: string; + readonly project: string; + /** A blob that could not be taken to F. */ + readonly onFailure?: (sha: string) => void; + readonly fetch?: typeof fetch; +} + +export interface RemoteBlobs extends BlobSide { + /** Fetches what the document names and is not here; done when every answer has come. */ + fetchMissing(doc: Y.Doc): Promise; +} + +export function remoteBlobs({ server, project, onFailure, fetch: request = (input, init) => fetch(input, init) }: RemoteBlobsOptions): RemoteBlobs { + const address = (path: string) => `${new URL(path, server).href}?project=${encodeURIComponent(project)}`; + const upload = async (sha: string, bytes: Uint8Array, tries: number): Promise => { + try { + const response = await request(address('/blobs'), { method: 'POST', credentials: 'include', headers: { 'content-type': 'application/octet-stream' }, body: bytes as BodyInit }); + const answer = response.ok ? ((await response.json()) as { sha?: unknown }) : undefined; + if (answer?.sha === sha) return; + } catch { + // Tried again below. + } + if (tries > 1) { + await new Promise((resolve) => setTimeout(resolve, UPLOAD_RETRY_MILLISECONDS)); + return upload(sha, bytes, tries - 1); + } + onFailure?.(sha); + }; + const side = memoryBlobSide((sha, bytes) => void upload(sha, bytes, 2)); + const asked = new Set(); + /** Blobs F has not got (not asked for until their wait is over), and how often F said so. */ + const unavailable = new Set(); + const misses = new Map(); + const blobs: RemoteBlobs = { + bytes: side.bytes, + keep: side.keep, + onArrive: side.onArrive, + async fetchMissing(doc) { + const wanted = missingBlobs(doc).filter((sha) => !asked.has(sha) && !unavailable.has(sha)); + for (const sha of wanted) asked.add(sha); + const found = await Promise.all( + wanted.map(async (sha): Promise<[string, Uint8Array] | undefined> => { + try { + const response = await request(address(`/blobs/${sha}`), { credentials: 'include' }); + if (response.ok) return [sha, new Uint8Array(await response.arrayBuffer())]; + if (response.status === 404) { + const asks = (misses.get(sha) ?? 0) + 1; + misses.set(sha, asks); + unavailable.add(sha); + if (asks < BLOB_ASKS) + setTimeout(() => { + unavailable.delete(sha); + void blobs.fetchMissing(doc); + }, BLOB_ASK_AGAIN_MILLISECONDS); + } + return undefined; + } catch { + return undefined; + } finally { + asked.delete(sha); + } + }), + ); + side.arrived(new Map(found.filter((each) => each !== undefined))); + }, + }; + return blobs; +} diff --git a/packages/editor/src/sheet/sheet.css b/packages/editor/src/sheet/sheet.css index 8e2af9d..84af0be 100644 --- a/packages/editor/src/sheet/sheet.css +++ b/packages/editor/src/sheet/sheet.css @@ -396,3 +396,19 @@ height: 10px; border-radius: 5px; } + +/* Over the editor on a project of F: something went wrong that the person should know. */ +.remote-notice { + position: fixed; + left: 50%; + bottom: 16px; + z-index: 100; + transform: translateX(-50%); + max-width: min(90vw, 560px); + margin: 0; + padding: 10px 14px; + border: 1px solid var(--danger); + border-radius: 8px; + background: var(--danger-soft); + color: var(--danger-on-soft); +} diff --git a/packages/editor/test/remoteBlobs.test.ts b/packages/editor/test/remoteBlobs.test.ts new file mode 100644 index 0000000..6545566 --- /dev/null +++ b/packages/editor/test/remoteBlobs.test.ts @@ -0,0 +1,135 @@ +// The editor's blob side on F (remoteBlobs), on a fetch the test answers: a +// blob made here is kept at once and taken to F with the cookie, tried +// again once, then told as failed (also when F names it otherwise); the +// blobs a document names are fetched once each and told as they come; one F +// has not got is not asked for again, one the network lost is. + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import * as Y from 'yjs'; +import { attachBlobs, BLOB_THRESHOLD_BYTES, blobSha, changeFiles, emptyProject, memoryBlobSide, missingBlobs, ROOTS, writeProject } from '@mdg-filius/doc'; +import { BLOB_ASK_AGAIN_MILLISECONDS, BLOB_ASKS, remoteBlobs, UPLOAD_RETRY_MILLISECONDS } from '../src/remote/remoteBlobs.js'; + +const picture = (seed: number) => Uint8Array.from({ length: BLOB_THRESHOLD_BYTES }, (_, index) => (index + seed) % 256); + +interface Call { + readonly url: string; + readonly init: RequestInit | undefined; +} + +function fakeFetch(answer: (call: Call) => Response | Promise) { + const calls: Call[] = []; + const fetch = (async (url: string, init?: RequestInit) => { + const call = { url, init }; + calls.push(call); + return answer(call); + }) as typeof globalThis.fetch; + return { calls, fetch }; +} + +/** A document naming these pictures as blobs on a computer, made with a side of its own. */ +function docNaming(...pictures: Uint8Array[]): Y.Doc { + const made = new Y.Doc(); + attachBlobs(made, memoryBlobSide()); + writeProject(made, emptyProject()); + made.transact(() => made.getMap>(ROOTS.nodes).set('pc', new Y.Map())); + changeFiles(made, 'pc', 'person', (files) => pictures.forEach((each, index) => files.writeImageFile(`/p${index}.png`, each, 'image/png'))); + const copy = new Y.Doc(); + Y.applyUpdate(copy, Y.encodeStateAsUpdate(made)); + return copy; +} + +beforeEach(() => vi.useFakeTimers()); +afterEach(() => vi.useRealTimers()); + +describe("the editor's blobs on F", () => { + it('keeps a blob at once and takes it to F with the cookie', async () => { + const bytes = picture(1); + const { calls, fetch } = fakeFetch(() => Response.json({ sha: blobSha(bytes) }, { status: 201 })); + const failed: string[] = []; + const side = remoteBlobs({ server: 'https://f.test/', project: 'p 1', fetch, onFailure: (sha) => failed.push(sha) }); + side.keep(blobSha(bytes), bytes); + expect(side.bytes(blobSha(bytes))).toBe(bytes); + await vi.runAllTimersAsync(); + expect(calls.map(({ url, init }) => [url, init?.method, init?.credentials, init?.body])).toEqual([['https://f.test/blobs?project=p%201', 'POST', 'include', bytes]]); + expect(failed).toEqual([]); + // Kept once: the same again goes nowhere. + side.keep(blobSha(bytes), bytes); + await vi.runAllTimersAsync(); + expect(calls).toHaveLength(1); + }); + + it('tries again once after a while, then says it failed — also when F names the bytes otherwise', async () => { + for (const answer of [() => new Response('', { status: 500 }), () => Response.json({ sha: 'other' }), () => Promise.reject(new Error('offline'))]) { + const bytes = picture(2); + const { calls, fetch } = fakeFetch(answer); + const failed: string[] = []; + remoteBlobs({ server: 'https://f.test', project: 'p', fetch, onFailure: (sha) => failed.push(sha) }).keep(blobSha(bytes), bytes); + await vi.advanceTimersByTimeAsync(UPLOAD_RETRY_MILLISECONDS - 1); + expect(calls).toHaveLength(1); + await vi.advanceTimersByTimeAsync(1); + await vi.runAllTimersAsync(); + expect([calls.length, failed]).toEqual([2, [blobSha(bytes)]]); + } + // A second try that works says nothing. + const bytes = picture(3); + let tries = 0; + const { fetch } = fakeFetch(() => (++tries === 1 ? new Response('', { status: 503 }) : Response.json({ sha: blobSha(bytes) }))); + const failed: string[] = []; + remoteBlobs({ server: 'https://f.test', project: 'p', fetch, onFailure: (sha) => failed.push(sha) }).keep(blobSha(bytes), bytes); + await vi.runAllTimersAsync(); + expect([tries, failed]).toEqual([2, []]); + }); + + it('fetches the blobs a document names once each, and tells when they came', async () => { + const [first, second] = [picture(4), picture(5)]; + const doc = docNaming(first, second); + const { calls, fetch } = fakeFetch(({ url }) => new Response(url.includes(blobSha(first)) ? first : second)); + const side = remoteBlobs({ server: 'https://f.test', project: 'p', fetch }); + attachBlobs(doc, side); + let heard = 0; + side.onArrive(() => (heard += 1)); + expect(missingBlobs(doc).sort()).toEqual([blobSha(first), blobSha(second)].sort()); + await Promise.all([side.fetchMissing(doc), side.fetchMissing(doc)]); + expect(calls.map(({ url, init }) => [url, init?.credentials]).sort()).toEqual( + [ + [`https://f.test/blobs/${blobSha(first)}?project=p`, 'include'], + [`https://f.test/blobs/${blobSha(second)}?project=p`, 'include'], + ].sort(), + ); + expect([missingBlobs(doc), heard]).toEqual([[], 1]); + expect(side.bytes(blobSha(first))).toEqual(first); + }); + + it('asks again a few times, a while apart, for a blob F has not got yet; for one the network lost with the next change', async () => { + const [late, gone, lost] = [picture(6), picture(7), picture(8)]; + const doc = docNaming(late, gone, lost); + let offline = true; + let lateAsked = 0; + const { calls, fetch } = fakeFetch(({ url }) => { + if (url.includes(blobSha(late))) return ++lateAsked < 3 ? new Response('', { status: 404 }) : new Response(late); + if (url.includes(blobSha(gone))) return new Response('', { status: 404 }); + if (offline) return Promise.reject(new Error('offline')); + return new Response(lost); + }); + const side = remoteBlobs({ server: 'https://f.test', project: 'p', fetch }); + attachBlobs(doc, side); + await side.fetchMissing(doc); + offline = false; + // Without a change, only what F has not got is asked for again. + await vi.advanceTimersByTimeAsync(BLOB_ASK_AGAIN_MILLISECONDS - 1); + expect(calls).toHaveLength(3); + await vi.advanceTimersByTimeAsync(1); + await vi.runAllTimersAsync(); + expect(side.bytes(blobSha(late))).toEqual(late); + expect(calls.filter(({ url }) => url.includes(blobSha(late)))).toHaveLength(3); + expect(calls.filter(({ url }) => url.includes(blobSha(gone)))).toHaveLength(BLOB_ASKS); + // The one the network lost comes with the next change (its fetch then asks for what is still missing). + await side.fetchMissing(doc); + expect(side.bytes(blobSha(lost))).toEqual(lost); + expect(missingBlobs(doc)).toEqual([blobSha(gone)]); + // Given up on: not asked for again. + await side.fetchMissing(doc); + await vi.runAllTimersAsync(); + expect(calls.filter(({ url }) => url.includes(blobSha(gone)))).toHaveLength(BLOB_ASKS); + }); +}); diff --git a/packages/server/src/blobRoutes.ts b/packages/server/src/blobRoutes.ts new file mode 100644 index 0000000..54fe1f6 --- /dev/null +++ b/packages/server/src/blobRoutes.ts @@ -0,0 +1,72 @@ +// The editor's way to the blobs (blobs.ts): GET /blobs/?project=… for +// whoever has a grant to the project, POST /blobs?project=… (the bytes as +// the body) for whoever may write in it — not in a frozen project, and only +// from the editor's origin (a page elsewhere under the same domain would +// bring the person's cookie along). The answer to a POST is the name F gave +// the bytes. For the editor's origins F answers CORS, so that a page served +// from elsewhere (the development page) can reach it with its cookie. + +import type { IncomingMessage, ServerResponse } from 'node:http'; +import { BLOB_MAX_BYTES, type BlobStore } from './blobs.js'; +import type { Log } from './log.js'; +import type { SessionStore } from './sessions.js'; +import { cookieValue, SESSION_COOKIE } from './sync.js'; +import { isUuid } from './ticket.js'; + +export interface BlobRouteOptions { + readonly blobs: BlobStore; + readonly sessions: SessionStore; + readonly log: Log; + readonly origins: readonly string[]; +} + +export async function handleBlobs(request: IncomingMessage, response: ServerResponse, url: URL, { blobs, sessions, log, origins }: BlobRouteOptions): Promise { + const origin = request.headers.origin; + const allowed = origin !== undefined && origins.includes(origin); + const headers: Record = allowed ? { 'access-control-allow-origin': origin, 'access-control-allow-credentials': 'true', vary: 'Origin' } : {}; + const send = (status: number, value: object) => { + response.writeHead(status, { ...headers, 'content-type': 'application/json', 'cache-control': 'no-store' }); + response.end(JSON.stringify(value)); + }; + if (request.method === 'OPTIONS') { + response.writeHead(allowed ? 204 : 403, { ...headers, 'access-control-allow-methods': 'GET, POST', 'access-control-allow-headers': 'content-type', 'access-control-max-age': '600' }); + response.end(); + return; + } + const project = url.searchParams.get('project'); + if (!isUuid(project)) return send(400, { error: 'notAProject' }); + const key = cookieValue(request.headers.cookie, SESSION_COOKIE); + const access = key === undefined ? ({ kind: 'notSignedIn' } as const) : await sessions.access(key, project); + if (access.kind !== 'granted') return send(403, { error: access.kind }); + const name = /^\/blobs\/([^/]+)$/.exec(url.pathname)?.[1]; + if (request.method === 'GET' && name !== undefined) { + const bytes = (await blobs.get([name])).get(name); + if (bytes === undefined) return send(404, { error: 'unknownBlob' }); + // A blob's bytes never change under its name. + response.writeHead(200, { ...headers, 'content-type': 'application/octet-stream', 'content-length': String(bytes.length), 'cache-control': 'private, max-age=31536000, immutable' }); + response.end(bytes); + return; + } + if (request.method === 'POST' && url.pathname === '/blobs') { + if (!allowed) { + log.warn('blob.foreignOrigin', { project, origin: origin ?? null }); + return send(403, { error: 'foreignOrigin' }); + } + if (access.level !== 'write') return send(403, { error: 'readOnly' }); + if (access.frozen) return send(403, { error: 'frozen' }); + const parts: Buffer[] = []; + let size = 0; + for await (const part of request as AsyncIterable) { + size += part.length; + if (size > BLOB_MAX_BYTES) { + log.warn('blob.tooLarge', { project, user: access.person.login }); + return send(413, { error: 'tooLarge' }); + } + parts.push(part); + } + const sha = await blobs.put(Buffer.concat(parts)); + log.info('blob.kept', { project, user: access.person.login, sha, size }); + return send(201, { sha }); + } + send(404, { error: 'notFound' }); +} diff --git a/packages/server/src/blobs.ts b/packages/server/src/blobs.ts new file mode 100644 index 0000000..22f967f --- /dev/null +++ b/packages/server/src/blobs.ts @@ -0,0 +1,52 @@ +// The large binary files of the projects' computers (plan §5.1, +// packages/doc's blobs.ts), content-addressed: named by the SHA-256 of +// their bytes, kept once however many projects hold them. F computes the +// name itself; what a client says a blob is called is not believed. +// +// Not yet: forgetting a blob no project names any more (they stay; the +// commits keep theirs in the dashboard, plan §10.1). + +import { createHash } from 'node:crypto'; +import type pg from 'pg'; + +/** A blob at most (a picture or a file put into a simulated computer). */ +export const BLOB_MAX_BYTES = 16 * 1024 * 1024; + +export interface BlobStore { + migrate(): Promise; + /** Bytes kept; their name. */ + put(bytes: Uint8Array): Promise; + /** The blobs of these names that F has. */ + get(shas: readonly string[]): Promise>; +} + +export const shaOf = (bytes: Uint8Array) => createHash('sha256').update(bytes).digest('hex'); + +const SCHEMA = ` +CREATE TABLE IF NOT EXISTS filius_blob ( + sha text PRIMARY KEY, + bytes bytea NOT NULL, + size integer NOT NULL, + created_at timestamptz NOT NULL DEFAULT now() +); +`; + +export class PostgresBlobs implements BlobStore { + constructor(private readonly pool: pg.Pool) {} + + async migrate(): Promise { + await this.pool.query(SCHEMA); + } + + async put(bytes: Uint8Array): Promise { + const sha = shaOf(bytes); + await this.pool.query('INSERT INTO filius_blob (sha, bytes, size) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING', [sha, Buffer.from(bytes), bytes.length]); + return sha; + } + + async get(shas: readonly string[]): Promise> { + if (shas.length === 0) return new Map(); + const result = await this.pool.query<{ sha: string; bytes: Buffer }>('SELECT sha, bytes FROM filius_blob WHERE sha = ANY($1)', [shas]); + return new Map(result.rows.map((row) => [row.sha, new Uint8Array(row.bytes.buffer, row.bytes.byteOffset, row.bytes.byteLength)])); + } +} diff --git a/packages/server/src/index.ts b/packages/server/src/index.ts index 2bf50f8..014b6c6 100644 --- a/packages/server/src/index.ts +++ b/packages/server/src/index.ts @@ -1,6 +1,8 @@ // Server F (plan §8): the shared documents, the simulation rooms, the // sign-in and the dashboard's service. export * from './auth.js'; +export * from './blobRoutes.js'; +export * from './blobs.js'; export * from './log.js'; export * from './persistence.js'; export * from './server.js'; diff --git a/packages/server/src/main.ts b/packages/server/src/main.ts index 3a6c7ec..08e3d39 100644 --- a/packages/server/src/main.ts +++ b/packages/server/src/main.ts @@ -7,6 +7,7 @@ // machine without HTTPS). The rooms' thread is the bundle beside this one // (dist/simWorker.mjs). +import { PostgresBlobs } from './blobs.js'; import { jsonLog } from './log.js'; import { startServer } from './server.js'; import { PostgresSessions } from './sessions.js'; @@ -23,12 +24,15 @@ const secureCookie = process.env.FILIUS_INSECURE_COOKIE !== '1'; if (!secureCookie) log.warn('auth.insecureCookie', { note: 'the session cookie goes over plain HTTP' }); const store = PostgresStore.connect(config.DATABASE_URL!); const sessions = new PostgresSessions(store.pool); +const blobs = new PostgresBlobs(store.pool); await store.migrate(); await sessions.migrate(); +await blobs.migrate(); const server = await startServer({ port: Number(process.env.FILIUS_PORT ?? 8080), store, sessions, + blobs, log, ticketSecret: config.FILIUS_TICKET_SECRET!, serviceKey: config.FILIUS_SERVICE_KEY!, diff --git a/packages/server/src/server.ts b/packages/server/src/server.ts index 4ec5907..4262ede 100644 --- a/packages/server/src/server.ts +++ b/packages/server/src/server.ts @@ -1,6 +1,7 @@ // 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 WebSockets /sync for the shared +// the dashboard (with the service key), /blobs for the large binary files +// of the projects' computers, 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 @@ -18,6 +19,8 @@ import type { AddressInfo, Socket } from 'node:net'; import { SIM_REFUSED_CODE, type SyncRefusal } from '@mdg-filius/protocol'; import { WebSocketServer, type WebSocket } from 'ws'; import { exchangeTicket } from './auth.js'; +import { handleBlobs } from './blobRoutes.js'; +import type { BlobStore } from './blobs.js'; import { errorText, type Log } from './log.js'; import { handleService } from './service.js'; import { holdMessages, SimRooms } from './simRooms.js'; @@ -46,6 +49,7 @@ export interface ServerOptions extends SyncOptions { readonly secureCookie: boolean; /** The bundled room thread (dist/simWorker.mjs). */ readonly workerUrl: URL; + readonly blobs: BlobStore; readonly simThreads?: number; readonly lingerMilliseconds?: number; readonly pingMilliseconds?: number; @@ -65,6 +69,7 @@ export async function startServer(options: ServerOptions): Promise const { hocuspocus } = sync; const sims = new SimRooms({ hocuspocus, + blobs: options.blobs, log: options.log, workerUrl: options.workerUrl, ...(options.simThreads === undefined ? {} : { threads: options.simThreads }), @@ -105,6 +110,7 @@ export async function startServer(options: ServerOptions): Promise 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, sims }); + else if (path === '/blobs' || path.startsWith('/blobs/')) await handleBlobs(request, response, new URL(request.url ?? '/', 'http://filius'), options); else json(response, 404, { error: 'notFound' }); }; const http: Server = createServer((request, response) => { diff --git a/packages/server/src/simChannel.ts b/packages/server/src/simChannel.ts index 3a0a43d..8d92ed0 100644 --- a/packages/server/src/simChannel.ts +++ b/packages/server/src/simChannel.ts @@ -4,10 +4,16 @@ // 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). +// numbers them). A room's large binary files (blobs) are F's: the thread +// asks for the ones it has not got, and gives F the ones the simulation +// makes before the document's change that names them. import type { HostMessage } from '@mdg-filius/protocol'; +/** How often, and how long apart, a thread asks again for a blob F has not got (yet). */ +export const BLOB_ASKS = 5; +export const BLOB_ASK_AGAIN_MILLISECONDS = 1_000; + export type ToRoomThread = /** A room for a project, its document as it is now. */ | { readonly type: 'open'; readonly room: string; readonly state: Uint8Array } @@ -19,10 +25,16 @@ export type ToRoomThread = /** 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 }; + | { readonly type: 'close'; readonly room: string; readonly writeBack: boolean } + /** The blobs asked for: those F has, and the names of those it has not. */ + | { readonly type: 'blobs'; readonly room: string; readonly blobs: readonly { readonly sha: string; readonly bytes: Uint8Array }[]; readonly missing: readonly string[] }; 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 } + /** Blobs the document names that the thread has not got. */ + | { readonly type: 'needBlobs'; readonly room: string; readonly shas: readonly string[] } + /** A blob the simulation made (before the change that names it). */ + | { readonly type: 'blob'; readonly room: string; readonly bytes: Uint8Array } | { readonly type: 'closed'; readonly room: string }; diff --git a/packages/server/src/simRooms.ts b/packages/server/src/simRooms.ts index e604fbf..a364739 100644 --- a/packages/server/src/simRooms.ts +++ b/packages/server/src/simRooms.ts @@ -3,7 +3,11 @@ // 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. +// it; what the room writes comes back as the room's own change. The blobs +// its document names come from F's store when the thread asks; a blob the +// simulation makes is kept before the change that names it is applied (the +// room's messages are taken in order), so that no one is told of a file +// whose bytes are not there. // // 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 @@ -23,6 +27,7 @@ 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 type { BlobStore } from './blobs.js'; import { errorText, type Log } from './log.js'; import type { FromRoomThread, ToRoomThread } from './simChannel.js'; @@ -39,6 +44,7 @@ export interface SimPerson { export interface SimRoomsOptions { readonly hocuspocus: Hocuspocus; + readonly blobs: BlobStore; readonly log: Log; /** The bundled room thread (dist/simWorker.mjs). */ readonly workerUrl: URL; @@ -64,6 +70,8 @@ interface Room { frozen: boolean; linger: ReturnType | undefined; ended: boolean; + /** The thread's changes and blobs, taken in order. */ + taken: Promise; } type RawData = Buffer | ArrayBuffer | Buffer[]; @@ -124,9 +132,30 @@ export class SimRooms { private fromThread(thread: Thread, message: FromRoomThread): void { const room = [...thread.rooms].find((each) => each.project === message.room); if (room === undefined) return; + const inOrder = (work: () => Promise | void) => { + room.taken = room.taken.then(work).catch((error: unknown) => this.options.log.error('sim.blobFailed', { project: room.project, error: errorText(error) })); + }; switch (message.type) { case 'doc': - Y.applyUpdate(room.doc, message.update, room); + inOrder(() => Y.applyUpdate(room.doc, message.update, room)); + return; + case 'blob': + inOrder(async () => void (await this.options.blobs.put(message.bytes))); + return; + case 'needBlobs': + void this.options.blobs.get(message.shas).then( + (found) => + this.post(room, { + type: 'blobs', + room: room.project, + blobs: [...found].map(([sha, bytes]) => ({ sha, bytes })), + missing: message.shas.filter((sha) => !found.has(sha)), + }), + (error: unknown) => { + this.options.log.error('sim.blobFailed', { project: room.project, error: errorText(error) }); + this.post(room, { type: 'blobs', room: room.project, blobs: [], missing: message.shas }); + }, + ); return; case 'host': room.clients.get(message.client)?.socket.send(JSON.stringify(message.message)); @@ -168,6 +197,7 @@ export class SimRooms { frozen, linger: undefined, ended: false, + taken: Promise.resolve(), }; thread.rooms.add(room); this.post(room, { type: 'open', room: project, state: Y.encodeStateAsUpdate(doc) }); @@ -267,6 +297,8 @@ export class SimRooms { const done = (async () => { this.rooms.delete(room.project); await this.ask(room, { type: 'close', room: room.project, writeBack }); + // What it wrote back is in the document before the document is let go. + await room.taken; this.forget(room); await room.direct.disconnect(); this.options.log.info('sim.roomClosed', { project: room.project, writeBack }); diff --git a/packages/server/src/simWorker.ts b/packages/server/src/simWorker.ts index ed87db1..1cb9950 100644 --- a/packages/server/src/simWorker.ts +++ b/packages/server/src/simWorker.ts @@ -1,12 +1,18 @@ // 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. +// (dist/simWorker.mjs), started by the main thread. The blobs a room's +// document names are asked of the main thread as they come into it; a start +// waits until the ones asked for have come (or F said it has not got them), +// so that the simulation starts with the files' bytes. One F has not got — +// yet: the editor that made it may still be taking it there — is asked for +// again a few times; when it comes, the running simulation reads it. import { parentPort } from 'node:worker_threads'; import * as Y from 'yjs'; +import { attachBlobs, memoryBlobSide, missingBlobs } from '@mdg-filius/doc'; import { SimulationRoom, TICK_MILLISECONDS } from '@mdg-filius/sim'; -import type { FromRoomThread, ToRoomThread } from './simChannel.js'; +import { BLOB_ASK_AGAIN_MILLISECONDS, BLOB_ASKS, type FromRoomThread, type ToRoomThread } from './simChannel.js'; /** The origin of what came from the main thread: not sent back. */ const MAIN_ORIGIN = 'main'; @@ -16,21 +22,43 @@ interface HeldRoom { readonly room: SimulationRoom; /** The main thread's number of a person → the room's. */ readonly clients: Map; + readonly blobs: ReturnType; + /** Blobs asked for and not answered yet. */ + readonly asked: Set; + /** Blobs F said it has not got: asked again a while later, a few times (`misses` counts). */ + readonly unavailable: Set; + readonly misses: Map; + /** Starts waiting for the blobs asked for. */ + readonly waiting: { readonly client: number; readonly message: unknown }[]; } const port = parentPort!; const rooms = new Map(); const post = (message: FromRoomThread) => port.postMessage(message); +/** Asks the main thread for the blobs the room's document names and the room has not got. */ +function askForBlobs(name: string, held: HeldRoom): void { + const shas = missingBlobs(held.doc).filter((sha) => !held.asked.has(sha) && !held.unavailable.has(sha)); + if (shas.length === 0) return; + for (const sha of shas) held.asked.add(sha); + post({ type: 'needBlobs', room: name, shas }); +} + +const isStart = (message: unknown) => typeof message === 'object' && message !== null && (message as { kind?: unknown }).kind === 'start'; + port.on('message', (message: ToRoomThread) => { if (message.type === 'open') { const doc = new Y.Doc(); + const blobs = memoryBlobSide((_sha, bytes) => post({ type: 'blob', room: message.room, bytes })); + attachBlobs(doc, blobs); 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() }); + const held: HeldRoom = { doc, room, clients: new Map(), blobs, asked: new Set(), unavailable: new Set(), misses: new Map(), waiting: [] }; + rooms.set(message.room, held); + askForBlobs(message.room, held); return; } const held = rooms.get(message.room); @@ -38,7 +66,32 @@ port.on('message', (message: ToRoomThread) => { switch (message.type) { case 'doc': Y.applyUpdate(held.doc, message.update, MAIN_ORIGIN); + askForBlobs(message.room, held); return; + case 'blobs': { + held.blobs.arrived(new Map(message.blobs.map(({ sha, bytes }) => [sha, bytes]))); + for (const { sha } of message.blobs) held.asked.delete(sha); + if (message.blobs.length > 0) held.room.blobsArrived(); + for (const sha of message.missing) { + held.asked.delete(sha); + const asks = (held.misses.get(sha) ?? 0) + 1; + held.misses.set(sha, asks); + held.unavailable.add(sha); + if (asks < BLOB_ASKS) + setTimeout(() => { + if (rooms.get(message.room) !== held) return; + held.unavailable.delete(sha); + askForBlobs(message.room, held); + }, BLOB_ASK_AGAIN_MILLISECONDS); + } + if (held.asked.size === 0) { + for (const { client, message: waiting } of held.waiting.splice(0)) { + const roomClient = held.clients.get(client); + if (roomClient !== undefined) held.room.receive(roomClient, waiting); + } + } + return; + } case 'connect': held.clients.set( message.client, @@ -47,7 +100,9 @@ port.on('message', (message: ToRoomThread) => { return; case 'client': { const client = held.clients.get(message.client); - if (client !== undefined) held.room.receive(client, message.message); + if (client === undefined) return; + if (isStart(message.message) && held.asked.size > 0) held.waiting.push({ client: message.client, message: message.message }); + else held.room.receive(client, message.message); return; } case 'disconnect': { diff --git a/packages/server/test/blobs.test.ts b/packages/server/test/blobs.test.ts new file mode 100644 index 0000000..a46ee04 --- /dev/null +++ b/packages/server/test/blobs.test.ts @@ -0,0 +1,268 @@ +// 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/ 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): Promise { + 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) => 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>> { + 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>(ROOTS.nodes).get(node)!.get('files') as Y.Map>; +} + +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>(ROOTS.nodes).get(notebook)!.get('files') as Y.Map>; + // 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>(ROOTS.nodes).get(notebook)!.get('files') as Y.Map>; + expect(files.get('/kopie.png')?.get('blob')).toBe(blobSha(bytes)); + expect(BLOB_ASKS).toBe(5); + }, 30_000); + + it("starts a simulation only once the blobs asked for have come, however slow F's store is", async () => { + const { project, notebook, bytes } = await projectWithPicture(51); + await blobs.put(bytes); + const server = await startTestServer(store, sessions, { blobs: slowly(blobs, { get: 800 }) }); + const room = await enterRoom(server, project); + room.send({ kind: 'start' }); + 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>(ROOTS.nodes).get(notebook)!.get('files') as Y.Map>; + expect([files.get('/fehlt.png')?.get('blob'), files.has('/neu.txt')]).toEqual([blobSha(bytes), true]); + }, 30_000); +}); diff --git a/packages/server/test/harness.ts b/packages/server/test/harness.ts index a2819cf..82aabd2 100644 --- a/packages/server/test/harness.ts +++ b/packages/server/test/harness.ts @@ -16,7 +16,7 @@ import { expect } from 'vitest'; import WebSocket from 'ws'; import * as Y from 'yjs'; import { emptyProject, writeProject } from '@mdg-filius/doc'; -import { startServer, type DocStore, type FiliusServer, type Log, type ServerOptions, type SessionStore } from '../src/index.js'; +import { PostgresBlobs, startServer, type DocStore, type PostgresStore, type FiliusServer, type Log, type ServerOptions, type SessionStore } from '../src/index.js'; import { signTicket } from './tickets.js'; export const TICKET_SECRET = 'test-ticket-secret'; @@ -38,6 +38,13 @@ export function memoryLog(): MemoryLog { export type TestServer = FiliusServer & { readonly log: MemoryLog; readonly base: string }; +/** F's blob store in the test database (the table made, once a file). */ +async function testBlobs(store: DocStore): Promise { + const blobs = new PostgresBlobs((store as PostgresStore).pool); + await blobs.migrate(); + return blobs; +} + let roomThread: Promise | undefined; /** src/simWorker.ts bundled into a directory of its own, once a test file. */ @@ -63,6 +70,7 @@ export async function startTestServer(store: DocStore, sessions: SessionStore, o origins: [EDITOR_ORIGIN], secureCookie: false, workerUrl: await bundledRoomThread(), + blobs: await testBlobs(store), simThreads: 1, ...options, }); diff --git a/packages/sim/src/room.ts b/packages/sim/src/room.ts index bf9eee1..632da88 100644 --- a/packages/sim/src/room.ts +++ b/packages/sim/src/room.ts @@ -191,6 +191,15 @@ export class SimulationRoom { } } + /** + * The bytes of blobs came (on F, after the simulation read the files + * without them): every node's files read anew into the running engine. + */ + blobsArrived(): void { + if (this.mode !== 'simulation') return; + this.guarded(() => this.syncFiles(new Set(this.engine.state.topologySnapshot.fileSystemsByNodeId.keys()))); + } + /** Whether a node's files changed since they were last written back; marked written if so. */ private takeChangedFiles(id: Id, files: { readonly revision: number } | undefined): boolean { const written = this.writtenFiles.get(id);