From 10374c3696371290e8d0e4e3f200443a309d6541 Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Sat, 7 Nov 2026 22:01:42 +0800 Subject: [PATCH] Add durable canvas runtime, scheduling and crash recovery --- package.json | 4 +- src/lib/canvas-runtime/contracts.ts | 21 +++ src/lib/canvas-runtime/engine.ts | 132 ++++++++++++++++ src/lib/canvas-runtime/ledger.ts | 108 +++++++++++++ src/lib/canvas-runtime/reconcile.ts | 51 +++++++ src/lib/canvas-runtime/recovery.ts | 112 ++++++++++++++ src/lib/canvas-runtime/scheduler.ts | 184 +++++++++++++++++++++++ src/lib/canvas-runtime/storage.ts | 150 ++++++++++++++++++ src/lib/canvas-runtime/structures.ts | 45 ++++++ tests/headless/browser-storage-cases.mjs | 101 +++++++++++++ tests/headless/browser-storage.mjs | 55 +++++++ tests/headless/engine.test.mjs | 129 ++++++++++++++++ tests/headless/ledger.test.mjs | 59 ++++++++ tests/headless/scheduler.test.mjs | 100 ++++++++++++ 14 files changed, 1250 insertions(+), 1 deletion(-) create mode 100644 src/lib/canvas-runtime/contracts.ts create mode 100644 src/lib/canvas-runtime/engine.ts create mode 100644 src/lib/canvas-runtime/ledger.ts create mode 100644 src/lib/canvas-runtime/reconcile.ts create mode 100644 src/lib/canvas-runtime/recovery.ts create mode 100644 src/lib/canvas-runtime/scheduler.ts create mode 100644 src/lib/canvas-runtime/storage.ts create mode 100644 src/lib/canvas-runtime/structures.ts create mode 100644 tests/headless/browser-storage-cases.mjs create mode 100644 tests/headless/browser-storage.mjs create mode 100644 tests/headless/engine.test.mjs create mode 100644 tests/headless/ledger.test.mjs create mode 100644 tests/headless/scheduler.test.mjs diff --git a/package.json b/package.json index 3f19ece..26349d1 100644 --- a/package.json +++ b/package.json @@ -6,7 +6,9 @@ "dev": "next dev --hostname localhost --port 17321", "build": "next build", "start": "next start --hostname localhost --port 17321", - "typecheck": "tsc --noEmit" + "typecheck": "tsc --noEmit", + "test:headless": "node --test tests/headless/*.test.mjs", + "test:headless:browser": "node tests/headless/browser-storage.mjs" }, "dependencies": { "@radix-ui/react-dialog": "1.3.1", diff --git a/src/lib/canvas-runtime/contracts.ts b/src/lib/canvas-runtime/contracts.ts new file mode 100644 index 0000000..601c97e --- /dev/null +++ b/src/lib/canvas-runtime/contracts.ts @@ -0,0 +1,21 @@ +import type { Document, Id, Json, OperationIntent, Snapshot } from '../canvas-core/types.ts'; +import type { LedgerSnapshot, FrozenRequest } from './ledger.ts'; +import type { CompiledRequest, OperationStatus } from './scheduler.ts'; + +export type RuntimeState = { + readonly ledger: LedgerSnapshot; + readonly acknowledged: Document; + readonly versions: ReadonlyMap; + readonly receipts: ReadonlyMap; + readonly conflicts: ReadonlyMap; +}; +export type EngineSnapshot = Snapshot; +export type CanonicalProposal = { readonly document: Document; readonly versions: ReadonlyMap }; +export type PersistencePort = { + supportsRequest?: (request: FrozenRequest) => boolean; + compile(intent: OperationIntent, requestId: Id, snapshot: EngineSnapshot): CompiledRequest; + submit(request: FrozenRequest, signal: AbortSignal): Promise; + query(request: FrozenRequest, signal: AbortSignal): Promise; + /** Pure codec, merging only participants into the latest base, never an old working snapshot. */ + canonical(request: FrozenRequest, status: OperationStatus, snapshot: EngineSnapshot): CanonicalProposal; +}; diff --git a/src/lib/canvas-runtime/engine.ts b/src/lib/canvas-runtime/engine.ts new file mode 100644 index 0000000..8c2e5db --- /dev/null +++ b/src/lib/canvas-runtime/engine.ts @@ -0,0 +1,132 @@ +import type { Document, Id, Registry } from '../canvas-core/types.ts'; +import { createEditor, type EditorOptions } from '../canvas-core/editor.ts'; +import { createDocument } from '../canvas-core/document.ts'; +import { createSessions } from '../canvas-core/gestures.ts'; +import { equal, fault, freeze, readMap } from '../canvas-core/values.ts'; +import type { EngineSnapshot, PersistencePort, RuntimeState } from './contracts.ts'; +import { createLedger, type FrozenRequest } from './ledger.ts'; +import { operationCanvases, replayPending } from './reconcile.ts'; +import { decodeRecovery, encodeRecovery } from './recovery.ts'; +import { createScheduler, type Clock, type OperationStatus } from './scheduler.ts'; +import type { StoragePort, StorageScope } from './storage.ts'; + +export type EngineOptions = { + scope: StorageScope; registry: Registry; document: Document; versions: ReadonlyMap; + storage: StoragePort; persistence: PersistencePort; createId?: () => Id; + historyLimit?: number; policy?: EditorOptions['policy']; clock?: Clock; + onError?: (error: unknown) => void; +}; + +/** One project owner. Detaching a view does not close this engine or abort persistent work. */ +export async function openProjectEngine(options: EngineOptions) { + if (options.scope.projectId !== options.document.projectId) throw fault('project_mismatch'); + const leader = await options.storage.claim(options.scope); + try { + const image = await options.storage.recover(leader), recovered = image ? decodeRecovery(image, options.registry) : undefined; + if (recovered && recovered.document.projectId !== options.scope.projectId) throw fault('project_mismatch'); + for (const canvas of options.document.canvases.keys()) if (!Number.isSafeInteger(options.versions.get(canvas)) || options.versions.get(canvas)! < 1) throw fault('missing_server_version', [canvas]); + let closing = false, closed = false, durableRootRevision = image?.rootRevision ?? -1; + const report = (error: unknown) => { try { options.onError?.(error); } catch { /* Error reporting cannot alter state. */ } }; + const initial: RuntimeState = recovered?.runtime ?? Object.freeze({ acknowledged: createDocument(options.document.projectId, [...options.document.canvases.values()], [...options.document.elements.values()], options.document.tombstones), versions: readMap(options.versions), ledger: createLedger().getSnapshot(), receipts: readMap(new Map()), conflicts: readMap(new Map()) }); + const editor = createEditor({ registry: options.registry, document: options.document, recovery: recovered, runtime: initial, createId: options.createId, historyLimit: options.historyLimit, + policy(command, document) { if (closing || closed) return { enabled: false, code: 'engine_closing', targets: [] }; return options.policy?.(command, document) ?? { enabled: true }; }, + traversalPolicy() { return closing || closed ? { enabled: false, code: 'engine_closing', targets: [] } : { enabled: true }; }, + onListenerError: report, + reduceRuntime(previous, next, commit) { + if (!commit.operationIds.length) return previous; + const ledger = createLedger(previous.ledger); + for (const id of commit.operationIds) { + const intent = next.operations.get(id)!; + // Last unresolved writer on every participating canvas preserves causal order. + const last = new Map(); + for (const record of [...previous.ledger.operations.values()].sort((a, b) => a.intent.sequence - b.intent.sequence)) if (record.phase !== 'resolved') for (const canvas of record.intent.canvasIds) last.set(canvas, record.intent.id); + const dependencies = [...new Set([...intent.dependencies, ...intent.canvasIds.flatMap(canvas => last.has(canvas) ? [last.get(canvas)!] : [])])]; + const basePresence: Record = {}; + for (const change of intent.changes) if (change.kind === 'element') basePresence[change.id] = change.before ? 'exists' : 'absent'; else if (change.kind === 'field') basePresence[change.target.id] ??= 'exists'; + ledger.add(freeze({ ...intent, dependencies }), basePresence); + } + return Object.freeze({ ...previous, ledger: ledger.getSnapshot() }); + }, + }); + const pending = new Map(); + if (!recovered) pending.set(editor.getSnapshot().rootRevision, editor.getSnapshot()); + let writing: Promise | undefined, scheduled = false; + function persist(): Promise { + if (writing) return writing; + const task = (async () => { + while (pending.size) { + const [revision, snapshot] = pending.entries().next().value!; + const receipt = await options.storage.commit(leader, { image: encodeRecovery(snapshot) }); + if (receipt.durableRootRevision !== revision) throw fault('invalid_durable_receipt'); + durableRootRevision = revision; pending.delete(revision); + if (editor.getSnapshot().runtime.ledger.durableGeneration < snapshot.runtime.ledger.generation) mutate(ledger => ledger.markDurable(snapshot.runtime.ledger.generation)); + } + })(); + writing = task.finally(() => { writing = undefined; }); + return writing; + } + function mutate(reduce: (ledger: ReturnType) => void) { + const result = editor.updateRuntime(previous => { const ledger = createLedger(previous.ledger); reduce(ledger); return Object.freeze({ ...previous, ledger: ledger.getSnapshot() }); }); + if (result.status === 'rejected') throw fault(result.code, result.targets); + } + function complete(request: FrozenRequest, status: OperationStatus): { readonly blocked: true } | void { + const root = editor.getSnapshot(), previous = root.runtime, ledger = createLedger(previous.ledger); + if (request.phase === 'resolved') return; + if (status.outcome !== 'applied') { mutate(draft => draft.result(request.id, status.outcome, status.error)); return; } + const receipts = new Map(previous.receipts), conflicts = new Map(previous.conflicts); + if (status.canonical !== undefined) receipts.set(request.id, freeze(status.canonical)); + try { + const proposal = options.persistence.canonical(request, status, root); + if (proposal.document.projectId !== root.document.projectId) throw fault('project_mismatch'); + for (const canvas of request.canvasIds) { + const version = proposal.versions.get(canvas); + if (!version || version <= request.expectedVersions[canvas] || version < (previous.versions.get(canvas) ?? 0)) throw fault('stale_canonical_version', [canvas]); + } + // A codec cannot roll back an unrelated canvas while reconciling this request. + for (const [canvas, version] of previous.versions) if (!request.canvasIds.includes(canvas)) { + if (proposal.versions.get(canvas) !== version || !equal(proposal.document.canvases.get(canvas), previous.acknowledged.canvases.get(canvas)) || !equal([...proposal.document.elements.values()].filter(element => element.ownerCanvasId === canvas), [...previous.acknowledged.elements.values()].filter(element => element.ownerCanvasId === canvas))) throw fault('canonical_scope_mismatch', [canvas]); + } + ledger.result(request.id, 'applied'); + const acknowledged = createDocument(proposal.document.projectId, [...proposal.document.canvases.values()], [...proposal.document.elements.values()], proposal.document.tombstones); + const document = replayPending(acknowledged, ledger.getSnapshot(), options.registry); + conflicts.delete(request.id); + const runtime: RuntimeState = Object.freeze({ ledger: ledger.getSnapshot(), acknowledged, versions: readMap(proposal.versions), receipts: readMap(receipts), conflicts: readMap(conflicts) }); + const result = editor.reconcile(document, runtime, root.rootRevision); + if (result.status === 'rejected') throw fault(result.code, result.targets); + } catch (error) { + // Preserve both the known receipt and the user's working document. No rollback or lane release. + const waiting = createLedger(previous.ledger); waiting.result(request.id, 'applied', (error as Error).message, false); + conflicts.set(request.id, (error as Error).message || 'reconcile_conflict'); + const result = editor.updateRuntime(() => Object.freeze({ ...previous, ledger: waiting.getSnapshot(), receipts: readMap(receipts), conflicts: readMap(conflicts) }), root.rootRevision); + if (result.status === 'rejected') throw fault(result.code, result.targets); + report(error); return { blocked: true }; + } + } + const scheduler = createScheduler({ getLedger: () => editor.getSnapshot().runtime.ledger, mutate, persist, + createId: options.createId ?? (() => crypto.randomUUID()), + participants: intent => operationCanvases(intent, editor.getSnapshot().document), + compile: (intent, id) => options.persistence.compile(intent, id, editor.getSnapshot()), + submit: (request, signal) => options.persistence.submit(request, signal), query: (request, signal) => options.persistence.query(request, signal), supportsRequest: options.persistence.supportsRequest, complete, onError: report, clock: options.clock, + }); + const sessions = createSessions(editor); + const unsubscribe = editor.subscribe((commit, snapshot) => { + pending.set(snapshot.rootRevision, snapshot); + if (!scheduled) { scheduled = true; queueMicrotask(() => { scheduled = false; void persist().catch(report); }); } + // Intents already exist in this root; notification only schedules their IDs. + queueMicrotask(() => { if (!closing && !closed) scheduler.notify(commit.operationIds); }); + }); + scheduler.start(); + return Object.freeze({ + editor, sessions, getSnapshot: editor.getSnapshot, persist, + getDurability: () => Object.freeze({ rootRevision: editor.getSnapshot().rootRevision, durableRootRevision }), + retry: () => scheduler.resume(), + async settle() { await Promise.resolve(); await scheduler.settle(); await persist(); }, + async close() { + if (closed) return; + closing = true; sessions.dispose(); scheduler.stop(); + // Failure keeps the leader/recovery material available for an explicit retry of close. + await persist(); unsubscribe(); editor.dispose(); await leader.release(); closed = true; + }, + }); + } catch (error) { await leader.release(); throw error; } +} diff --git a/src/lib/canvas-runtime/ledger.ts b/src/lib/canvas-runtime/ledger.ts new file mode 100644 index 0000000..4b0a07c --- /dev/null +++ b/src/lib/canvas-runtime/ledger.ts @@ -0,0 +1,108 @@ +import type { Change, Id, Json, OperationIntent } from '../canvas-core/types.ts'; +import { clone, equal, fault, freeze, readMap, validJson } from '../canvas-core/values.ts'; + +export type Presence = 'absent' | 'exists' | 'unknown'; +export type RequestPhase = 'frozen' | 'dispatch-marked' | 'waiting' | 'resolved'; +export type Outcome = 'unresolved' | 'accepted' | 'applied' | 'not-applied'; +export type FrozenRequest = { + readonly id: Id; readonly protocolVersion: number; readonly generation: number; readonly body: string; + readonly operationIds: readonly Id[]; readonly canvasIds: readonly Id[]; readonly expectedVersions: Readonly>; + readonly phase: RequestPhase; readonly outcome: Outcome; readonly error?: string; readonly attempt: number; + readonly retryAt?: number; readonly retryAction?: 'submit' | 'query'; +}; +export type OperationRecord = { readonly intent: OperationIntent; readonly phase: 'pending' | 'frozen' | 'resolved' | 'blocked'; readonly requestId?: Id; readonly blockers: readonly string[]; readonly basePresence: Readonly> }; +export type LedgerSnapshot = { readonly generation: number; readonly durableGeneration: number; readonly operations: ReadonlyMap; readonly requests: ReadonlyMap }; + +/** A field-safe compressor for a wholly unsent segment. Frozen records never enter here. */ +export function compressChanges(changes: readonly Change[]): readonly Change[] { + const result: Change[] = []; + for (const change of changes) { + const id = change.kind === 'field' ? change.target.id : change.id; + const previousIndex = result.findLastIndex(item => (item.kind === 'field' ? item.target.id : item.id) === id); + const previous = result[previousIndex]; + if (previous?.kind === 'element' && previous.after && !previous.before && change.kind === 'field') { + // Reuse the same controlled field application semantics; no full-node stale overwrite. + const after = clone(previous.after); + if (change.target.incarnation !== after.incarnation) throw fault('stale_incarnation', [id]); + if (change.field.startsWith('data.')) { + const key = change.field.slice(5), data = { ...after.data }; + if (change.afterPresent === false) delete data[key]; else data[key] = clone(change.after); + result[previousIndex] = freeze({ ...previous, after: { ...after, data } }); + } else result[previousIndex] = freeze({ ...previous, after: { ...after, [change.field]: clone(change.after) } }) as Change; + } else if (previous?.kind === 'element' && previous.after && !previous.before && change.kind === 'element' && !change.after && equal(previous.after, change.before)) { + result.splice(previousIndex, 1); + } else if (previous?.kind === 'field' && change.kind === 'field' && previous.field === change.field && previous.target.incarnation === change.target.incarnation && equal(previous.after, change.before) && previous.afterPresent === change.beforePresent) { + const merged = freeze({ ...previous, after: change.after, afterPresent: change.afterPresent }); + if (equal(merged.before, merged.after) && merged.beforePresent === merged.afterPresent) result.splice(previousIndex, 1); else result[previousIndex] = merged; + } else result.push(change); + } + return freeze(result); +} + +export function createLedger(initial?: LedgerSnapshot) { + let generation = initial?.generation ?? 0, durableGeneration = initial?.durableGeneration ?? 0; + const operations = new Map(initial?.operations), requests = new Map(initial?.requests); + let cached: LedgerSnapshot; + const refresh = () => { cached = Object.freeze({ generation, durableGeneration, operations: readMap(operations), requests: readMap(requests) }); }; + refresh(); + function changed() { generation++; refresh(); } + function requireRequest(id: Id) { const request = requests.get(id); if (!request) throw fault('request_missing', [id]); return request; } + function assertAcyclic(intent: OperationIntent) { + const visiting = new Set(); + const visit = (id: Id): void => { + if (id === intent.id || visiting.has(id)) throw fault('dependency_cycle', [id]); + const operation = operations.get(id); + if (!operation) throw fault('dependency_missing', [id]); + visiting.add(id); for (const dependency of operation.intent.dependencies) visit(dependency); visiting.delete(id); + }; + for (const id of intent.dependencies) visit(id); + } + return Object.freeze({ + getSnapshot: () => cached, + add(intent: OperationIntent, basePresence: Readonly>) { + if (!intent.id || !Number.isSafeInteger(intent.sequence) || intent.sequence < 1 || new Set(intent.dependencies).size !== intent.dependencies.length || Object.values(basePresence).some(value => !['absent', 'exists', 'unknown'].includes(value))) throw fault('invalid_operation'); + const existing = operations.get(intent.id); + if (existing) { if (!equal(existing.intent, intent)) throw fault('operation_identity_conflict', [intent.id]); return; } + assertAcyclic(intent); + operations.set(intent.id, freeze({ intent: clone(intent), phase: 'pending', blockers: [], basePresence: { ...basePresence } })); changed(); + }, + eligible(id: Id) { + const operation = operations.get(id); + return !!operation && operation.phase === 'pending' && !operation.blockers.length && operation.intent.dependencies.every(dependency => operations.get(dependency)?.phase === 'resolved'); + }, + block(id: Id, reasons: readonly string[]) { const operation = operations.get(id); if (!operation || !['pending', 'blocked'].includes(operation.phase)) return; operations.set(id, freeze({ ...operation, phase: reasons.length ? 'blocked' : 'pending', blockers: [...reasons] })); changed(); }, + freeze(input: { id: Id; operationIds: readonly Id[]; canvasIds: readonly Id[]; expectedVersions: Readonly>; payload: Json }) { + if (!input.id || requests.has(input.id) || !input.operationIds.length || !input.canvasIds.length || !validJson(input.payload) || new Set(input.operationIds).size !== input.operationIds.length || new Set(input.canvasIds).size !== input.canvasIds.length || Object.keys(input.expectedVersions).length !== input.canvasIds.length || input.canvasIds.some(id => !id || !Number.isSafeInteger(input.expectedVersions[id]) || input.expectedVersions[id] < 1)) throw fault('invalid_frozen_request'); + for (const id of input.operationIds) { + const operation = operations.get(id); + if (!operation || operation.phase !== 'pending' || operation.blockers.length || operation.intent.dependencies.some(dependency => operations.get(dependency)?.phase !== 'resolved' && !input.operationIds.includes(dependency))) throw fault('operation_not_ready', [id]); + } + const request: FrozenRequest = freeze({ id: input.id, protocolVersion: 1, generation: generation + 1, body: JSON.stringify(input.payload), operationIds: [...input.operationIds], canvasIds: [...input.canvasIds].sort(), expectedVersions: { ...input.expectedVersions }, phase: 'frozen', outcome: 'unresolved', attempt: 0 }); + requests.set(request.id, request); + for (const id of input.operationIds) operations.set(id, freeze({ ...operations.get(id)!, phase: 'frozen', requestId: request.id })); + changed(); return request; + }, + markDispatch(id: Id) { + const request = requireRequest(id); + if (request.phase === 'resolved') throw fault('request_resolved', [id]); + const next: FrozenRequest = freeze({ ...request, phase: 'dispatch-marked', attempt: request.attempt + 1, retryAt: undefined, retryAction: undefined }); requests.set(id, next); changed(); return next; + }, + result(id: Id, outcome: Outcome, error?: string, reconciled = true) { + const request = requireRequest(id); + if ((request.phase === 'resolved' || request.outcome === 'applied') && request.outcome !== outcome) throw fault('receipt_conflict', [id]); + const phase = (outcome === 'applied' && reconciled) || outcome === 'not-applied' ? 'resolved' : 'waiting'; + requests.set(id, freeze({ ...request, phase, outcome, error, retryAt: undefined, retryAction: undefined })); + if (outcome === 'applied' && reconciled) for (const operationId of request.operationIds) operations.set(operationId, freeze({ ...operations.get(operationId)!, phase: 'resolved' })); + if (outcome === 'not-applied') for (const operationId of request.operationIds) operations.set(operationId, freeze({ ...operations.get(operationId)!, phase: 'blocked', blockers: [error || 'not_applied'] })); + changed(); + }, + scheduleRetry(id: Id, dueAt: number, action: 'submit' | 'query') { + const request = requireRequest(id); + if (request.phase === 'resolved' || !Number.isFinite(dueAt)) throw fault('invalid_retry', [id]); + requests.set(id, freeze({ ...request, retryAt: dueAt, retryAction: action })); changed(); + }, + blockRequest(id: Id, error: string) { const request = requireRequest(id); if (request.phase === 'resolved' || request.error === error) return; requests.set(id, freeze({ ...request, error, retryAt: undefined, retryAction: undefined })); changed(); }, + markDurable(through: number) { if (!Number.isSafeInteger(through) || through < durableGeneration || through > generation) throw fault('invalid_durable_generation'); durableGeneration = through; refresh(); }, + possiblyApplied(id: Id) { const request = requireRequest(id); return request.phase !== 'frozen' && request.outcome !== 'not-applied'; }, + }); +} diff --git a/src/lib/canvas-runtime/reconcile.ts b/src/lib/canvas-runtime/reconcile.ts new file mode 100644 index 0000000..c75bf34 --- /dev/null +++ b/src/lib/canvas-runtime/reconcile.ts @@ -0,0 +1,51 @@ +import type { Change, Document, Id, OperationIntent, Registry } from '../canvas-core/types.ts'; +import { applyChanges, validateDocument } from '../canvas-core/document.ts'; +import { equal, fault } from '../canvas-core/values.ts'; +import type { LedgerSnapshot } from './ledger.ts'; + +/** The footprint follows ownership; geometry does not decide which canvas owns an object. */ +export function operationCanvases(intent: OperationIntent, document: Document): readonly Id[] { + if (intent.canvasIds?.length) return intent.canvasIds; + const ids = new Set(); + for (const change of intent.changes) { + if (change.kind === 'canvas') { + ids.add(change.id); + if (change.before?.parent) ids.add(change.before.parent.canvasId); + if (change.after?.parent) ids.add(change.after.parent.canvasId); + } else if (change.kind === 'element') { + if (change.before) ids.add(change.before.ownerCanvasId); + if (change.after) ids.add(change.after.ownerCanvasId); + } else { + const owner = document.elements.get(change.target.id)?.ownerCanvasId; + if (owner) ids.add(owner); + if (change.field === 'ownerCanvasId') { ids.add(change.before as Id); ids.add(change.after as Id); } + } + } + if (!ids.size) throw fault('missing_participants', [intent.id]); + return [...ids].sort(); +} + +/** Replay once, by logical sequence. A receipt can already contain a desired field value. */ +export function replayPending(base: Document, ledger: LedgerSnapshot, registry: Registry): Document { + let document = base; + for (const record of [...ledger.operations.values()].sort((a, b) => a.intent.sequence - b.intent.sequence)) { + if (record.phase === 'resolved') continue; + for (const change of record.intent.changes) { + if (satisfied(document, change)) continue; + document = applyChanges(document, [change]); + } + } + validateDocument(document, registry); + return document; +} +function satisfied(document: Document, change: Change) { + if (change.kind === 'canvas') return equal(document.canvases.get(change.id) ?? null, change.after); + if (change.kind === 'element') return equal(document.elements.get(change.id) ?? null, change.after); + const element = document.elements.get(change.target.id); + if (!element || element.incarnation !== change.target.incarnation) return false; + if (change.field.startsWith('data.')) { + const key = change.field.slice(5); + return Object.hasOwn(element.data, key) === (change.afterPresent ?? true) && equal(element.data[key] ?? null, change.after); + } + return equal((element as unknown as Record)[change.field], change.after); +} diff --git a/src/lib/canvas-runtime/recovery.ts b/src/lib/canvas-runtime/recovery.ts new file mode 100644 index 0000000..cba9193 --- /dev/null +++ b/src/lib/canvas-runtime/recovery.ts @@ -0,0 +1,112 @@ +import type { Change, CoreSnapshot, Document, HistoryEntry, Id, Json, OperationIntent, Registry } from '../canvas-core/types.ts'; +import { createDocument, validateDocument } from '../canvas-core/document.ts'; +import { fault, freeze, readMap, validJson } from '../canvas-core/values.ts'; +import type { EngineSnapshot, RuntimeState } from './contracts.ts'; +import type { FrozenRequest, LedgerSnapshot, OperationRecord } from './ledger.ts'; +import type { RecoveryImage } from './storage.ts'; + +const documentValue = (document: Document) => ({ projectId: document.projectId, canvases: [...document.canvases.values()], elements: [...document.elements.values()], tombstones: [...document.tombstones] }); +/** Plain records only: functions, credentials, renderer objects, URLs and Map methods never serialize. */ +export function encodeRecovery(snapshot: EngineSnapshot): RecoveryImage { + const payload = JSON.parse(JSON.stringify({ + document: documentValue(snapshot.document), + history: { undo: snapshot.history.undo, redo: snapshot.history.redo, entries: [...snapshot.history.entries] }, + operations: [...snapshot.operations], + runtime: { acknowledged: documentValue(snapshot.runtime.acknowledged), versions: [...snapshot.runtime.versions], receipts: [...snapshot.runtime.receipts], conflicts: [...snapshot.runtime.conflicts], ledger: { ...snapshot.runtime.ledger, operations: [...snapshot.runtime.ledger.operations], requests: [...snapshot.runtime.ledger.requests] } }, + })) as Json; + if (!validJson(payload)) throw fault('invalid_recovery_payload'); + return freeze({ formatVersion: 1, rootRevision: snapshot.rootRevision, localRevision: snapshot.revision, payload }); +} +type ObjectValue = Record; +function object(value: unknown): ObjectValue { if (!value || Array.isArray(value) || typeof value !== 'object') throw fault('corrupt_recovery_image'); return value as ObjectValue; } +function array(value: unknown): unknown[] { if (!Array.isArray(value)) throw fault('corrupt_recovery_image'); return value; } +function string(value: unknown): string { if (typeof value !== 'string' || !value) throw fault('corrupt_recovery_image'); return value; } +function integer(value: unknown, min = 0): number { if (typeof value !== 'number' || !Number.isSafeInteger(value) || value < min) throw fault('corrupt_recovery_image'); return value; } +function strings(value: unknown): string[] { return array(value).map(string); } +function table(value: unknown, decode: (entry: unknown, id: Id) => T): ReadonlyMap { + const result = new Map(); + for (const item of array(value)) { const pair = array(item), id = string(pair[0]); if (pair.length !== 2 || result.has(id)) throw fault('corrupt_recovery_image'); result.set(id, decode(pair[1], id)); } + return readMap(result); +} +function readDocument(value: unknown, registry: Registry): Document { + const raw = object(value); + const tombstones = table(raw.tombstones, entry => integer(entry, 1)); + const document = createDocument(string(raw.projectId), array(raw.canvases) as never, array(raw.elements) as never, tombstones); + validateDocument(document, registry); return document; +} +function changes(value: unknown): readonly Change[] { + for (const item of array(value)) { + const change = object(item); + if (change.kind === 'field') { + const target = object(change.target); string(target.id); integer(target.incarnation, 1); + const field = string(change.field); + if (!['geometry', 'ownerCanvasId', 'source', 'target', 'path'].includes(field) && (!field.startsWith('data.') || !field.slice(5))) throw fault('corrupt_recovery_image'); + for (const flag of ['beforePresent', 'afterPresent']) if (change[flag] !== undefined && typeof change[flag] !== 'boolean') throw fault('corrupt_recovery_image'); + } else if (change.kind === 'element' || change.kind === 'canvas') { + const id = string(change.id); + for (const key of ['before', 'after']) if (change[key] !== null && string(object(change[key]).id) !== id) throw fault('corrupt_recovery_image'); + } else throw fault('corrupt_recovery_image'); + if (!Object.hasOwn(change, 'before') || !Object.hasOwn(change, 'after')) throw fault('corrupt_recovery_image'); + } + return freeze(value as Change[]); +} +function intent(value: unknown, id: Id): OperationIntent { + const raw = object(value); + if (string(raw.id) !== id) throw fault('corrupt_recovery_image'); + string(raw.operationType); integer(raw.operationSchemaVersion, 1); integer(raw.sequence, 1); string(raw.origin); + strings(raw.canvasIds); strings(raw.dependencies); changes(raw.changes); + return freeze(value as OperationIntent); +} +export function decodeRecovery(image: RecoveryImage, registry: Registry): CoreSnapshot & { readonly rootRevision: number; readonly runtime: RuntimeState } { + if (image.formatVersion !== 1) throw fault('unsupported_recovery_version'); + integer(image.rootRevision); integer(image.localRevision); + if (image.localRevision > image.rootRevision || !validJson(image.payload)) throw fault('corrupt_recovery_image'); + // Own the parsed records; mutating a caller's RecoveryImage must not mutate live state. + const raw = object(JSON.parse(JSON.stringify(image.payload))), document = readDocument(raw.document, registry), history = object(raw.history), runtime = object(raw.runtime), ledger = object(runtime.ledger); + const operations = table(raw.operations, intent), historyEntries = table(history.entries, (value, id) => { + const raw = object(value); if (string(raw.id) !== id || typeof raw.label !== 'string') throw fault('corrupt_recovery_image'); + changes(raw.changes); for (const operationId of strings(raw.operationIds)) if (!operations.has(operationId)) throw fault('corrupt_recovery_image'); + if (raw.groupId !== undefined) string(raw.groupId); + return freeze(value as HistoryEntry); + }); + const undo = strings(history.undo), redo = strings(history.redo), stackIds = [...undo, ...redo]; + if (new Set(stackIds).size !== stackIds.length || stackIds.length !== historyEntries.size || stackIds.some(id => !historyEntries.has(id))) throw fault('corrupt_recovery_image'); + const records = table(ledger.operations, (value, id) => { + const raw = object(value); intent(raw.intent, id); strings(raw.blockers); object(raw.basePresence); + if (!operations.has(id) || !['pending', 'frozen', 'resolved', 'blocked'].includes(string(raw.phase))) throw fault('corrupt_recovery_image'); + return freeze(value as OperationRecord); + }); + if (operations.size !== records.size) throw fault('corrupt_recovery_image'); + // Validate the dependency DAG iteratively, including missing dependencies, without recursion overflow. + const counts = new Map(), dependents = new Map(), ready: Id[] = []; + for (const [id, record] of records) { + const dependencies = record.intent.dependencies; + if (new Set(dependencies).size !== dependencies.length) throw fault('corrupt_recovery_image'); + counts.set(id, dependencies.length); if (!dependencies.length) ready.push(id); + for (const dependency of dependencies) { if (!records.has(dependency)) throw fault('corrupt_recovery_image'); dependents.set(dependency, [...(dependents.get(dependency) ?? []), id]); } + } + for (let index = 0; index < ready.length; index++) for (const id of dependents.get(ready[index]) ?? []) { counts.set(id, counts.get(id)! - 1); if (!counts.get(id)) ready.push(id); } + if (ready.length !== records.size) throw fault('corrupt_recovery_image'); + const requests = table(ledger.requests, (value, id) => { + const raw = object(value); + if (string(raw.id) !== id || typeof raw.body !== 'string') throw fault('corrupt_recovery_image'); + integer(raw.protocolVersion, 1); + integer(raw.generation, 1); integer(raw.attempt); strings(raw.canvasIds); object(raw.expectedVersions); + if (!['frozen', 'dispatch-marked', 'waiting', 'resolved'].includes(string(raw.phase)) || !['unresolved', 'accepted', 'applied', 'not-applied'].includes(string(raw.outcome))) throw fault('corrupt_recovery_image'); + for (const operationId of strings(raw.operationIds)) if (records.get(operationId)?.requestId !== id) throw fault('corrupt_recovery_image'); + if (raw.retryAt !== undefined && (typeof raw.retryAt !== 'number' || !Number.isFinite(raw.retryAt))) throw fault('corrupt_recovery_image'); + if (raw.retryAction !== undefined && !['submit', 'query'].includes(string(raw.retryAction))) throw fault('corrupt_recovery_image'); + return freeze(value as FrozenRequest); + }); + for (const record of records.values()) if (record.requestId && !requests.has(record.requestId) || record.phase === 'frozen' && !record.requestId) throw fault('corrupt_recovery_image'); + const generation = integer(ledger.generation), durableGeneration = integer(ledger.durableGeneration); + if (durableGeneration > generation) throw fault('corrupt_recovery_image'); + const acknowledged = readDocument(runtime.acknowledged, registry); + if (acknowledged.projectId !== document.projectId) throw fault('project_mismatch'); + const versions = table(runtime.versions, value => integer(value, 1)); + for (const id of acknowledged.canvases.keys()) if (!versions.has(id)) throw fault('corrupt_recovery_image'); + const restoredLedger: LedgerSnapshot = Object.freeze({ generation, durableGeneration, operations: records, requests }); + return Object.freeze({ revision: image.localRevision, rootRevision: image.rootRevision, document, history: Object.freeze({ undo: freeze(undo), redo: freeze(redo), entries: historyEntries }), operations, + runtime: Object.freeze({ ledger: restoredLedger, acknowledged, versions, receipts: table(runtime.receipts, value => freeze(value as Json)), conflicts: table(runtime.conflicts, string) }), + }); +} diff --git a/src/lib/canvas-runtime/scheduler.ts b/src/lib/canvas-runtime/scheduler.ts new file mode 100644 index 0000000..b784f82 --- /dev/null +++ b/src/lib/canvas-runtime/scheduler.ts @@ -0,0 +1,184 @@ +import type { Id, Json, OperationIntent } from '../canvas-core/types.ts'; +import { fault, readMap } from '../canvas-core/values.ts'; +import { createLedger, type FrozenRequest, type LedgerSnapshot, type Outcome } from './ledger.ts'; +import { createDeque, createMinHeap } from './structures.ts'; + +export type OperationStatus = { readonly outcome: Outcome; readonly error?: string; readonly retryAfterMs?: number; readonly canonical?: Json }; +export type CompiledRequest = { readonly payload: Json; readonly canvasIds: readonly Id[]; readonly expectedVersions: Readonly> }; +export type Clock = { now(): number; setTimer(callback: () => void, delay: number): unknown; clearTimer(handle: unknown): void }; +const browserClock: Clock = { now: () => Date.now(), setTimer: (fn, delay) => setTimeout(fn, delay), clearTimer: handle => clearTimeout(handle as ReturnType) }; +type Ledger = ReturnType; +export type SchedulerOptions = { + getLedger(): LedgerSnapshot; + mutate(reduce: (ledger: Ledger) => void): void; + persist(): Promise; + createId(): Id; + participants(intent: OperationIntent): readonly Id[]; + compile(intent: OperationIntent, requestId: Id): CompiledRequest; + submit(request: FrozenRequest, signal: AbortSignal): Promise; + query(request: FrozenRequest, signal: AbortSignal): Promise; + supportsRequest?: (request: FrozenRequest) => boolean; + /** Atomically reconcile the receipt/base/working document and the ledger outcome. */ + complete(request: FrozenRequest, status: OperationStatus): { readonly blocked: true } | void; + onError(error: unknown): void; + clock?: Clock; + concurrency?: number; + retryDelay?: (attempt: number) => number; +}; + +/** Queue/heap/claims are indexes only. Every authoritative transition lives in the ledger. */ +export function createScheduler(options: SchedulerOptions) { + const clock = options.clock ?? browserClock, concurrency = options.concurrency ?? 2; + if (!Number.isSafeInteger(concurrency) || concurrency < 1) throw fault('invalid_concurrency'); + const ready = createDeque(), queued = new Set(), claims = new Map(); + const waiters = new Map>(), dependents = new Map>(), indexed = new Set(); + const retry = createMinHeap<{ id: Id; dueAt: number; sequence: number; token: number }>((a, b) => a.dueAt - b.dueAt || a.sequence - b.sequence); + const tokens = new Map(), active = new Map }>(); + let timer: unknown, enqueueSequence = 0, stopped = true, closed = false, pumping = false; + const delay = (attempt: number) => Math.max(1, options.retryDelay?.(attempt) ?? Math.min(30_000, 500 * 2 ** Math.min(attempt, 6))); + const report = (error: unknown) => { try { options.onError(error); } catch { /* Reporting cannot change a receipt. */ } }; + const supported = (request: FrozenRequest) => options.supportsRequest?.(request) ?? request.protocolVersion === 1; + + function enqueue(id: Id) { + if (queued.has(id)) return; + const record = options.getLedger().operations.get(id); + if (!record || record.phase !== 'pending' || record.blockers.length || record.intent.dependencies.some(dependency => options.getLedger().operations.get(dependency)?.phase !== 'resolved')) return; + queued.add(id); ready.push(id); + } + function index(id: Id) { + const record = options.getLedger().operations.get(id); + if (!record) return; + if (!indexed.has(id)) { + indexed.add(id); + for (const dependency of record.intent.dependencies) { + const set = dependents.get(dependency) ?? new Set(); set.add(id); dependents.set(dependency, set); + } + } + enqueue(id); + } + function own(request: FrozenRequest) { + for (const canvas of request.canvasIds) { + const prior = claims.get(canvas); + if (prior && prior !== request.id) throw fault('recovery_claim_conflict', [canvas]); + } + for (const canvas of request.canvasIds) claims.set(canvas, request.id); + } + function release(request: FrozenRequest) { + for (const canvas of request.canvasIds) if (claims.get(canvas) === request.id) { + claims.delete(canvas); + const waiting = waiters.get(canvas); waiters.delete(canvas); + for (const id of waiting ?? []) enqueue(id); + } + for (const id of request.operationIds) for (const dependent of dependents.get(id) ?? []) enqueue(dependent); + } + function schedule(id: Id, dueAt: number, action: 'submit' | 'query', persist = true) { + if (persist) options.mutate(ledger => ledger.scheduleRetry(id, dueAt, action)); + const token = (tokens.get(id) ?? 0) + 1; tokens.set(id, token); + retry.push({ id, dueAt, sequence: ++enqueueSequence, token }); armTimer(); + } + function validRetry(item: { id: Id; token: number }) { + return tokens.get(item.id) === item.token && options.getLedger().requests.get(item.id)?.phase !== 'resolved'; + } + function armTimer() { + if (timer !== undefined) { clock.clearTimer(timer); timer = undefined; } + while (retry.peek() && !validRetry(retry.peek()!)) retry.pop(); + if (stopped || !retry.peek()) return; + timer = clock.setTimer(() => { timer = undefined; pump(); }, Math.max(0, retry.peek()!.dueAt - clock.now())); + } + function launch(id: Id, action: 'submit' | 'query') { + if (active.has(id) || stopped) return; + const controller = new AbortController(); + // Start in a microtask so the handle exists before any synchronous adapter error. + const promise = Promise.resolve().then(async () => { + try { + if (stopped) return; + if (action === 'submit') options.mutate(ledger => ledger.markDispatch(id)); + // This barrier includes the exact body/key and the conservative dispatch marker. + await options.persist(); + if (stopped) return; + const request = options.getLedger().requests.get(id)!; + let status: OperationStatus; + try { status = await (action === 'submit' ? options.submit(request, controller.signal) : options.query(request, controller.signal)); } + catch { status = { outcome: 'unresolved', error: 'transport_unknown' }; } + if (stopped) return; // A stale identity/runtime cannot accept a late callback. + const completion = options.complete(request, status); + const current = options.getLedger().requests.get(id)!; + if (current.phase === 'resolved') { tokens.delete(id); await options.persist(); if (!stopped) release(current); } + else if (completion?.blocked) { await options.persist(); } + else { + // A missing receipt is unknown, so the only write retry reuses this frozen request. + const nextAction = status.outcome === 'accepted' || action === 'submit' ? 'query' : 'submit'; + schedule(id, clock.now() + Math.max(1, status.retryAfterMs ?? delay(current.attempt)), nextAction); + await options.persist(); + } + } catch (error) { report(error); } // Storage/reconciliation failure keeps its claims held. + }).finally(() => { active.delete(id); if (!stopped) pump(); }); + active.set(id, { controller, promise }); + } + function pump() { + if (stopped || pumping) return; + pumping = true; + try { + while (active.size < concurrency && retry.peek() && retry.peek()!.dueAt <= clock.now()) { + const item = retry.pop()!; + if (!validRetry(item)) continue; + const request = options.getLedger().requests.get(item.id); + if (request && !active.has(item.id)) launch(item.id, request.retryAction ?? 'query'); + } + let budget = ready.size; + while (active.size < concurrency && budget-- > 0) { + const id = ready.pop(); if (!id) break; + queued.delete(id); + const record = options.getLedger().operations.get(id); + if (!record || record.phase !== 'pending' || record.blockers.length || record.intent.dependencies.some(dependency => options.getLedger().operations.get(dependency)?.phase !== 'resolved')) continue; + try { + const canvases = [...new Set(options.participants(record.intent))].sort(); + if (!canvases.length) throw fault('missing_participants'); + const occupied = canvases.filter(canvas => claims.has(canvas)); + if (occupied.length) { + for (const canvas of occupied) { const set = waiters.get(canvas) ?? new Set(); set.add(id); waiters.set(canvas, set); } + continue; + } + const requestId = options.createId(), compiled = options.compile(record.intent, requestId); + if (canvases.length !== compiled.canvasIds.length || canvases.some(canvas => !compiled.canvasIds.includes(canvas))) throw fault('participant_mismatch'); + options.mutate(ledger => ledger.freeze({ id: requestId, operationIds: [id], ...compiled })); + own(options.getLedger().requests.get(requestId)!); launch(requestId, 'submit'); + } catch (error) { options.mutate(ledger => ledger.block(id, [(error as Error).message || 'compile_failed'])); report(error); } + } + } finally { pumping = false; armTimer(); } + } + return Object.freeze({ + start() { + if (closed) throw fault('scheduler_closed'); + if (!stopped) return; + // Validate the entire recovered claim set before sending anything. + claims.clear(); + for (const request of options.getLedger().requests.values()) if (request.phase !== 'resolved') own(request); + stopped = false; + for (const operation of [...options.getLedger().operations.values()].sort((a, b) => a.intent.sequence - b.intent.sequence)) index(operation.intent.id); + for (const request of options.getLedger().requests.values()) if (request.phase !== 'resolved') { + if (!supported(request)) { options.mutate(ledger => ledger.blockRequest(request.id, 'request_interpreter_unavailable')); continue; } + schedule(request.id, request.retryAt ?? clock.now(), request.retryAction ?? (request.phase === 'frozen' ? 'submit' : 'query'), false); + } + pump(); + }, + notify(operationIds: readonly Id[]) { for (const id of operationIds) index(id); pump(); }, + /** Explicit retry after storage/auth/permission recovery; preserves frozen bodies. */ + async resume() { + if (stopped) return; + // A receipt may be in memory after its durable write failed. Keep claims until retry succeeds. + await options.persist(); + if (stopped) return; + for (const request of options.getLedger().requests.values()) { + if (active.has(request.id)) continue; + if (request.phase === 'resolved') release(request); + else if (supported(request)) schedule(request.id, request.retryAt ?? clock.now(), request.retryAction ?? 'query', false); + } + for (const id of indexed) enqueue(id); + pump(); + }, + getClaims: () => readMap(claims), + async settle() { while (active.size) await Promise.all([...active.values()].map(item => item.promise)); }, + stop() { closed = true; stopped = true; if (timer !== undefined) clock.clearTimer(timer); timer = undefined; for (const item of active.values()) item.controller.abort(); }, + }); +} diff --git a/src/lib/canvas-runtime/storage.ts b/src/lib/canvas-runtime/storage.ts new file mode 100644 index 0000000..158a9f0 --- /dev/null +++ b/src/lib/canvas-runtime/storage.ts @@ -0,0 +1,150 @@ +import type { Json } from '../canvas-core/types.ts'; +import { clone, equal, fault, validJson } from '../canvas-core/values.ts'; + +export type StorageScope = { readonly accountId: string; readonly workspaceId: string; readonly projectId: string }; +export type RecoveryImage = { readonly formatVersion: 1; readonly rootRevision: number; readonly localRevision: number; readonly payload: Json }; +export type DurableBatch = { readonly image: RecoveryImage; readonly files?: readonly { readonly id: string; readonly bytes: Blob }[] }; +export type StorageLeader = { readonly scope: StorageScope; readonly fence: number; readonly active: boolean; release(): Promise }; +export type StoragePort = { + claim(scope: StorageScope): Promise; + recover(leader: StorageLeader): Promise; + commit(leader: StorageLeader, batch: DurableBatch): Promise<{ readonly durableRootRevision: number }>; + readFile(leader: StorageLeader, id: string): Promise; + close(): void; +}; +type Metadata = { fence: number; throughRevision?: number; checkpointRevision?: number }; +type StoredImage = { readonly image: RecoveryImage; readonly checksum: string }; +async function checksum(image: RecoveryImage) { + const bytes = new TextEncoder().encode(JSON.stringify(image)); + return [...new Uint8Array(await crypto.subtle.digest('SHA-256', bytes))].map(value => value.toString(16).padStart(2, '0')).join(''); +} +const keyFor = (scope: StorageScope) => { + if (![scope.accountId, scope.workspaceId, scope.projectId].every(value => typeof value === 'string' && value.length > 0)) throw fault('invalid_storage_scope'); + return JSON.stringify([scope.accountId, scope.workspaceId, scope.projectId]); +}; +function validateImage(image: RecoveryImage) { + if (image.formatVersion !== 1) throw fault('unsupported_recovery_version'); + if (!Number.isSafeInteger(image.rootRevision) || image.rootRevision < 0 || !Number.isSafeInteger(image.localRevision) || image.localRevision < 0 || image.localRevision > image.rootRevision || !validJson(image.payload)) throw fault('corrupt_recovery_image'); +} +const requestValue = (request: IDBRequest) => new Promise((resolve, reject) => { request.onsuccess = () => resolve(request.result); request.onerror = () => reject(request.error); }); +const transactionDone = (transaction: IDBTransaction) => new Promise((resolve, reject) => { transaction.oncomplete = () => resolve(); transaction.onabort = () => reject(transaction.error ?? fault('storage_aborted')); transaction.onerror = () => { /* onabort is the terminal barrier. */ }; }); + +/** Single-origin browser adapter. Web Locks serialize tabs; a DB fence rejects stale handles. */ +export function createBrowserStorage(options: { indexedDB?: IDBFactory; locks?: LockManager; name?: string; checkpointEvery?: number } = {}): StoragePort { + const factory = options.indexedDB ?? globalThis.indexedDB, locks = options.locks ?? globalThis.navigator?.locks; + if (!factory || !locks) throw fault('persistent_storage_unavailable'); + const checkpointEvery = options.checkpointEvery ?? 32; + if (!Number.isSafeInteger(checkpointEvery) || checkpointEvery < 1) throw fault('invalid_checkpoint_interval'); + let closed = false; + const handles = new Set(); + const opening = new Promise((resolve, reject) => { + const request = factory.open(options.name ?? 'endoleap-headless', 1); + request.onupgradeneeded = () => { + for (const name of ['metadata', 'checkpoints', 'journal', 'files']) if (!request.result.objectStoreNames.contains(name)) request.result.createObjectStore(name); + }; + request.onerror = () => reject(request.error); + request.onblocked = () => reject(fault('storage_upgrade_blocked')); + request.onsuccess = () => { + if (closed) { request.result.close(); reject(fault('storage_closed')); return; } + request.result.onversionchange = () => { closed = true; request.result.close(); for (const handle of handles) void handle.release(); }; + resolve(request.result); + }; + }); + // Opening failures are observable through operations without an unhandled background rejection. + void opening.catch(() => {}); + function check(leader: StorageLeader) { + if (closed) throw fault('storage_closed'); + if (!handles.has(leader) || !leader.active) throw fault('storage_leader_lost'); + return keyFor(leader.scope); + } + async function execute(leader: StorageLeader, stores: string[], mode: IDBTransactionMode, run: (transaction: IDBTransaction, metadata: Metadata, key: string) => Promise): Promise { + const key = check(leader), database = await opening; + check(leader); + const transaction = database.transaction([...new Set(['metadata', ...stores])], mode), done = transactionDone(transaction); + try { + const metadata = await requestValue(transaction.objectStore('metadata').get(key)); + if (!metadata || metadata.fence !== leader.fence) throw fault('storage_fence_lost'); + check(leader); + const result = await run(transaction, metadata, key); + await done; // Never resolve durability on an individual put callback. + return result; + } catch (error) { try { transaction.abort(); } catch { /* Already terminal. */ } await done.catch(() => {}); throw error; } + } + return Object.freeze({ + async claim(scope: StorageScope) { + const key = keyFor(scope); + if (closed) throw fault('storage_closed'); + let unlock!: () => void, acquired!: (leader: StorageLeader) => void, failed!: (error: unknown) => void; + const released = new Promise(resolve => { unlock = resolve; }); + const result = new Promise((resolve, reject) => { acquired = resolve; failed = reject; }); + const held = locks.request(`endoleap-headless:${options.name ?? 'endoleap-headless'}:${key}`, { mode: 'exclusive', ifAvailable: true }, async lock => { + if (!lock) throw fault('storage_scope_in_use'); + const database = await opening; + if (closed) throw fault('storage_closed'); + const transaction = database.transaction('metadata', 'readwrite'), done = transactionDone(transaction); + try { + const store = transaction.objectStore('metadata'), previous = await requestValue(store.get(key)); + const fence = (previous?.fence ?? 0) + 1; + if (!Number.isSafeInteger(fence)) throw fault('storage_fence_exhausted'); + store.put({ ...previous, fence }, key); await done; + let active = true; + const handle: StorageLeader = Object.freeze({ scope: Object.freeze({ ...scope }), fence, get active() { return active; }, async release() { if (!active) return; active = false; handles.delete(handle); unlock(); await held; } }); + handles.add(handle); acquired(handle); + await released; + } catch (error) { try { transaction.abort(); } catch { /* Already terminal. */ } await done.catch(() => {}); throw error; } + }); + void held.catch(failed); + return result; + }, + async recover(leader: StorageLeader) { + const images = await execute(leader, ['checkpoints', 'journal'], 'readonly', async (transaction, metadata, key) => { + if (metadata.throughRevision === undefined) return []; + if (metadata.checkpointRevision === undefined) throw fault('recovery_checkpoint_missing'); + const checkpoint = await requestValue(transaction.objectStore('checkpoints').get([key, metadata.checkpointRevision])); + if (!checkpoint?.image) throw fault('recovery_checkpoint_missing'); + let image = checkpoint.image; + validateImage(image); + if (image.rootRevision !== metadata.checkpointRevision) throw fault('corrupt_recovery_image'); + const journal = metadata.checkpointRevision === metadata.throughRevision ? [] : await requestValue(transaction.objectStore('journal').getAll(IDBKeyRange.bound([key, metadata.checkpointRevision + 1], [key, metadata.throughRevision]))); + for (const next of journal) { validateImage(next.image); if (next.image.rootRevision !== image.rootRevision + 1 || next.image.localRevision < image.localRevision) throw fault('recovery_journal_gap'); image = next.image; } + if (image.rootRevision !== metadata.throughRevision) throw fault('recovery_journal_gap'); + return [checkpoint, ...journal]; + }); + // Hashing must run after the IDB transaction, which would otherwise auto-close while awaiting it. + for (const stored of images) if (stored.checksum !== await checksum(stored.image)) throw fault('recovery_checksum_mismatch'); + check(leader); + return images.at(-1)?.image ?? null; + }, + async commit(leader: StorageLeader, batch: DurableBatch) { + // Copy before awaiting: callers cannot change the durable cut while a transaction queues. + const image = clone(batch.image), files = [...(batch.files ?? [])]; validateImage(image); + const stored: StoredImage = { image, checksum: await checksum(image) }; + return execute(leader, ['checkpoints', 'journal', 'files'], 'readwrite', async (transaction, metadata, key) => { + const previous = metadata.throughRevision; + if (previous !== undefined && image.rootRevision !== previous + 1) { + if (image.rootRevision !== previous) throw fault('storage_revision_gap'); + const saved = await requestValue(transaction.objectStore(metadata.checkpointRevision === previous ? 'checkpoints' : 'journal').get([key, previous])); + if (!equal(saved, stored) || files.length) throw fault('storage_revision_conflict'); + return { durableRootRevision: previous }; + } + if (previous !== undefined && image.localRevision < (await requestValue(transaction.objectStore(metadata.checkpointRevision === previous ? 'checkpoints' : 'journal').get([key, previous]))).image.localRevision) throw fault('storage_revision_conflict'); + for (const file of files) { + if (!file.id || !(file.bytes instanceof Blob)) throw fault('invalid_staged_file'); + // Immutable resource identity: replacing bytes requires a new resource ID. + if (await requestValue(transaction.objectStore('files').getKey([key, file.id]))) throw fault('resource_identity_conflict', [file.id]); + transaction.objectStore('files').put(file.bytes, [key, file.id]); + } + if (previous === undefined || image.rootRevision - (metadata.checkpointRevision ?? 0) >= checkpointEvery) { + transaction.objectStore('checkpoints').put(stored, [key, image.rootRevision]); + if (metadata.checkpointRevision !== undefined) transaction.objectStore('checkpoints').delete([key, metadata.checkpointRevision]); + transaction.objectStore('journal').delete(IDBKeyRange.bound([key, 0], [key, image.rootRevision])); + metadata.checkpointRevision = image.rootRevision; + } else transaction.objectStore('journal').put(stored, [key, image.rootRevision]); + transaction.objectStore('metadata').put({ ...metadata, throughRevision: image.rootRevision }, key); + return { durableRootRevision: image.rootRevision }; + }); + }, + readFile(leader: StorageLeader, id: string) { return execute(leader, ['files'], 'readonly', (transaction, _metadata, key) => requestValue(transaction.objectStore('files').get([key, id]))); }, + close() { closed = true; for (const handle of handles) void handle.release(); void opening.then(database => database.close()).catch(() => {}); }, + }); +} diff --git a/src/lib/canvas-runtime/structures.ts b/src/lib/canvas-runtime/structures.ts new file mode 100644 index 0000000..3ddd5ef --- /dev/null +++ b/src/lib/canvas-runtime/structures.ts @@ -0,0 +1,45 @@ +/** FIFO with amortized O(1) removal, without Array.shift on each item. */ +export function createDeque() { + let items: T[] = [], head = 0; + return { + get size() { return items.length - head; }, + push(value: T) { items.push(value); }, + peek(): T | undefined { return items[head]; }, + pop(): T | undefined { + if (head === items.length) return undefined; + const value = items[head++]; + if (head > 1024 && head * 2 >= items.length) { items = items.slice(head); head = 0; } + return value; + }, + clear() { items = []; head = 0; }, + values() { return items.slice(head); }, + }; +} +export function createMinHeap(compare: (left: T, right: T) => number) { + const items: T[] = []; + return { + get size() { return items.length; }, + peek(): T | undefined { return items[0]; }, + push(value: T) { + let index = items.length; items.push(value); + while (index > 0) { const parent = (index - 1) >> 1; if (compare(items[parent], value) <= 0) break; items[index] = items[parent]; index = parent; } + items[index] = value; + }, + pop(): T | undefined { + if (!items.length) return undefined; + const first = items[0], last = items.pop()!; + if (items.length) { + let index = 0; + while (index * 2 + 1 < items.length) { + let child = index * 2 + 1; + if (child + 1 < items.length && compare(items[child + 1], items[child]) < 0) child++; + if (compare(last, items[child]) <= 0) break; + items[index] = items[child]; index = child; + } + items[index] = last; + } + return first; + }, + clear() { items.length = 0; }, + }; +} diff --git a/tests/headless/browser-storage-cases.mjs b/tests/headless/browser-storage-cases.mjs new file mode 100644 index 0000000..5113b11 --- /dev/null +++ b/tests/headless/browser-storage-cases.mjs @@ -0,0 +1,101 @@ +import { createBrowserStorage } from '../../src/lib/canvas-runtime/storage.ts'; +import { openProjectEngine } from '../../src/lib/canvas-runtime/engine.ts'; +import { createDocument, createRegistry, builtinComponents, applyChanges } from '../../src/lib/canvas-core/index.ts'; + +async function engineRecoveryCase(name, assert) { + const scope = { accountId: 'account', workspaceId: 'workspace', projectId: 'project' }, registry = createRegistry(builtinComponents()); + const document = createDocument('project', [{ id: 'canvas', parent: null }], [{ id: 'element', incarnation: 1, ownerCanvasId: 'canvas', componentId: 'text', dataVersion: 1, structure: 'node', geometry: { x: 0, y: 0, width: 100, height: 100, rotation: 0 }, data: { title: 'initial', text: '' } }]); + let server = document; const receipts = new Map(), calls = []; + const port = { + compile(intent, requestId, snapshot) { return { payload: { intent, requestId }, canvasIds: intent.canvasIds, expectedVersions: Object.fromEntries(snapshot.runtime.versions) }; }, + async submit(request) { calls.push(['submit', request]); const intent = JSON.parse(request.body).intent; server = applyChanges(server, intent.changes); receipts.set(request.id, { document: server, versions: new Map([['canvas', 2]]) }); return { outcome: 'unresolved' }; }, + async query(request) { calls.push(['query', request]); return { outcome: 'applied', canonical: { receiptId: request.id } }; }, + canonical(request) { return receipts.get(request.id); }, + }; + const storage = createBrowserStorage({ name: `${name}-engine`, checkpointEvery: 4 }), options = { scope, registry, document, versions: new Map([['canvas', 1]]), storage, persistence: port }; + let engine; + try { + engine = await openProjectEngine(options); + const detach = engine.sessions.attach('view'); + engine.editor.dispatch({ type: 'element.data', target: { id: 'element', incarnation: 1 }, patch: { title: 'durable edit' } }); + detach(); await engine.settle(); + assert(calls.length === 1, 'detaching last view must not abort persistent work'); + const request = calls[0][1]; + await engine.close(); + const futureClock = { now: () => Date.now() + 100_000, setTimer: (callback, delay) => setTimeout(callback, delay), clearTimer: handle => clearTimeout(handle) }; + engine = await openProjectEngine({ ...options, clock: futureClock }); + await new Promise(resolve => setTimeout(resolve, 10)); await engine.settle(); + assert(calls.length === 2 && calls[1][0] === 'query', 'recovered dispatched request must query before a new submit'); + assert(calls[1][1].id === request.id && calls[1][1].body === request.body, 'request ID/body must survive real IndexedDB recovery exactly'); + assert(engine.getSnapshot().document.elements.get('element').data.title === 'durable edit', 'recovered pending changes must not replay twice'); + assert(engine.getSnapshot().history.undo.length === 1 && engine.getSnapshot().operations.size === 1, 'history and ledger must recover the same state cut'); + assert(engine.sessions.getSnapshot().views.size === 0, 'transient views must not be recovered as document data'); + assert(engine.getDurability().rootRevision === engine.getDurability().durableRootRevision, 'final receipt and replay must cross the durable barrier'); + } finally { await engine?.close(); storage.close(); indexedDB.deleteDatabase(`${name}-engine`); } +} + +export async function run() { + const results = [], name = `headless-test-${crypto.randomUUID()}`; + const scope = { accountId: 'test-account', workspaceId: 'test-workspace', projectId: 'test-project' }; + const storage = createBrowserStorage({ name, checkpointEvery: 3 }); + const image = revision => ({ formatVersion: 1, rootRevision: revision, localRevision: Math.min(revision, 1), payload: { document: { title: `v${revision}` }, ledger: { frozenBody: '{"request_id":"stable"}' }, history: ['history-id'] } }); + const assert = (condition, message) => { if (!condition) throw new Error(message); }; + const rejects = async (callback, message) => { try { await callback(); } catch (error) { assert(error.message === message, `expected ${message}, received ${error.message}`); return; } throw new Error(`expected ${message}`); }; + let leader, other; + try { + leader = await storage.claim(scope); + assert(await storage.recover(leader) === null, 'empty scope must be distinguishable from a corrupt checkpoint'); + await rejects(() => storage.claim(scope), 'storage_scope_in_use'); + await storage.commit(leader, { image: image(0), files: [{ id: 'asset', bytes: new Blob(['original']) }] }); + assert((await storage.recover(leader)).rootRevision === 0, 'initial checkpoint should recover'); + assert(await (await storage.readFile(leader, 'asset')).text() === 'original', 'committed bytes should survive'); + results.push('initial checkpoint and file bytes commit together; duplicate leader rejected'); + + await storage.commit(leader, { image: image(1) }); + await storage.commit(leader, { image: image(1) }); + await rejects(() => storage.commit(leader, { image: { ...image(1), payload: null } }), 'storage_revision_conflict'); + await rejects(() => storage.commit(leader, { image: image(3) }), 'storage_revision_gap'); + await rejects(() => storage.commit(leader, { image: { ...image(2), formatVersion: 2 } }), 'unsupported_recovery_version'); + results.push('same-cut retry is idempotent; conflicting, skipped and future-format cuts reject'); + + // The first put succeeds before a later duplicate resource aborts the whole transaction. + await rejects(() => storage.commit(leader, { image: image(2), files: [{ id: 'orphan', bytes: new Blob(['must-rollback']) }, { id: 'asset', bytes: new Blob(['replacement']) }] }), 'resource_identity_conflict'); + assert(await storage.readFile(leader, 'orphan') === undefined, 'aborted resource bytes must roll back'); + assert((await storage.recover(leader)).rootRevision === 1, 'aborted document/history/ledger cut must not advance'); + assert(await (await storage.readFile(leader, 'asset')).text() === 'original', 'existing resource bytes must remain immutable'); + results.push('late transaction failure rolls back new bytes and the entire state cut'); + + await storage.commit(leader, { image: image(2) }); + const originalFence = leader.fence; + await leader.release(); + other = createBrowserStorage({ name, checkpointEvery: 3 }); + const next = await other.claim(scope); + await rejects(() => storage.commit(leader, { image: image(3) }), 'storage_leader_lost'); + assert(next.fence > originalFence, 'reacquire must advance fencing generation'); + assert((await other.recover(next)).rootRevision === 2, 'new adapter must replay the durable journal'); + await other.commit(next, { image: image(3) }); + assert((await other.recover(next)).payload.document.title === 'v3', 'compacted checkpoint must cover the same document/history/ledger cut'); + leader = next; + results.push('fresh adapter replays journal; released handle is fenced; checkpoint compaction preserves latest state'); + + // Deliberately remove a journal item to exercise the corruption boundary, not an empty fallback. + await other.commit(next, { image: image(4) }); + const database = await new Promise((resolve, reject) => { const request = indexedDB.open(name); request.onsuccess = () => resolve(request.result); request.onerror = () => reject(request.error); }); + const storageKey = [JSON.stringify([scope.accountId, scope.workspaceId, scope.projectId]), 4]; + const tamper = database.transaction('journal', 'readwrite'), read = tamper.objectStore('journal').get(storageKey); + let intact; + read.onsuccess = () => { intact = read.result; const corrupted = structuredClone(intact); corrupted.image.payload.document.title = 'altered'; tamper.objectStore('journal').put(corrupted, storageKey); }; + await new Promise((resolve, reject) => { tamper.oncomplete = resolve; tamper.onabort = () => reject(tamper.error); }); + await rejects(() => other.recover(next), 'recovery_checksum_mismatch'); + results.push('well-formed data tampering is detected by the stored checksum'); + const transaction = database.transaction('journal', 'readwrite'); + transaction.objectStore('journal').delete(storageKey); + await new Promise((resolve, reject) => { transaction.oncomplete = resolve; transaction.onabort = () => reject(transaction.error); }); + database.close(); + await rejects(() => other.recover(next), 'recovery_journal_gap'); + results.push('journal gap is reported as corruption and never converted to an empty document'); + await engineRecoveryCase(name, assert); + results.push('project engine reopens real IndexedDB with one history unit and the original request; ACK resolves without duplicate edit'); + return results; + } finally { await leader?.release(); other?.close(); storage.close(); indexedDB.deleteDatabase(name); } +} diff --git a/tests/headless/browser-storage.mjs b/tests/headless/browser-storage.mjs new file mode 100644 index 0000000..1c0ee73 --- /dev/null +++ b/tests/headless/browser-storage.mjs @@ -0,0 +1,55 @@ +/** Real Chromium IndexedDB/Web Locks checks, in an isolated temporary browser profile. */ +import { createServer } from 'node:http'; +import { readFile, mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { resolve, join } from 'node:path'; +import { spawn } from 'node:child_process'; +import { stripTypeScriptTypes } from 'node:module'; +import { setTimeout as pause } from 'node:timers/promises'; + +const root = resolve(import.meta.dirname, '../..'), profile = await mkdtemp(join(tmpdir(), 'endoleap-storage-browser-')); +const server = createServer(async (request, response) => { + try { + const path = new URL(request.url, 'http://localhost').pathname; + if (path === '/') { response.setHeader('Content-Type', 'text/html'); response.end('Headless storage checks'); return; } + const filename = resolve(root, `.${path}`); + if (!filename.startsWith(`${root}/src/lib/`) && !filename.startsWith(`${root}/tests/headless/`)) { response.writeHead(404).end(); return; } + let source = await readFile(filename, 'utf8'); + if (filename.endsWith('.ts')) source = stripTypeScriptTypes(source); + response.setHeader('Content-Type', 'text/javascript'); response.end(source); + } catch (error) { response.writeHead(500).end(String(error)); } +}); +await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)); +const chrome = spawn(process.env.CHROME_BIN ?? '/Applications/Google Chrome.app/Contents/MacOS/Google Chrome', ['--headless=new', '--no-first-run', '--disable-background-networking', '--disable-component-update', '--remote-debugging-port=0', `--user-data-dir=${profile}`, `http://127.0.0.1:${server.address().port}/`], { stdio: 'ignore' }); +let socket; +try { + let port; + for (let attempt = 0; attempt < 100; attempt++) { + try { port = Number((await readFile(join(profile, 'DevToolsActivePort'), 'utf8')).split('\n')[0]); break; } catch { await pause(100); } + } + if (!port) throw new Error('isolated Chromium did not start'); + const tabs = await fetch(`http://127.0.0.1:${port}/json/list`).then(response => response.json()); + socket = new WebSocket(tabs.find(tab => tab.type === 'page').webSocketDebuggerUrl); + await new Promise((resolve, reject) => { socket.onopen = resolve; socket.onerror = reject; }); + let next = 0; const pending = new Map(); + socket.onmessage = event => { const message = JSON.parse(event.data); if (message.id) { const callbacks = pending.get(message.id); pending.delete(message.id); if (message.error) callbacks.reject(new Error(JSON.stringify(message.error))); else callbacks.resolve(message.result); } }; + const call = (method, params) => new Promise((resolve, reject) => { const id = ++next; pending.set(id, { resolve, reject }); socket.send(JSON.stringify({ id, method, params })); }); + const origin = `http://127.0.0.1:${server.address().port}`; + await call('Page.navigate', { url: `${origin}/` }); + let loaded = false; + for (let attempt = 0; attempt < 100; attempt++) { + const state = await call('Runtime.evaluate', { expression: `location.origin === ${JSON.stringify(origin)} && document.readyState === 'complete'`, returnByValue: true }); + if (state.result?.value === true) { loaded = true; break; } + await pause(50); + } + if (!loaded) throw new Error('browser test origin did not finish loading'); + const result = await call('Runtime.evaluate', { expression: "import('/tests/headless/browser-storage-cases.mjs').then(module => module.run())", awaitPromise: true, returnByValue: true, timeout: 30000 }); + if (result.exceptionDetails) throw new Error(result.exceptionDetails.exception?.description ?? result.exceptionDetails.text); + for (const message of result.result.value) console.log(`PASS ${message}`); +} finally { + socket?.close(); + const exited = new Promise(resolve => { if (chrome.exitCode !== null) resolve(); else chrome.once('exit', resolve); }); + chrome.kill(); await exited; + await new Promise(resolve => server.close(resolve)); + await rm(profile, { recursive: true, force: true, maxRetries: 3 }); +} diff --git a/tests/headless/engine.test.mjs b/tests/headless/engine.test.mjs new file mode 100644 index 0000000..e7182f9 --- /dev/null +++ b/tests/headless/engine.test.mjs @@ -0,0 +1,129 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { createDocument, applyChanges, builtinComponents, createRegistry } from '../../src/lib/canvas-core/index.ts'; +import { openProjectEngine } from '../../src/lib/canvas-runtime/engine.ts'; +import { encodeRecovery, decodeRecovery } from '../../src/lib/canvas-runtime/recovery.ts'; + +const registry = createRegistry(builtinComponents()); +const original = createDocument('project', [{ id: 'canvas', parent: null }], [{ id: 'a', incarnation: 1, structure: 'node', ownerCanvasId: 'canvas', componentId: 'text', dataVersion: 1, data: { title: 'a', text: '' }, geometry: { x: 0, y: 0, width: 100, height: 100, rotation: 0 } }]); +const scope = { accountId: 'account', workspaceId: 'workspace', projectId: 'project' }, target = { id: 'a', incarnation: 1 }; +const deferred = () => { let resolve; const promise = new Promise(yes => { resolve = yes; }); return { promise, resolve }; }; +const tick = async () => { for (let i = 0; i < 30; i++) await Promise.resolve(); }; +function storageFixture() { + let image = null, owner = null, fence = 0; + const saved = [], fixture = { + broken: false, saved, + async claim(scope) { assert.equal(owner, null); const leader = { scope, fence: ++fence, active: true, async release() { owner = null; leader.active = false; } }; owner = leader; return leader; }, + async recover() { return image && structuredClone(image); }, + async commit(leader, batch) { if (fixture.broken) throw new Error('quota'); assert.equal(owner, leader); assert.equal(leader.active, true); if (image) assert.ok(batch.image.rootRevision === image.rootRevision + 1 || batch.image.rootRevision === image.rootRevision); image = structuredClone(batch.image); saved.push(image); return { durableRootRevision: image.rootRevision }; }, + async readFile() {}, close() {}, + image: () => structuredClone(image), owner: () => owner, + }; + return fixture; +} +function persistenceFixture(overrides = {}) { + let server = original, version = 1; + const calls = [], receipts = new Map(); + const fixture = { + calls, receipts, + compile(intent, requestId, snapshot) { return { payload: { requestId, intent }, canvasIds: intent.canvasIds, expectedVersions: Object.fromEntries(intent.canvasIds.map(id => [id, snapshot.runtime.versions.get(id)])) }; }, + async submit(request) { + calls.push(['submit', request]); + if (!receipts.has(request.id)) { const { intent } = JSON.parse(request.body); server = applyChanges(server, intent.changes); version++; receipts.set(request.id, { document: server, versions: new Map([['canvas', version]]) }); } + return { outcome: 'applied', canonical: { receipt: request.id } }; + }, + async query(request) { calls.push(['query', request]); return receipts.has(request.id) ? { outcome: 'applied', canonical: { receipt: request.id } } : { outcome: 'unresolved' }; }, + canonical(request) { return receipts.get(request.id); }, ...overrides, + }; + return fixture; +} +async function open(storage, persistence, extra = {}) { + let next = 0; const errors = []; + const engine = await openProjectEngine({ scope, registry, document: original, versions: new Map([['canvas', 1]]), storage, persistence, createId: () => `identity-${++next}`, onError: error => errors.push(error), ...extra }); + return { engine, errors }; +} + +test('local document/history/ledger publish together; ACK replays edits made while a request is in flight', async () => { + const storage = storageFixture(), port = persistenceFixture(), gate = deferred(), realSubmit = port.submit; + port.submit = async request => { if (!port.calls.length) await gate.promise; return realSubmit(request); }; + const { engine } = await open(storage, port); await engine.persist(); + const first = engine.editor.dispatch({ type: 'element.data', target, patch: { title: 'local' } }); + assert.equal(engine.getSnapshot().history.undo.length, 1); + assert.ok(engine.getSnapshot().runtime.ledger.operations.has(first.operationIds[0])); + await tick(); + engine.editor.dispatch({ type: 'element.geometry', target, patch: { x: 200 } }); + assert.equal(engine.getSnapshot().document.elements.get('a').geometry.x, 200); + gate.resolve(); await engine.settle(); + assert.equal(port.calls.length, 2); + assert.equal(port.calls[1][1].expectedVersions.canvas, 2); + assert.equal(engine.getSnapshot().document.elements.get('a').data.title, 'local'); + assert.equal(engine.getSnapshot().document.elements.get('a').geometry.x, 200); + assert.equal(engine.getSnapshot().history.undo.length, 2); + assert.ok([...engine.getSnapshot().runtime.ledger.operations.values()].every(operation => operation.phase === 'resolved')); + assert.equal(engine.getDurability().durableRootRevision, engine.getSnapshot().rootRevision); + assert.equal(engine.getSnapshot().runtime.ledger.durableGeneration, engine.getSnapshot().runtime.ledger.generation); + const payload = storage.image().payload; + assert.equal(payload.history.undo.length, 2); assert.equal(payload.runtime.ledger.requests.length, 2); + await engine.close(); +}); + +test('closing a view keeps jobs alive; closing/reopening the engine restores history and queries the original unknown request', async () => { + const storage = storageFixture(), port = persistenceFixture(), realSubmit = port.submit; + port.submit = async request => { await realSubmit(request); return { outcome: 'unresolved' }; }; + const { engine } = await open(storage, port); + const detach = engine.sessions.attach('view'); + engine.editor.dispatch({ type: 'element.data', target, patch: { title: 'once' } }); detach(); + await engine.settle(); + assert.equal(port.calls.length, 1); const request = port.calls[0][1]; + await engine.close(); + const futureClock = { now: () => Date.now() + 100_000, setTimer: (callback, delay) => setTimeout(callback, delay), clearTimer: clearTimeout }; + const recovered = await open(storage, port, { clock: futureClock }); + await new Promise(resolve => setTimeout(resolve, 1)); await recovered.engine.settle(); + assert.equal(port.calls[1][0], 'query'); assert.equal(port.calls[1][1].id, request.id); assert.equal(port.calls[1][1].body, request.body); + assert.equal(recovered.engine.getSnapshot().history.undo.length, 1); + assert.equal(recovered.engine.getSnapshot().document.elements.get('a').data.title, 'once'); + assert.equal(recovered.engine.getSnapshot().operations.size, 1); + await recovered.engine.close(); +}); + +test('failed local durability preserves editing, prevents sends, and failed close retains the recovery leader', async () => { + const storage = storageFixture(), port = persistenceFixture(), { engine, errors } = await open(storage, port); + await engine.persist(); storage.broken = true; + engine.editor.dispatch({ type: 'element.data', target, patch: { title: 'unsaved' } }); await tick(); + await assert.rejects(() => engine.persist(), /quota/); await tick(); + assert.equal(engine.getSnapshot().document.elements.get('a').data.title, 'unsaved'); assert.equal(port.calls.length, 0); assert.ok(errors.length); + await assert.rejects(() => engine.close(), /quota/); assert.ok(storage.owner()); + assert.equal(engine.editor.dispatch({ type: 'element.data', target, patch: { title: 'too-late' } }).code, 'engine_closing'); + storage.broken = false; await engine.close(); assert.equal(storage.owner(), null); +}); + +test('canonical replay conflict preserves working content and the known receipt without releasing the writer', async () => { + const storage = storageFixture(), port = persistenceFixture(), gate = deferred(), realSubmit = port.submit; + port.submit = async request => { await gate.promise; return realSubmit(request); }; + const realCanonical = port.canonical; + port.canonical = request => { const proposal = realCanonical(request); return { ...proposal, document: applyChanges(proposal.document, [{ kind: 'field', target, field: 'geometry', before: original.elements.get('a').geometry, after: { ...original.elements.get('a').geometry, x: 500 } }]) }; }; + const { engine } = await open(storage, port); + engine.editor.dispatch({ type: 'element.data', target, patch: { title: 'first' } }); await tick(); + engine.editor.dispatch({ type: 'element.geometry', target, patch: { x: 200 } }); + gate.resolve(); await engine.settle(); + assert.equal(engine.getSnapshot().document.elements.get('a').geometry.x, 200); + assert.equal(engine.getSnapshot().runtime.conflicts.size, 1); assert.equal(port.calls.length, 1); + const request = [...engine.getSnapshot().runtime.ledger.requests.values()][0]; + assert.equal(request.outcome, 'applied'); assert.equal(request.phase, 'waiting'); + assert.ok(engine.getSnapshot().runtime.receipts.has(request.id)); + await engine.close(); +}); + +test('recovery codec restores immutable maps and rejects broken history/dependency references', async () => { + const storage = storageFixture(), port = persistenceFixture(), { engine } = await open(storage, port); + engine.editor.dispatch({ type: 'element.data', target, patch: { title: 'encoded' } }); await engine.settle(); + const image = encodeRecovery(engine.getSnapshot()), restored = decodeRecovery(image, registry); + assert.equal(typeof restored.document.elements.set, 'undefined'); assert.equal(restored.history.undo.length, 1); + const corrupted = structuredClone(image); corrupted.payload.history.undo[0] = 'missing'; + assert.throws(() => decodeRecovery(corrupted, registry), /corrupt_recovery_image/); + const cycle = structuredClone(image), record = cycle.payload.runtime.ledger.operations[0][1]; record.intent.dependencies = [record.intent.id]; + assert.throws(() => decodeRecovery(cycle, registry), /corrupt_recovery_image/); + const changed = structuredClone(image); changed.payload.document.elements[0].data.title = 'mutated'; + assert.equal(restored.document.elements.get('a').data.title, 'encoded'); + await engine.close(); +}); diff --git a/tests/headless/ledger.test.mjs b/tests/headless/ledger.test.mjs new file mode 100644 index 0000000..f762f9d --- /dev/null +++ b/tests/headless/ledger.test.mjs @@ -0,0 +1,59 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { createDeque, createMinHeap } from '../../src/lib/canvas-runtime/structures.ts'; +import { createLedger, compressChanges } from '../../src/lib/canvas-runtime/ledger.ts'; +const node = { id: 'a', incarnation: 1, ownerCanvasId: 'root', componentId: 'text', dataVersion: 1, data: { title: 'a', text: '' }, structure: 'node', geometry: { x: 0, y: 0, width: 20, height: 20, rotation: 0 } }; +const intent = (id, changes = [], dependencies = []) => ({ id, operationType: 'document.change', operationSchemaVersion: 1, sequence: 1, origin: 'user', changes, dependencies }); +const frozen = ledger => ledger.freeze({ id: 'request', operationIds: ['create'], canvasIds: ['root'], expectedVersions: { root: 1 }, payload: { changes: [] } }); + +test('FIFO retains order across compaction; retry heap orders deadline then sequence', () => { + const queue = createDeque(); for (let i = 0; i < 4000; i++) queue.push(i); + for (let i = 0; i < 4000; i++) assert.equal(queue.pop(), i); + assert.equal(queue.size, 0); assert.equal(queue.pop(), undefined); + const heap = createMinHeap((a, b) => a.due - b.due || a.sequence - b.sequence); + for (const value of [{ due: 5, sequence: 2 }, { due: 2, sequence: 9 }, { due: 5, sequence: 1 }, { due: 1, sequence: 8 }]) heap.push(value); + assert.deepEqual([heap.pop(), heap.pop(), heap.pop(), heap.pop()].map(item => [item.due, item.sequence]), [[1,8],[2,9],[5,1],[5,2]]); + assert.equal(heap.pop(), undefined); +}); + +test('wholly unsent create-data-delete disappears but a position update preserves the title', () => { + const create = { kind: 'element', id: 'a', before: null, after: node }; + const rename = { kind: 'field', target: { id: 'a', incarnation: 1 }, field: 'data.title', before: 'a', after: 'renamed', beforePresent: true, afterPresent: true }; + const move = { kind: 'field', target: { id: 'a', incarnation: 1 }, field: 'geometry', before: node.geometry, after: { ...node.geometry, x: 100 } }; + const compressed = compressChanges([create, rename, move]); + assert.equal(compressed.length, 1); assert.equal(compressed[0].after.data.title, 'renamed'); + assert.deepEqual(compressChanges([...compressed, { kind: 'element', id: 'a', before: compressed[0].after, after: null }]), []); +}); + +test('frozen identity remains exact after unknown/accepted outcomes and deletion is a separate pending operation', () => { + const ledger = createLedger(); ledger.add(intent('create'), { a: 'absent' }); + const request = frozen(ledger), body = request.body; + ledger.markDurable(ledger.getSnapshot().generation); ledger.markDispatch('request'); + ledger.result('request', 'unresolved', 'transport_unknown'); + ledger.add(intent('delete', [], ['create']), { a: 'exists' }); + assert.equal(ledger.eligible('delete'), false); assert.equal(ledger.possiblyApplied('request'), true); + ledger.result('request', 'accepted'); + assert.equal(ledger.eligible('delete'), false); assert.equal(ledger.getSnapshot().requests.get('request').body, body); + ledger.markDispatch('request'); assert.equal(ledger.getSnapshot().requests.get('request').body, body); + ledger.result('request', 'applied'); assert.equal(ledger.eligible('delete'), true); +}); + +test('dependencies are explicit, duplicate identities conflict and late durable completion does not mark newer work durable', () => { + const ledger = createLedger(); + assert.throws(() => ledger.add(intent('child', [], ['missing']), {}), /dependency_missing/); + ledger.add(intent('first'), {}); const oldGeneration = ledger.getSnapshot().generation; + ledger.add(intent('second', [], ['first']), {}); ledger.markDurable(oldGeneration); + assert.ok(ledger.getSnapshot().durableGeneration < ledger.getSnapshot().generation); + assert.throws(() => ledger.add(intent('first', [], ['second']), {}), /operation_identity_conflict/); + assert.equal(ledger.eligible('second'), false); +}); + +test('recovery retains possible application; explicit terminal zero-write does not pretend dependent writes succeeded', () => { + const original = createLedger(); original.add(intent('create'), {}); frozen(original); original.markDispatch('request'); + const restored = createLedger(original.getSnapshot()); + assert.equal(restored.possiblyApplied('request'), true); + restored.result('request', 'not-applied', 'permission'); + assert.equal(restored.possiblyApplied('request'), false); + assert.equal(restored.getSnapshot().operations.get('create').phase, 'blocked'); + assert.throws(() => restored.result('request', 'applied'), /receipt_conflict/); +}); diff --git a/tests/headless/scheduler.test.mjs b/tests/headless/scheduler.test.mjs new file mode 100644 index 0000000..fdceea7 --- /dev/null +++ b/tests/headless/scheduler.test.mjs @@ -0,0 +1,100 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { createLedger } from '../../src/lib/canvas-runtime/ledger.ts'; +import { createScheduler } from '../../src/lib/canvas-runtime/scheduler.ts'; + +const deferred = () => { let resolve, reject; const promise = new Promise((yes, no) => { resolve = yes; reject = no; }); return { promise, resolve, reject }; }; +const tick = async () => { for (let i = 0; i < 20; i++) await Promise.resolve(); }; +function setup(overrides = {}, initial) { + let ledger = createLedger(initial), time = 1000, next = 0, timerId = 0; + const timers = new Map(), calls = [], persisted = [], errors = []; + const scheduler = createScheduler({ + getLedger: () => ledger.getSnapshot(), + mutate(reduce) { const draft = createLedger(ledger.getSnapshot()); reduce(draft); ledger = draft; }, + async persist() { persisted.push(ledger.getSnapshot()); }, + createId: () => `request-${++next}`, + participants: intent => intent.changes.map(change => change.id), + compile: (intent, requestId) => ({ canvasIds: intent.changes.map(change => change.id), expectedVersions: Object.fromEntries(intent.changes.map(change => [change.id, 1])), payload: { requestId, operationId: intent.id } }), + async submit(request) { calls.push(['submit', request]); return { outcome: 'applied' }; }, + async query(request) { calls.push(['query', request]); return { outcome: 'applied' }; }, + complete(request, status) { const draft = createLedger(ledger.getSnapshot()); draft.result(request.id, status.outcome, status.error); ledger = draft; }, + clock: { now: () => time, setTimer(callback, delay) { const id = ++timerId; timers.set(id, { callback, at: time + delay }); return id; }, clearTimer: id => timers.delete(id) }, + retryDelay: () => 100, + onError: error => errors.push(error), ...overrides, + }); + return { scheduler, calls, persisted, errors, ledger: () => ledger, + add(id, canvases = ['canvas'], dependencies = []) { ledger.add({ id, operationType: 'document.change', operationSchemaVersion: 1, sequence: ledger.getSnapshot().operations.size + 1, origin: 'user', changes: canvases.map(id => ({ kind: 'canvas', id, before: null, after: { id, parent: null } })), dependencies }, {}); scheduler.notify([id]); }, + async advance(ms) { time += ms; for (const [id, timer] of [...timers]) if (timer.at <= time) { timers.delete(id); timer.callback(); } await tick(); await scheduler.settle(); }, + timerCount: () => timers.size, + }; +} + +test('freeze and dispatch evidence must be durable before submit, then dependencies wake', async () => { + const barrier = deferred(), sent = [], state = setup({ persist: () => barrier.promise, submit: async request => { sent.push(request); return { outcome: 'applied' }; } }); + state.add('create'); state.add('delete', ['canvas'], ['create']); state.scheduler.start(); await tick(); + assert.equal(sent.length, 0); + const frozen = [...state.ledger().getSnapshot().requests.values()][0]; + assert.equal(frozen.phase, 'dispatch-marked'); assert.equal(state.scheduler.getClaims().get('canvas'), frozen.id); + barrier.resolve(); await state.scheduler.settle(); + assert.equal(sent.length, 2); assert.equal(state.ledger().getSnapshot().operations.get('delete').phase, 'resolved'); + assert.equal(state.scheduler.getClaims().size, 0); state.scheduler.stop(); +}); + +test('unknown request holds its canvas; another canvas proceeds; accepted stays on query path', async () => { + const calls = []; let queries = 0; + const state = setup({ + async submit(request) { calls.push(['submit', request]); return request.operationIds[0] === 'create' ? { outcome: 'unresolved' } : { outcome: 'applied' }; }, + async query(request) { calls.push(['query', request]); return { outcome: ++queries === 1 ? 'accepted' : 'applied' }; }, + }); + state.add('create'); state.add('same-canvas'); state.add('independent', ['other']); state.scheduler.start(); await state.scheduler.settle(); + assert.deepEqual(calls.map(([_, request]) => request.operationIds[0]), ['create', 'independent']); + assert.equal(state.scheduler.getClaims().size, 1); assert.equal(state.timerCount(), 1); + await state.advance(100); assert.equal(calls.at(-1)[0], 'query'); + assert.equal(state.ledger().getSnapshot().operations.get('create').phase, 'frozen'); + await state.advance(100); + assert.deepEqual(calls.map(([action, request]) => [action, request.operationIds[0]]), [['submit','create'], ['submit','independent'], ['query','create'], ['query','create'], ['submit','same-canvas']]); + assert.equal(state.scheduler.getClaims().size, 0); state.scheduler.stop(); +}); + +test('receipt missing after recovery triggers exact retry, never a new identity/body', async () => { + const original = setup({ submit: async () => ({ outcome: 'unresolved' }) }); + original.add('create'); original.scheduler.start(); await original.scheduler.settle(); + const saved = original.ledger().getSnapshot(), before = saved.requests.get('request-1'); original.scheduler.stop(); + const calls = []; + const restored = setup({ query: async request => { calls.push(['query', request]); return { outcome: 'unresolved', error: 'receipt_unknown' }; }, submit: async request => { calls.push(['submit', request]); return { outcome: 'applied' }; } }, saved); + restored.scheduler.start(); await restored.advance(100); await restored.advance(100); + assert.deepEqual(calls.map(([action]) => action), ['query', 'submit']); + for (const [, request] of calls) { assert.equal(request.id, before.id); assert.equal(request.body, before.body); } + assert.equal(restored.ledger().getSnapshot().requests.size, 1); restored.scheduler.stop(); +}); + +test('storage failure never sends and keeps claims; explicit resume retries persistence first', async () => { + let broken = true; const sent = []; + const state = setup({ persist: async () => { if (broken) throw new Error('quota'); }, submit: async request => { sent.push(request); return { outcome: 'applied' }; } }); + state.add('create'); state.scheduler.start(); await state.scheduler.settle(); + assert.equal(sent.length, 0); assert.equal(state.errors[0].message, 'quota'); assert.equal(state.scheduler.getClaims().size, 1); + broken = false; await state.scheduler.resume(); await state.scheduler.settle(); + // Dispatch evidence means query first, even if the failed barrier preceded the network call. + assert.equal(state.calls[0][0], 'query'); assert.equal(sent.length, 0); assert.equal(state.scheduler.getClaims().size, 0); state.scheduler.stop(); +}); + +test('multi-canvas claims are atomic and stop fences late results', async () => { + const response = deferred(), state = setup({ submit: () => response.promise }); + state.add('cross', ['a', 'b']); state.add('single', ['b']); state.scheduler.start(); await tick(); + assert.equal(new Set(state.scheduler.getClaims().values()).size, 1); assert.equal(state.scheduler.getClaims().size, 2); + state.scheduler.stop(); response.resolve({ outcome: 'applied' }); await state.scheduler.settle(); + assert.equal(state.ledger().getSnapshot().requests.get('request-1').outcome, 'unresolved'); + assert.throws(() => state.scheduler.start(), /scheduler_closed/); +}); + +test('missing frozen-request interpreter preserves bytes and claims while unrelated work proceeds', async () => { + const initial = setup({ submit: async () => ({ outcome: 'unresolved' }) }); initial.add('create'); initial.scheduler.start(); await initial.scheduler.settle(); initial.scheduler.stop(); + const saved = initial.ledger().getSnapshot(), requests = new Map(saved.requests), old = requests.get('request-1'); + requests.set(old.id, { ...old, protocolVersion: 99 }); + const restored = setup({ createId: () => 'other-request' }, { ...saved, requests }); + restored.add('other', ['other']); restored.scheduler.start(); await restored.scheduler.settle(); + assert.equal(restored.calls.length, 1); assert.equal(restored.calls[0][1].id, 'other-request'); + assert.equal(restored.ledger().getSnapshot().requests.get(old.id).body, old.body); + assert.equal(restored.ledger().getSnapshot().requests.get(old.id).error, 'request_interpreter_unavailable'); + assert.equal(restored.scheduler.getClaims().get('canvas'), old.id); restored.scheduler.stop(); +});