From 8a76e53a9fe783563e927650a7d36c96d18d7466 Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Sat, 7 Nov 2026 23:24:00 +0800 Subject: [PATCH] Persist staged image uploads alongside document operations --- src/app/api/foundry/[...path]/route.ts | 7 +- src/lib/canvas-core/editor.ts | 7 +- src/lib/canvas-core/features.ts | 4 +- src/lib/canvas-core/registry.ts | 2 +- src/lib/canvas-core/types.ts | 3 +- src/lib/canvas-runtime/contracts.ts | 3 + src/lib/canvas-runtime/engine.ts | 144 ++++++++++++++++++--- src/lib/canvas-runtime/ledger.ts | 8 +- src/lib/canvas-runtime/reconcile.ts | 3 +- src/lib/canvas-runtime/recovery.ts | 28 +++- src/lib/canvas-runtime/resources.ts | 93 ++++++++++++++ src/lib/canvas-runtime/scheduler.ts | 41 ++++-- src/lib/canvas-service/assets.ts | 62 +++++++++ src/lib/canvas-service/foundry.ts | 4 +- src/lib/canvas-service/index.ts | 1 + tests/headless/asset-service.test.mjs | 50 ++++++++ tests/headless/browser-storage-cases.mjs | 48 +++++++ tests/headless/proxy.integration.mjs | 7 + tests/headless/resources.test.mjs | 156 +++++++++++++++++++++++ tests/headless/scheduler.test.mjs | 40 +++++- 20 files changed, 659 insertions(+), 52 deletions(-) create mode 100644 src/lib/canvas-runtime/resources.ts create mode 100644 src/lib/canvas-service/assets.ts create mode 100644 tests/headless/asset-service.test.mjs create mode 100644 tests/headless/resources.test.mjs diff --git a/src/app/api/foundry/[...path]/route.ts b/src/app/api/foundry/[...path]/route.ts index 6ac0e45..cd2412b 100644 --- a/src/app/api/foundry/[...path]/route.ts +++ b/src/app/api/foundry/[...path]/route.ts @@ -5,7 +5,7 @@ import { NextRequest, NextResponse } from "next/server"; const allowed = /^(profile|auth\/(login|register|verify-email|exchange|me|logout)|workspaces(?:\/[a-zA-Z0-9_-]+(?:\/(?:materials(?:\/[a-zA-Z0-9_-]+(?:\/canvas)?)?|search))?)?)$/; const canvasRoute = /^workspaces\/[a-zA-Z0-9_-]+\/materials\/[a-zA-Z0-9_-]+\/canvas(?:\/(?:model(?:\/(?:migrate|operations(?:\/[a-zA-Z0-9_-]+)?))?|operations(?:\/[a-zA-Z0-9_-]+)?|connections|capabilities|subgraphs(?:\/(?:migrate|[a-zA-Z0-9_-]+\/(?:members|dissolve)))?|transactions\/[a-zA-Z0-9_-]+\/undo|ai\/(?:models|runs(?:\/[a-zA-Z0-9_-]+(?:\/(?:cancel|apply))?)?)))?$/; -const assetRoute = /^workspaces\/[a-zA-Z0-9_-]+\/assets(?:\/[a-zA-Z0-9_-]+)?$/; +const assetRoute = /^workspaces\/[a-zA-Z0-9_-]+\/assets(?:\/(?:operations\/[a-zA-Z0-9_-]+|[a-zA-Z0-9_-]+))?$/; const bodyLimit = 1024 * 1024; // Route Handlers do not have the Pages Router bodyParser size limit. @@ -53,7 +53,8 @@ async function handler( } const path = (await params).path.join("/"); if (!allowed.test(path) && !canvasRoute.test(path) && !assetRoute.test(path)) return json({ error: "not_found" }, 404); - const asset = assetRoute.test(path); + const assetReceipt = assetRoute.test(path) && /\/assets\/operations\//.test(path); + const asset = assetRoute.test(path) && !assetReceipt; const canvas = canvasRoute.test(path); const canonical = canvas && /\/canvas\/model(?:\/|$)/.test(path); let requestBody: unknown; @@ -72,7 +73,7 @@ async function handler( if (canvas) headers.Accept = "application/vnd.foundry.canvas+json;version=2"; const session = req.cookies.get("foundry_session")?.value; if (session) headers.Authorization = "Bearer " + session; - for (const h of ["x-foundry-code", "x-foundry-verifier", "if-match", "idempotency-key", "x-expected-account"]) { + for (const h of ["x-foundry-code", "x-foundry-verifier", "if-match", "idempotency-key", "x-expected-account", "x-resource-id"]) { const value = req.headers.get(h); if (value !== null) headers[h] = value; } diff --git a/src/lib/canvas-core/editor.ts b/src/lib/canvas-core/editor.ts index 8d7e034..e96e330 100644 --- a/src/lib/canvas-core/editor.ts +++ b/src/lib/canvas-core/editor.ts @@ -12,6 +12,7 @@ export type EditorOptions = { runtime?: Runtime; /** Installed by the engine. Pure preparation: an error rejects the entire commit. */ reduceRuntime?: (previous: Runtime, next: CoreSnapshot, commit: Commit) => Runtime; + nextOperationSequence?: (runtime: Runtime, fallback: number) => number; }; export type Transaction = { dispatch(command: Command): Availability; getDocument(): Document }; export type Editor = ReturnType; @@ -113,6 +114,8 @@ export function createEditor(options: EditorOptions(); for (const change of changes) { @@ -143,7 +146,7 @@ export function createEditor(options: EditorOptions(options: EditorOptions defineComponent({ id, dataVersion: 1, structure: 'node', validate: textData, features: [movable(), resizable(), ...(id === 'frame' ? [] : [rotatable()]), editable()], - resourceFields: ['image', 'plain-image'].includes(id) ? ['asset'] : [], - })).concat([defineComponent({ id: 'connection', dataVersion: 1, structure: 'connection', validate: data => validJson(data), features: [editable()] })]); + resourceFields: ['image', 'plain-image'].includes(id) ? ['assetId','asset'] : [], + })).concat([defineComponent({ id: 'connection', dataVersion: 1, structure: 'connection', validate: data => validJson(data), features: [editable()], resourceFields: [] })]); } diff --git a/src/lib/canvas-core/registry.ts b/src/lib/canvas-core/registry.ts index cdf5d6d..5ef8ca3 100644 --- a/src/lib/canvas-core/registry.ts +++ b/src/lib/canvas-core/registry.ts @@ -7,7 +7,7 @@ export function defineFeature(id: string, handlers: Partial defineFeature(feature.id, feature.handlers)), migrations: definition.migrations ? { ...definition.migrations } : undefined, resourceFields: [...definition.resourceFields || []] }); + return freeze({ ...definition, features: definition.features.map(feature => defineFeature(feature.id, feature.handlers)), migrations: definition.migrations ? { ...definition.migrations } : undefined, resourceFields: definition.resourceFields ? [...definition.resourceFields] : undefined }); } export function createRegistry(definitions: readonly ComponentDefinition[]): Registry { const components = new Map(), capabilities = new Map>(), handlers = new Map>(); diff --git a/src/lib/canvas-core/types.ts b/src/lib/canvas-core/types.ts index a5db355..c9d0204 100644 --- a/src/lib/canvas-core/types.ts +++ b/src/lib/canvas-core/types.ts @@ -33,7 +33,8 @@ export type Applied = { readonly status: 'applied'; readonly revision: number; r export type Result = Applied | Rejected; export type HistoryEntry = { readonly id: Id; readonly label: string; readonly changes: readonly Change[]; readonly operationIds: readonly Id[]; readonly groupId?: Id }; export type HistoryState = { readonly undo: readonly Id[]; readonly redo: readonly Id[]; readonly entries: ReadonlyMap }; -export type OperationIntent = { readonly id: Id; readonly operationType: 'document.change'; readonly operationSchemaVersion: 1; readonly sequence: number; readonly origin: Origin; readonly canvasIds: readonly Id[]; readonly changes: readonly Change[]; readonly dependencies: readonly Id[] }; +type IntentBase = { readonly id: Id; readonly operationSchemaVersion: 1; readonly sequence: number; readonly origin: Origin; readonly canvasIds: readonly Id[]; readonly changes: readonly Change[]; readonly dependencies: readonly Id[] }; +export type OperationIntent = IntentBase & ({ readonly operationType: 'document.change' } | { readonly operationType: 'asset.upload'; readonly resourceId: Id }); export type Commit = { readonly revision: number; readonly origin: Origin; readonly changes: readonly Change[]; readonly operationIds: readonly Id[] }; export type CoreSnapshot = { readonly revision: number; readonly document: Document; readonly history: HistoryState; readonly operations: ReadonlyMap }; /** One publication boundary. Runtime metadata changes do not dirty the document revision. */ diff --git a/src/lib/canvas-runtime/contracts.ts b/src/lib/canvas-runtime/contracts.ts index 601c97e..fa95322 100644 --- a/src/lib/canvas-runtime/contracts.ts +++ b/src/lib/canvas-runtime/contracts.ts @@ -1,6 +1,7 @@ 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'; +import type { ResourceRecord } from './resources.ts'; export type RuntimeState = { readonly ledger: LedgerSnapshot; @@ -8,6 +9,8 @@ export type RuntimeState = { readonly versions: ReadonlyMap; readonly receipts: ReadonlyMap; readonly conflicts: ReadonlyMap; + readonly resources: ReadonlyMap; + readonly resourcePins: ReadonlyMap; }; export type EngineSnapshot = Snapshot; export type CanonicalProposal = { readonly document: Document; readonly versions: ReadonlyMap }; diff --git a/src/lib/canvas-runtime/engine.ts b/src/lib/canvas-runtime/engine.ts index 1257e2f..913b57a 100644 --- a/src/lib/canvas-runtime/engine.ts +++ b/src/lib/canvas-runtime/engine.ts @@ -1,5 +1,5 @@ import type { Document, Id, Registry } from '../canvas-core/types.ts'; -import { createEditor, type EditorOptions } from '../canvas-core/editor.ts'; +import { createEditor, type EditorOptions, type Transaction } 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'; @@ -8,11 +8,12 @@ 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 { conflictsWith, maxResourceBytes, operationFootprint, referencedResources, type AssetPort, type ResourceRecord, type ResourceHandle, type StagedFile } from './resources.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; + storage: StoragePort; persistence: PersistencePort; assets?: AssetPort; createId?: () => Id; historyLimit?: number; policy?: EditorOptions['policy']; clock?: Clock; onError?: (error: unknown) => void; }; @@ -27,25 +28,39 @@ export async function openProjectEngine(options: EngineOptions) { 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 createId = options.createId ?? (() => crypto.randomUUID()); + const displays = new Set<() => void>(); + const fileWrites = new Map(), digestTasks = new Map>(); + let staged: readonly ResourceRecord[] = []; + 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()), resources: readMap(new Map()), resourcePins: 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, + nextOperationSequence(runtime, fallback) { let next = fallback; for (const record of runtime.ledger.operations.values()) next = Math.max(next, record.intent.sequence + 1); return next + staged.length; }, reduceRuntime(previous, next, commit) { if (!commit.operationIds.length) return previous; - const ledger = createLedger(previous.ledger); + const ledger = createLedger(previous.ledger), resources = new Map(previous.resources); + const firstSequence = next.operations.get(commit.operationIds[0])!.sequence - staged.length; + staged.forEach((resource, index) => { + if (resources.has(resource.id) || ledger.getSnapshot().operations.has(resource.operationId)) throw fault('resource_identity_conflict', [resource.id]); + resources.set(resource.id, resource); + ledger.add(freeze({ id: resource.operationId, operationType: 'asset.upload', operationSchemaVersion: 1, sequence: firstSequence + index, origin: 'user', canvasIds: [], changes: [], dependencies: [], resourceId: resource.id }), {}); + ledger.block(resource.operationId, ['waiting_digest']); + }); 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 intent = next.operations.get(id)!, footprint = operationFootprint(intent, next.document); + const dependencies = new Set(intent.dependencies); + for (const record of ledger.getSnapshot().operations.values()) if (record.phase !== 'resolved' && record.intent.operationType === 'document.change' && conflictsWith(operationFootprint(record.intent, next.document), footprint)) dependencies.add(record.intent.id); + for (const id of referencedResources(intent.changes, options.registry)) { + const resource = resources.get(id); + if (resource && !resource.stored) dependencies.add(resource.operationId); + } const basePresence: Record = Object.create(null); 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); + ledger.add(freeze({ ...intent, dependencies: [...dependencies] }), basePresence); } - return Object.freeze({ ...previous, ledger: ledger.getSnapshot() }); + return Object.freeze({ ...previous, resources: readMap(resources), ledger: ledger.getSnapshot() }); }, }); const pending = new Map(); @@ -56,9 +71,11 @@ export async function openProjectEngine(options: EngineOptions) { const task = (async () => { while (pending.size) { const [revision, snapshot] = pending.entries().next().value!; - const receipt = await options.storage.commit(leader, { image: encodeRecovery(snapshot) }); + const files = [...fileWrites].filter(([id]) => snapshot.runtime.resources.has(id)).map(([id, bytes]) => ({ id, bytes })); + const receipt = await options.storage.commit(leader, { image: encodeRecovery(snapshot), files }); if (receipt.durableRootRevision !== revision) throw fault('invalid_durable_receipt'); durableRootRevision = revision; pending.delete(revision); + for (const file of files) fileWrites.delete(file.id); if (editor.getSnapshot().runtime.ledger.durableGeneration < snapshot.runtime.ledger.generation) mutate(ledger => ledger.markDurable(snapshot.runtime.ledger.generation)); } })(); @@ -73,6 +90,27 @@ export async function openProjectEngine(options: EngineOptions) { 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; } + if (request.kind === 'asset.upload') { + const receipts = new Map(previous.receipts); if (status.canonical !== undefined) receipts.set(request.id, freeze(status.canonical)); + try { + const intent = previous.ledger.operations.get(request.operationIds[0])?.intent; + if (intent?.operationType !== 'asset.upload' || !options.assets) throw fault('asset_interpreter_unavailable'); + const resource = previous.resources.get(intent.resourceId); + if (!resource) throw fault('resource_missing'); + const stored = freeze(options.assets.accept(request, status, resource)); + if (stored.id !== resource.id || stored.bytes !== resource.byteLength) throw fault('asset_receipt_mismatch'); + ledger.result(request.id, 'applied'); + const resources = new Map(previous.resources); resources.set(resource.id, freeze({ ...resource, stored })); + const result = editor.updateRuntime(() => Object.freeze({ ...previous, receipts: readMap(receipts), resources: readMap(resources), ledger: ledger.getSnapshot() }), root.rootRevision); + if (result.status === 'rejected') throw fault(result.code, result.targets); + return; + } catch (error) { + ledger.result(request.id, 'applied', (error as Error).message, false); + const result = editor.updateRuntime(() => Object.freeze({ ...previous, receipts: readMap(receipts), ledger: ledger.getSnapshot() }), root.rootRevision); + if (result.status === 'rejected') throw fault(result.code, result.targets); + report(error); return { blocked: true }; + } + } const receipts = new Map(previous.receipts), conflicts = new Map(previous.conflicts); if (status.canonical !== undefined) receipts.set(request.id, freeze(status.canonical)); try { @@ -90,7 +128,7 @@ export async function openProjectEngine(options: EngineOptions) { 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 runtime: RuntimeState = Object.freeze({ ...previous, 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) { @@ -105,25 +143,93 @@ export async function openProjectEngine(options: EngineOptions) { 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, + compile(intent, id) { + if (intent.operationType === 'document.change') return options.persistence.compile(intent, id, editor.getSnapshot()); + const resource = editor.getSnapshot().runtime.resources.get(intent.resourceId); + if (!options.assets || !resource?.digest) throw fault('asset_not_ready'); + return { kind: 'asset.upload', protocolVersion: 2, canvasIds: [], expectedVersions: {}, payload: { protocol_version: 2, request_id: id, resource_id: resource.id, operation_id: resource.operationId, sha256: resource.digest, byte_length: resource.byteLength, name: resource.name, mime_type: resource.mimeType } }; + }, + async submit(request, signal) { + if (request.kind !== 'asset.upload') return options.persistence.submit(request, signal); + if (!options.assets) return { outcome: 'unresolved', blocked: true, error: 'asset_interpreter_unavailable' }; + const payload = JSON.parse(request.body), bytes = await options.storage.readFile(leader, payload.resource_id); + if (!bytes || bytes.size !== payload.byte_length || await digest(bytes) !== payload.sha256) return { outcome: 'unresolved', blocked: true, error: 'resource_bytes_unavailable' }; + return options.assets.upload(request, bytes, signal); + }, + query: (request, signal) => request.kind === 'asset.upload' ? options.assets?.query(request, signal) ?? Promise.resolve({ outcome: 'unresolved', blocked: true, error: 'asset_interpreter_unavailable' }) : options.persistence.query(request, signal), + supportsRequest: request => request.kind === 'asset.upload' ? !!options.assets && request.protocolVersion === 2 : options.persistence.supportsRequest?.(request) ?? request.protocolVersion === 1, 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); }); + const operationIds = [...commit.operationIds, ...staged.map(resource => resource.operationId)]; + queueMicrotask(() => { if (!closing && !closed) { scheduler.notify(operationIds); if (operationIds.length) prepareDigests(); } }); }); - scheduler.start(); + async function digest(bytes: Blob) { return [...new Uint8Array(await crypto.subtle.digest('SHA-256', await bytes.arrayBuffer()))].map(value => value.toString(16).padStart(2, '0')).join(''); } + function prepareDigests() { + if (closing || closed) return; + for (const resource of editor.getSnapshot().runtime.resources.values()) { + if (digestTasks.size >= 2) break; + if (resource.digest || digestTasks.has(resource.id)) continue; + const operation = editor.getSnapshot().runtime.ledger.operations.get(resource.operationId); + if (operation?.phase !== 'blocked' || !operation.blockers.includes('waiting_digest')) continue; + const task = Promise.resolve().then(async () => { + const bytes = fileWrites.get(resource.id) ?? await options.storage.readFile(leader, resource.id); + if (!bytes || bytes.size !== resource.byteLength) throw fault('resource_bytes_unavailable', [resource.id]); + const hash = await digest(bytes); + if (closing || closed) return; + const result = editor.updateRuntime(previous => { + if (previous.resources.get(resource.id) !== resource) throw fault('resource_identity_conflict', [resource.id]); + const resources = new Map(previous.resources), ledger = createLedger(previous.ledger); + resources.set(resource.id, freeze({ ...resource, digest: hash })); ledger.block(resource.operationId, []); + return Object.freeze({ ...previous, resources: readMap(resources), ledger: ledger.getSnapshot() }); + }); + if (result.status === 'rejected') throw fault(result.code, result.targets); + scheduler.notify([resource.operationId]); + }).catch(error => { if (!closing && !closed) { mutate(ledger => ledger.block(resource.operationId, ['resource_bytes_unavailable'])); report(error); } }).finally(() => { digestTasks.delete(resource.id); prepareDigests(); }); + digestTasks.set(resource.id, task); + } + } + scheduler.start(); prepareDigests(); return Object.freeze({ - editor, sessions, getSnapshot: editor.getSnapshot, persist, + editor, sessions, + stageFiles(files: readonly StagedFile[], callback: (resources: readonly ResourceHandle[], transaction: Transaction) => unknown) { + if (closing || closed) throw fault('engine_closing'); + if (staged.length) throw fault('reentrant_dispatch'); + if (!options.assets) throw fault('asset_interpreter_unavailable'); + const resources = files.map(file => { + if (!(file.bytes instanceof Blob) || !file.bytes.size || file.bytes.size > maxResourceBytes || !file.bytes.type.startsWith('image/') || typeof file.name !== 'string' || !file.name) throw fault('invalid_staged_file'); + return freeze({ id: createId(), operationId: createId(), name: file.name, mimeType: file.bytes.type, byteLength: file.bytes.size }); + }); + if (new Set(resources.flatMap(resource => [resource.id, resource.operationId])).size !== resources.length * 2 || resources.some(resource => editor.getSnapshot().runtime.resources.has(resource.id) || fileWrites.has(resource.id))) throw fault('resource_identity_conflict'); + staged = resources; resources.forEach((resource, index) => fileWrites.set(resource.id, files[index].bytes)); + try { + const handles = freeze(resources.map(resource => ({ id: resource.id, src: options.assets!.sourceURL(resource.id) }))); + const result = editor.transaction(tx => callback(handles, tx), { label: 'Add images' }); + if (result.status === 'rejected' || !result.operationIds.length) for (const resource of resources) fileWrites.delete(resource.id); + return result; + } finally { staged = []; for (const resource of resources) if (!editor.getSnapshot().runtime.resources.has(resource.id)) fileWrites.delete(resource.id); } + }, + async displayResource(id: Id) { + if (closing || closed) throw fault('engine_closing'); + const bytes = fileWrites.get(id) ?? await options.storage.readFile(leader, id); + if (closing || closed) throw fault('engine_closing'); + if (!bytes && !options.assets) throw fault('resource_missing', [id]); + const url = bytes ? URL.createObjectURL(bytes) : options.assets!.displayURL(id); + let released = false; + const release = () => { if (!released) { released = true; displays.delete(release); if (bytes) URL.revokeObjectURL(url); } }; + displays.add(release); return Object.freeze({ url, release }); + }, + 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 settle() { await Promise.resolve(); while (digestTasks.size) await Promise.all([...digestTasks.values()]); await scheduler.settle(); await persist(); }, async close() { if (closed) return; closing = true; sessions.dispose(); scheduler.stop(); + for (const release of [...displays]) release(); // Failure keeps the leader/recovery material available for an explicit retry of close. await persist(); unsubscribe(); editor.dispose(); await leader.release(); closed = true; }, diff --git a/src/lib/canvas-runtime/ledger.ts b/src/lib/canvas-runtime/ledger.ts index 7eb6d37..15b0d7b 100644 --- a/src/lib/canvas-runtime/ledger.ts +++ b/src/lib/canvas-runtime/ledger.ts @@ -5,6 +5,7 @@ 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 kind?: 'document.change' | 'asset.upload'; 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; @@ -72,13 +73,14 @@ export function createLedger(initial?: LedgerSnapshot) { 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; protocolVersion?: number; operationIds: readonly Id[]; canvasIds: readonly Id[]; expectedVersions: Readonly>; payload: Json }) { - if (!input.id || requests.has(input.id) || !Number.isSafeInteger(input.protocolVersion ?? 1) || (input.protocolVersion ?? 1) < 1 || !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] < 0)) throw fault('invalid_frozen_request'); + freeze(input: { id: Id; kind?: FrozenRequest['kind']; protocolVersion?: number; operationIds: readonly Id[]; canvasIds: readonly Id[]; expectedVersions: Readonly>; payload: Json }) { + if (!input.id || requests.has(input.id) || !Number.isSafeInteger(input.protocolVersion ?? 1) || (input.protocolVersion ?? 1) < 1 || !input.operationIds.length || (!input.canvasIds.length && input.kind!=='asset.upload') || !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] < 0)) throw fault('invalid_frozen_request'); for (const id of input.operationIds) { const operation = operations.get(id); + if ((input.kind ?? 'document.change') !== operation?.intent.operationType) throw fault('request_kind_mismatch',[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: input.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 }); + const request: FrozenRequest = freeze({ id: input.id, kind: input.kind ?? 'document.change', protocolVersion: input.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; diff --git a/src/lib/canvas-runtime/reconcile.ts b/src/lib/canvas-runtime/reconcile.ts index c75bf34..e0cc96d 100644 --- a/src/lib/canvas-runtime/reconcile.ts +++ b/src/lib/canvas-runtime/reconcile.ts @@ -5,6 +5,7 @@ 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.operationType==='asset.upload')return[]; if (intent.canvasIds?.length) return intent.canvasIds; const ids = new Set(); for (const change of intent.changes) { @@ -29,7 +30,7 @@ export function operationCanvases(intent: OperationIntent, document: Document): 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; + if (record.phase === 'resolved' || record.intent.operationType!=='document.change') continue; for (const change of record.intent.changes) { if (satisfied(document, change)) continue; document = applyChanges(document, [change]); diff --git a/src/lib/canvas-runtime/recovery.ts b/src/lib/canvas-runtime/recovery.ts index cba9193..e890813 100644 --- a/src/lib/canvas-runtime/recovery.ts +++ b/src/lib/canvas-runtime/recovery.ts @@ -3,6 +3,7 @@ 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 { ResourceRecord } from './resources.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] }); @@ -12,7 +13,7 @@ export function encodeRecovery(snapshot: EngineSnapshot): RecoveryImage { 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] } }, + runtime: { resources: [...snapshot.runtime.resources], resourcePins: [...snapshot.runtime.resourcePins], 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 }); @@ -53,7 +54,9 @@ function changes(value: unknown): readonly 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); + const kind = string(raw.operationType); + if (kind === 'asset.upload') { string(raw.resourceId); if (array(raw.changes).length || array(raw.canvasIds).length) throw fault('corrupt_recovery_image'); } + 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); } @@ -73,10 +76,10 @@ export function decodeRecovery(image: RecoveryImage, registry: Registry): CoreSn 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'); + if ((object(raw.intent).operationType !== 'asset.upload' && !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'); + if (operations.size !== [...records.values()].filter(record => record.intent.operationType !== 'asset.upload').length) 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) { @@ -92,8 +95,9 @@ export function decodeRecovery(image: RecoveryImage, registry: Registry): CoreSn 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 (raw.kind === 'asset.upload' && (array(raw.canvasIds).length || Object.keys(object(raw.expectedVersions)).length || array(raw.operationIds).length !== 1)) throw fault('corrupt_recovery_image'); 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'); + for (const operationId of strings(raw.operationIds)) if (records.get(operationId)?.requestId !== id || (raw.kind ?? 'document.change') !== records.get(operationId)?.intent.operationType) 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); @@ -105,8 +109,20 @@ export function decodeRecovery(image: RecoveryImage, registry: Registry): CoreSn 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 resources = table(runtime.resources ?? [], (value, id) => { + const raw = object(value); + if (string(raw.id) !== id || string(raw.operationId) === id) throw fault('corrupt_recovery_image'); + string(raw.name); string(raw.mimeType); integer(raw.byteLength, 1); + if (raw.digest !== undefined && !/^[a-f0-9]{64}$/.test(string(raw.digest))) throw fault('corrupt_recovery_image'); + const operation = records.get(raw.operationId as string)?.intent; + if (operation?.operationType !== 'asset.upload' || operation.resourceId !== id) throw fault('corrupt_recovery_image'); + if (raw.stored !== undefined) { const stored = object(raw.stored); if (string(stored.id) !== id || integer(stored.bytes, 1) !== raw.byteLength) throw fault('corrupt_recovery_image'); string(stored.url); string(stored.mime_type); integer(stored.width, 1); integer(stored.height, 1); } + return freeze(value as ResourceRecord); + }); + for (const record of records.values()) if (record.intent.operationType === 'asset.upload' && resources.get(record.intent.resourceId)?.operationId !== record.intent.id) throw fault('corrupt_recovery_image'); + const resourcePins = table(runtime.resourcePins ?? [], strings); 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) }), + runtime: Object.freeze({ resources, resourcePins, 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/resources.ts b/src/lib/canvas-runtime/resources.ts new file mode 100644 index 0000000..99e73f1 --- /dev/null +++ b/src/lib/canvas-runtime/resources.ts @@ -0,0 +1,93 @@ +import type { Change, CoreSnapshot, Document, ElementRecord, Id, OperationIntent, Registry } from '../canvas-core/types.ts'; +import { ancestors } from '../canvas-core/document.ts'; +import type { FrozenRequest, LedgerSnapshot } from './ledger.ts'; +import type { OperationStatus } from './scheduler.ts'; +import { readMap, readSet } from '../canvas-core/values.ts'; + +export type StoredAsset = { readonly id: Id; readonly url: string; readonly width: number; readonly height: number; readonly bytes: number; readonly mime_type: string }; +export type ResourceRecord = { readonly id: Id; readonly operationId: Id; readonly name: string; readonly mimeType: string; readonly byteLength: number; readonly digest?: string; readonly stored?: StoredAsset }; +export type DisplayHandle = { readonly url: string; release(): void }; +export type AssetPort = { + sourceURL(resourceId: Id): string; + displayURL(resourceId: Id): string; + upload(request: FrozenRequest, bytes: Blob, signal: AbortSignal): Promise; + query(request: FrozenRequest, signal: AbortSignal): Promise; + accept(request: FrozenRequest, status: OperationStatus, resource: ResourceRecord): StoredAsset; +}; +export type StagedFile = { readonly bytes: Blob; readonly name: string }; +export type ResourceHandle = { readonly id: Id; readonly src: string }; +export const maxResourceBytes = 20 * 1024 * 1024; + +export function referencedResources(changes: readonly Change[], registry: Registry): Set { + const result = new Set(), fields = new Set([...registry.components.values()].flatMap(component => component.resourceFields ?? [])); + const collect = (element: ElementRecord | null) => { + if (!element) return; + for (const field of registry.components.get(element.componentId)?.resourceFields ?? []) if (typeof element.data[field] === 'string') result.add(element.data[field] as string); + }; + for (const change of changes) { + if (change.kind === 'element') { collect(change.before); collect(change.after); } + else if (change.kind === 'field' && change.field.startsWith('data.') && fields.has(change.field.slice(5))) { + if (typeof change.before === 'string') result.add(change.before); + if (typeof change.after === 'string') result.add(change.after); + } + } + return result; +} + +/** Read/write dependencies use object identities, rather than blocking every edit behind a pending image. */ +export function operationFootprint(intent: OperationIntent, document: Document) { + const reads = new Set(), writes = new Set(); + if (intent.operationType === 'asset.upload') { writes.add(`resource:${intent.resourceId}`); return { reads, writes }; } + const readCanvas = (id: Id) => { + if (document.canvases.has(id)) for (const ancestor of ancestors(document, id)) reads.add(`canvas:${ancestor}`); + else reads.add(`canvas:${id}`); + }; + const record = (element: ElementRecord | null) => { + if (!element) return; + readCanvas(element.ownerCanvasId); + if (element.structure === 'connection') { reads.add(`element:${element.source.elementId}`); reads.add(`element:${element.target.elementId}`); } + }; + for (const change of intent.changes) { + if (change.kind === 'canvas') { + writes.add(`canvas:${change.id}`); + for (const canvas of [change.before, change.after]) if (canvas?.parent) { readCanvas(canvas.parent.canvasId); reads.add(`element:${canvas.parent.entryElementId}`); } + } else if (change.kind === 'element') { + writes.add(`element:${change.id}`); record(change.before); record(change.after); + } else { + writes.add(`element:${change.target.id}`); record(document.elements.get(change.target.id) ?? null); + if (change.field === 'ownerCanvasId') { readCanvas(change.before as string); readCanvas(change.after as string); } + if (change.field === 'source' || change.field === 'target') for (const endpoint of [change.before, change.after]) { + if (endpoint && typeof endpoint === 'object' && !Array.isArray(endpoint) && typeof endpoint.elementId === 'string') reads.add(`element:${endpoint.elementId}`); + } + } + } + return { reads, writes }; +} +export function conflictsWith(prior: ReturnType, next: ReturnType) { + return [...prior.writes].some(id => next.writes.has(id) || next.reads.has(id)) || [...prior.reads].some(id => next.writes.has(id)); +} + +/** Pins are derived from their owners. There is no second mutable reference count. */ +export function resourcePins(snapshot: CoreSnapshot, ledger: LedgerSnapshot, registry: Registry, explicit: ReadonlyMap) { + const pins = new Map>(); + let complete = true; + const pin = (id: Id, owner: string) => { const set = pins.get(id) ?? new Set(); set.add(owner); pins.set(id, set); }; + const element = (record: ElementRecord | null, owner: string) => { + if (!record) return; + const component = registry.components.get(record.componentId); + if (registry.resolve(record) !== 'known' || component?.resourceFields === undefined) { complete = false; return; } + for (const field of component.resourceFields) { const ref = record.data[field]; if (typeof ref === 'string') pin(ref, owner); } + }; + const changes = (values: readonly Change[], owner: string) => { + for (const change of values) if (change.kind === 'element') { element(change.before, owner); element(change.after, owner); } + for (const id of referencedResources(values, registry)) pin(id, owner); + }; + for (const record of snapshot.document.elements.values()) element(record, `element:${record.id}:${record.incarnation}`); + for (const entry of snapshot.history.entries.values()) changes(entry.changes, `history:${entry.id}`); + for (const record of ledger.operations.values()) if (record.phase !== 'resolved') { + if (record.intent.operationType === 'asset.upload') pin(record.intent.resourceId, `operation:${record.intent.id}`); + else changes(record.intent.changes, `operation:${record.intent.id}`); + } + for (const [owner, ids] of explicit) for (const id of ids) pin(id, owner); + return Object.freeze({ complete, pins: readMap(new Map([...pins].map(([id, owners]) => [id, readSet(owners)]))) }); +} diff --git a/src/lib/canvas-runtime/scheduler.ts b/src/lib/canvas-runtime/scheduler.ts index 87f1d71..4d99d21 100644 --- a/src/lib/canvas-runtime/scheduler.ts +++ b/src/lib/canvas-runtime/scheduler.ts @@ -4,7 +4,7 @@ import { createLedger, type FrozenRequest, type LedgerSnapshot, type Outcome } f import { createDeque, createMinHeap } from './structures.ts'; export type OperationStatus = { readonly outcome: Outcome; readonly error?: string; readonly retryAfterMs?: number; readonly canonical?: Json; readonly blocked?: boolean }; -export type CompiledRequest = { readonly protocolVersion?: number; readonly payload: Json; readonly canvasIds: readonly Id[]; readonly expectedVersions: Readonly> }; +export type CompiledRequest = { readonly kind?: FrozenRequest['kind']; readonly protocolVersion?: number; 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; @@ -33,11 +33,13 @@ export function createScheduler(options: SchedulerOptions) { 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 tokens = new Map(), active = new Map; kind: FrozenRequest['kind'] }>(); + const laneWaiters = new Map, Set>(), deferredRetries = new Map(); + let timer: unknown, pumpTimer: 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; + const canRun = (kind: FrozenRequest['kind']) => active.size < concurrency && (kind !== 'asset.upload' || ![...active.values()].some(task => task.kind === 'asset.upload')); function enqueue(id: Id) { if (queued.has(id)) return; @@ -82,7 +84,7 @@ export function createScheduler(options: SchedulerOptions) { function armTimer() { if (timer !== undefined) { clock.clearTimer(timer); timer = undefined; } while (retry.peek() && !validRetry(retry.peek()!)) retry.pop(); - if (stopped || !retry.peek()) return; + if (stopped || active.size >= concurrency || !retry.peek()) return; timer = clock.setTimer(() => { timer = undefined; pump(); }, Math.max(0, retry.peek()!.dueAt - clock.now())); } function launch(id: Id, action: 'submit' | 'query') { @@ -112,8 +114,16 @@ export function createScheduler(options: SchedulerOptions) { 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 }); + }).finally(() => { + active.delete(id); + if (!stopped) { + for (const item of deferredRetries.values()) retry.push(item); + deferredRetries.clear(); + for (const [kind, waiting] of laneWaiters) if (canRun(kind)) { for (const operationId of waiting) enqueue(operationId); laneWaiters.delete(kind); } + pump(); + } + }); + active.set(id, { controller, promise, kind: options.getLedger().requests.get(id)?.kind }); } function pump() { if (stopped || pumping) return; @@ -123,17 +133,23 @@ export function createScheduler(options: SchedulerOptions) { 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'); + if (request && !active.has(item.id)) { + if (!canRun(request.kind)) { deferredRetries.set(item.id, item); continue; } + launch(item.id, request.retryAction ?? 'query'); + } } - let budget = ready.size; + let budget = Math.min(ready.size, 64); 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 { + if (!canRun(record.intent.operationType)) { + const set = laneWaiters.get(record.intent.operationType) ?? new Set(); set.add(id); laneWaiters.set(record.intent.operationType, set); continue; + } const canvases = [...new Set(options.participants(record.intent))].sort(); - if (!canvases.length) throw fault('missing_participants'); + if (!canvases.length && record.intent.operationType !== 'asset.upload') 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); } @@ -145,7 +161,10 @@ export function createScheduler(options: SchedulerOptions) { 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(); } + } finally { + pumping = false; armTimer(); + if (!stopped && ready.size && active.size < concurrency && pumpTimer === undefined) pumpTimer = clock.setTimer(() => { pumpTimer = undefined; pump(); }, 0); + } } return Object.freeze({ start() { @@ -179,6 +198,6 @@ export function createScheduler(options: SchedulerOptions) { }, 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(); }, + stop() { closed = true; stopped = true; if (timer !== undefined) clock.clearTimer(timer); if (pumpTimer !== undefined) clock.clearTimer(pumpTimer); timer = pumpTimer = undefined; for (const item of active.values()) item.controller.abort(); }, }); } diff --git a/src/lib/canvas-service/assets.ts b/src/lib/canvas-service/assets.ts new file mode 100644 index 0000000..4d78591 --- /dev/null +++ b/src/lib/canvas-service/assets.ts @@ -0,0 +1,62 @@ +import type { Id, Json } from '../canvas-core/types.ts'; +import { fault, freeze, validJson } from '../canvas-core/values.ts'; +import type { AssetPort, ResourceRecord, StoredAsset } from '../canvas-runtime/resources.ts'; +import type { FrozenRequest } from '../canvas-runtime/ledger.ts'; +import type { OperationStatus } from '../canvas-runtime/scheduler.ts'; +import type { ServiceResponse, ServiceTransport } from './foundry.ts'; + +const uuid = /^[a-f\d]{8}-[a-f\d]{4}-[a-f\d]{4}-[a-f\d]{4}-[a-f\d]{12}$/i; +function object(value: unknown): Record { + if (!value || Array.isArray(value) || typeof value !== 'object') throw fault('invalid_asset_response'); + return value as Record; +} +function payload(request: FrozenRequest) { + const raw = object(JSON.parse(request.body)); + if (request.kind !== 'asset.upload' || request.protocolVersion !== 2 || raw.protocol_version !== 2 || raw.request_id !== request.id || !uuid.test(request.id) || typeof raw.resource_id !== 'string' || !uuid.test(raw.resource_id) || request.operationIds.length !== 1 || raw.operation_id !== request.operationIds[0] || request.canvasIds.length || Object.keys(request.expectedVersions).length || typeof raw.sha256 !== 'string' || !/^[a-f0-9]{64}$/.test(raw.sha256) || typeof raw.byte_length !== 'number' || !Number.isSafeInteger(raw.byte_length) || raw.byte_length < 1 || raw.byte_length > 20 * 1024 * 1024 || typeof raw.name !== 'string' || !raw.name || typeof raw.mime_type !== 'string') throw fault('invalid_asset_request'); + return raw; +} +function status(request: FrozenRequest, response: ServiceResponse, query: boolean): OperationStatus { + const body = object(response.body), matching = body.protocol_version === 2 && body.request_id === request.id; + if (matching && body.status === 'applied' && response.status >= 200 && response.status < 300) { + if (!validJson(body)) throw fault('invalid_asset_response'); + return freeze({ outcome: 'applied', canonical: JSON.parse(JSON.stringify(body)) as Json }); + } + if (query && matching && body.status === 'unknown' && response.status === 404) return { outcome: 'unresolved', error: 'asset_receipt_unknown', retryAfterMs: response.retryAfterMs }; + const error = typeof body.error === 'string' ? body.error : 'asset_transport_unknown'; + // A bare rejection does not prove whether an earlier upload committed. + return { outcome: 'unresolved', error, retryAfterMs: response.retryAfterMs, ...(response.status >= 400 && response.status < 500 ? { blocked: true } : {}) }; +} + +/** Immutable bytes and a preallocated resource ID survive response loss and view changes. */ +export function createFoundryAssets(options: { workspaceId: Id; transport: ServiceTransport; project?: string; origin?: string }): AssetPort { + if (!uuid.test(options.workspaceId)) throw fault('invalid_service_scope'); + const origin = new URL(options.origin ?? 'https://api.foundry.entroia.com'); + if (origin.protocol !== 'https:' || origin.username || origin.password || origin.search || origin.hash || origin.pathname !== '/') throw fault('invalid_asset_origin'); + const path = `workspaces/${encodeURIComponent(options.workspaceId)}/assets`; + const sourceURL = (id: Id) => `${origin.origin}/projects/${encodeURIComponent(options.project ?? 'endoleap')}/${path}/${encodeURIComponent(id)}`; + const accept = (request: FrozenRequest, result: OperationStatus, resource: ResourceRecord): StoredAsset => { + const raw = payload(request), receipt = object(result.canonical), asset = object(receipt.asset); + if (result.outcome !== 'applied' || receipt.protocol_version !== 2 || receipt.request_id !== request.id || receipt.status !== 'applied' || resource.id !== raw.resource_id || resource.operationId !== raw.operation_id || resource.digest !== raw.sha256 || resource.byteLength !== raw.byte_length || asset.id !== resource.id || asset.url !== sourceURL(resource.id) || asset.bytes !== resource.byteLength || typeof asset.width !== 'number' || typeof asset.height !== 'number' || !Number.isSafeInteger(asset.width) || !Number.isSafeInteger(asset.height) || asset.width <= 0 || asset.height <= 0 || asset.width * asset.height > 40_000_000 || !['image/png', 'image/jpeg', 'image/gif', 'image/webp'].includes(asset.mime_type as string)) throw fault('asset_receipt_mismatch'); + return freeze(JSON.parse(JSON.stringify(asset)) as StoredAsset); + }; + async function execute(request: FrozenRequest, bytes: Blob | undefined, signal: AbortSignal) { + try { + const raw = payload(request); + let body: FormData | undefined; + if (bytes) { + const digest = [...new Uint8Array(await crypto.subtle.digest('SHA-256', await bytes.arrayBuffer()))].map(value => value.toString(16).padStart(2, '0')).join(''); + if (bytes.size !== raw.byte_length || digest !== raw.sha256) throw fault('resource_bytes_mismatch'); + body = new FormData(); body.append('image', bytes, raw.name as string); + } + const response = await options.transport({ method: bytes ? 'POST' : 'GET', path: bytes ? path : `${path}/operations/${encodeURIComponent(request.id)}`, requestId: bytes ? request.id : undefined, resourceId: bytes ? raw.resource_id as string : undefined, body, signal }); + return status(request, response, !bytes); + } catch (error) { + const code = (error as { code?: string }).code; + return { outcome: 'unresolved' as const, error: code ?? 'asset_transport_unknown', ...(code && code !== 'invalid_asset_response' ? { blocked: true } : {}) }; + } + } + return Object.freeze({ sourceURL, displayURL: (id: Id) => `/api/foundry/${path}/${encodeURIComponent(id)}?variant=preview`, accept, + upload: (request: FrozenRequest, bytes: Blob, signal: AbortSignal) => execute(request, bytes, signal), + query: (request: FrozenRequest, signal: AbortSignal) => execute(request, undefined, signal), + }); +} diff --git a/src/lib/canvas-service/foundry.ts b/src/lib/canvas-service/foundry.ts index 64e24f2..78e4203 100644 --- a/src/lib/canvas-service/foundry.ts +++ b/src/lib/canvas-service/foundry.ts @@ -5,7 +5,7 @@ import type { EngineSnapshot, PersistencePort } from '../canvas-runtime/contract import type { FrozenRequest } from '../canvas-runtime/ledger.ts'; import type { OperationStatus } from '../canvas-runtime/scheduler.ts'; -export type ServiceRequest = { readonly path: string; readonly method: 'GET' | 'POST'; readonly requestId?: Id; readonly body?: string; readonly signal?: AbortSignal }; +export type ServiceRequest = { readonly path: string; readonly method: 'GET' | 'POST'; readonly requestId?: Id; readonly body?: string | FormData; readonly resourceId?: Id; readonly signal?: AbortSignal }; export type ServiceResponse = { readonly status: number; readonly body: unknown; readonly retryAfterMs?: number }; export type ServiceTransport = (request: ServiceRequest) => Promise; const uuid = /^[a-f\d]{8}-[a-f\d]{4}-[a-f\d]{4}-[a-f\d]{4}-[a-f\d]{12}$/i; @@ -26,7 +26,7 @@ export function createFoundryTransport(options: { accountId: Id; isCurrentIdenti return async input => { if (!options.isCurrentIdentity()) throw fault('identity_changed'); const response = await requestFetch(`${options.basePath ?? '/api/foundry/'}${input.path}`, { method: input.method, body: input.body, signal: input.signal, credentials: 'same-origin', cache: 'no-store', redirect: 'error', headers: { - Accept: 'application/json', 'Content-Type': 'application/json', 'X-Expected-Account': options.accountId, ...(input.requestId ? { 'Idempotency-Key': input.requestId } : {}), + Accept: 'application/json', ...(input.body instanceof FormData ? {} : { 'Content-Type': 'application/json' }), ...(input.resourceId ? { 'X-Resource-Id': input.resourceId } : {}), 'X-Expected-Account': options.accountId, ...(input.requestId ? { 'Idempotency-Key': input.requestId } : {}), } }); if (!options.isCurrentIdentity()) { await response.body?.cancel(); throw fault('identity_changed'); } let body: unknown; diff --git a/src/lib/canvas-service/index.ts b/src/lib/canvas-service/index.ts index f7e3c6c..762387b 100644 --- a/src/lib/canvas-service/index.ts +++ b/src/lib/canvas-service/index.ts @@ -1 +1,2 @@ export * from './foundry.ts'; +export * from './assets.ts'; diff --git a/tests/headless/asset-service.test.mjs b/tests/headless/asset-service.test.mjs new file mode 100644 index 0000000..c8f4fe8 --- /dev/null +++ b/tests/headless/asset-service.test.mjs @@ -0,0 +1,50 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { createFoundryAssets, createFoundryTransport } from '../../src/lib/canvas-service/index.ts'; +const id = n => `10000000-0000-4000-8000-${String(n).padStart(12, '0')}`; +async function fixture() { + const blob = new Blob(['immutable image'], { type: 'image/png' }); + const digest = [...new Uint8Array(await crypto.subtle.digest('SHA-256', await blob.arrayBuffer()))].map(value => value.toString(16).padStart(2, '0')).join(''); + const resource = { id: id(1), operationId: id(2), name: 'picture.png', byteLength: blob.size, mimeType: blob.type, digest }; + const request = { kind: 'asset.upload', protocolVersion: 2, id: id(3), operationIds: [resource.operationId], canvasIds: [], expectedVersions: {}, phase: 'frozen', outcome: 'unresolved', generation: 1, attempt: 0, + body: JSON.stringify({ protocol_version: 2, request_id: id(3), resource_id: resource.id, operation_id: resource.operationId, name: resource.name, mime_type: resource.mimeType, byte_length: resource.byteLength, sha256: resource.digest }) }; + const calls = [], receipt = { protocol_version: 2, request_id: request.id, status: 'applied', asset: { id: resource.id, bytes: resource.byteLength, mime_type: 'image/png', width: 2, height: 1, url: `https://api.foundry.entroia.com/projects/endoleap/workspaces/${id(4)}/assets/${resource.id}` } }; + let response = { status: 201, body: receipt }; + const assets = createFoundryAssets({ workspaceId: id(4), transport: async input => { calls.push(input); return response; } }); + return { blob, resource, request, calls, receipt, assets, respond: next => { response = next; } }; +} + +test('asset adapter preserves byte identity and headers and validates the scoped applied receipt', async () => { + const f = await fixture(), result = await f.assets.upload(f.request, f.blob, new AbortController().signal); + assert.equal(result.outcome, 'applied'); assert.equal(f.calls[0].requestId, f.request.id); assert.equal(f.calls[0].resourceId, f.resource.id); + assert.equal(f.calls[0].body.get('image').name, f.resource.name); assert.equal(await f.calls[0].body.get('image').text(), await f.blob.text()); + assert.equal(f.assets.accept(f.request, result, f.resource).id, f.resource.id); + for (const patch of [{ id: id(99) }, { bytes: 100 }, { url: 'https://other.example/image' }, { width: 0 }, { width: 1000000, height: 1000000 }]) { + const altered = { ...result, canonical: { ...f.receipt, asset: { ...f.receipt.asset, ...patch } } }; + assert.throws(() => f.assets.accept(f.request, altered, f.resource), /asset_receipt_mismatch/); + } + const changed = await f.assets.upload(f.request, new Blob(['different']), new AbortController().signal); + assert.equal(changed.outcome, 'unresolved'); assert.equal(changed.blocked, true); assert.equal(f.calls.length, 1); +}); + +test('missing and rejected upload receipts never become proof of zero writes', async () => { + const f = await fixture(), signal = new AbortController().signal; + f.respond({ status: 404, body: { protocol_version: 2, request_id: f.request.id, status: 'unknown' } }); + const unknown = await f.assets.query(f.request, signal); + assert.equal(unknown.outcome, 'unresolved'); assert.equal(unknown.blocked, undefined); + assert.ok(f.calls[0].path.endsWith(`/assets/operations/${f.request.id}`)); + f.respond({ status: 403, body: { error: 'forbidden' } }); + assert.equal((await f.assets.query(f.request, signal)).blocked, true); + f.respond({ status: 201, body: { ...f.receipt, request_id: id(99) } }); + assert.equal((await f.assets.upload(f.request, f.blob, signal)).outcome, 'unresolved'); +}); + +test('multipart browser transport leaves the boundary to fetch and binds the expected account', async () => { + let captured; + const transport = createFoundryTransport({ accountId: id(5), isCurrentIdentity: () => true, fetch: async (url, init) => { captured = { url, init }; return Response.json({ status: 'applied' }); } }); + const body = new FormData(); body.append('image', new Blob(['image']), 'a.png'); + await transport({ method: 'POST', path: 'workspaces/example/assets', body, requestId: id(3), resourceId: id(1) }); + const headers = new Headers(captured.init.headers); + assert.equal(headers.has('content-type'), false); assert.equal(headers.get('X-Expected-Account'), id(5)); + assert.equal(headers.get('X-Resource-Id'), id(1)); assert.equal(headers.get('Idempotency-Key'), id(3)); assert.equal(captured.init.body, body); +}); diff --git a/tests/headless/browser-storage-cases.mjs b/tests/headless/browser-storage-cases.mjs index 5f01e6a..60d798f 100644 --- a/tests/headless/browser-storage-cases.mjs +++ b/tests/headless/browser-storage-cases.mjs @@ -1,8 +1,54 @@ 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'; +import { createFoundryAssets } from '../../src/lib/canvas-service/assets.ts'; import { createFoundryPersistence } from '../../src/lib/canvas-service/foundry.ts'; +async function resourceRecoveryCase(name, assert) { + const scope = { accountId: crypto.randomUUID(), workspaceId: crypto.randomUUID(), projectId: crypto.randomUUID() }; + const registry = createRegistry(builtinComponents()), document = createDocument(scope.projectId, [{ id: scope.projectId, parent: null }], []); + const storage = createBrowserStorage({ name: `${name}-resources` }), receipts = new Map(), uploads = [], queries = [], writes = []; + const assets = createFoundryAssets({ workspaceId: scope.workspaceId, transport: async input => { + if (input.method === 'POST') { + uploads.push(input); const file = input.body.get('image'); + assert(await file.text() === 'durable-image-bytes', 'upload reconstructed different file bytes'); + receipts.set(input.requestId, { protocol_version: 2, request_id: input.requestId, status: 'applied', asset: { id: input.resourceId, bytes: file.size, mime_type: 'image/png', width: 2, height: 1, url: assets.sourceURL(input.resourceId) } }); + return { status: 502, body: { error: 'lost_upload_ack' } }; + } + queries.push(input); return { status: 200, body: receipts.get(input.path.split('/').at(-1)) }; + } }); + let server = document; + const persistence = { + compile: (intent, id, snapshot) => ({ payload: { intent, id }, canvasIds: intent.canvasIds, expectedVersions: { [scope.projectId]: snapshot.runtime.versions.get(scope.projectId) } }), + async submit(request) { writes.push(request); server = applyChanges(server, JSON.parse(request.body).intent.changes); return { outcome: 'applied' }; }, + async query() { return { outcome: 'unresolved' }; }, + canonical: () => ({ document: server, versions: new Map([[scope.projectId, writes.length + 1]]) }), + }; + const options = { scope, registry, document, versions: new Map([[scope.projectId, 1]]), storage, persistence, assets }; + let engine, resource; + try { + engine = await openProjectEngine(options); + const result = engine.stageFiles([{ name: 'picture.png', bytes: new Blob(['durable-image-bytes'], { type: 'image/png' }) }], ([ref], tx) => { + resource = ref; + tx.dispatch({ type: 'element.create', element: { id: 'picture', incarnation: 1, componentId: 'image', dataVersion: 1, structure: 'node', ownerCanvasId: scope.projectId, data: { title: 'Picture', text: '', src: ref.src, assetId: ref.id }, geometry: { x: 10, y: 20, width: 100, height: 80, rotation: 0 } } }); + }); + assert(result.status === 'applied' && engine.getSnapshot().document.elements.has('picture'), 'pending image was hidden until prepare/upload'); + const handle = await engine.displayResource(resource.id); assert(handle.url.startsWith('blob:'), 'pending image did not expose a local preview'); handle.release(); + await engine.settle(); assert(uploads.length === 1 && writes.length === 0, 'document was sent before its resource was confirmed'); + await engine.close(); + const leader = await storage.claim(scope), image = await storage.recover(leader); + assert(await (await storage.readFile(leader, resource.id)).text() === 'durable-image-bytes', 'file bytes did not survive actual IndexedDB recovery'); + assert(image.payload.runtime.resources[0][1].id === resource.id && image.payload.document.elements[0].data.assetId === resource.id, 'resource metadata and element belong to different durable cuts'); + await leader.release(); + engine = await openProjectEngine({ ...options, clock: { now: () => Date.now() + 100000, setTimer: (fn, ms) => setTimeout(fn, ms), clearTimer: id => clearTimeout(id) } }); + await new Promise(resolve => setTimeout(resolve, 10)); await engine.settle(); + assert(uploads.length === 1 && queries.length === 1 && writes.length === 1, 'reopen uploaded again or lost the pending element operation'); + assert(queries[0].path.endsWith(uploads[0].requestId), 'recovery queried a replacement request identity'); + assert(engine.getSnapshot().runtime.resources.get(resource.id).stored.id === resource.id, 'confirmed asset changed resource identity'); + assert(engine.getSnapshot().history.undo.length === 1 && engine.getSnapshot().document.elements.size === 1, 'resource ACK recreated the node/history'); + } finally { await engine?.close(); storage.close(); indexedDB.deleteDatabase(`${name}-resources`); } +} + async function protocolRecoveryCase(name,assert){ const root='10000000-0000-4000-8000-000000000001',workspace='10000000-0000-4000-8000-000000000003',account='10000000-0000-4000-8000-000000000004'; const registry=createRegistry(builtinComponents()),document=createDocument(root,[{id:root,parent:null}],[{id:'a',incarnation:1,ownerCanvasId:root,componentId:'text',dataVersion:1,structure:'node',geometry:{x:0,y:0,width:260,height:200,rotation:0},data:{title:'a',text:''}}]); @@ -123,6 +169,8 @@ export async function run() { 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'); + await resourceRecoveryCase(name, assert); + results.push('staged file and image recover from real IndexedDB; lost upload ACK queries stable resource identity'); await protocolRecoveryCase(name,assert); results.push('protocol 2 service adapter recovers exact frozen bytes and reconciles a lost ACK through real IndexedDB'); return results; diff --git a/tests/headless/proxy.integration.mjs b/tests/headless/proxy.integration.mjs index ce79c86..92e356d 100644 --- a/tests/headless/proxy.integration.mjs +++ b/tests/headless/proxy.integration.mjs @@ -42,6 +42,13 @@ try{ assert.equal(response.status,409);assert.equal(await response.text(),responseBytes); const request=received.find(item=>item.path.endsWith('/model/operations')); assert.ok(request);assert.equal(request.body.toString(),body);assert.equal(request.headers['idempotency-key'],key);assert.equal(request.headers['x-expected-account'],account);assert.equal(request.headers.authorization,'Bearer fixture-session'); + const receiptPath=`workspaces/10000000-0000-4000-8000-000000000003/assets/operations/${key}`; + const assetReceipt=await fetch(`http://127.0.0.1:${port}/api/foundry/${receiptPath}`);assert.equal(assetReceipt.status,200);assert.deepEqual(await assetReceipt.json(),{id:'fixture'}); + const multipart=new FormData();multipart.append('image',new Blob(['immutable file'],{type:'image/png'}),'fixture.png'); + const assetPath=receiptPath.split('/operations/')[0],resourceId='10000000-0000-4000-8000-000000000006'; + const uploaded=await fetch(`http://127.0.0.1:${port}/api/foundry/${assetPath}`,{method:'POST',body:multipart,headers:{'X-Resource-Id':resourceId,'Idempotency-Key':key,'X-Expected-Account':account}});await uploaded.text();assert.equal(uploaded.status,200); + const upload=received.find(item=>item.path.endsWith('/assets'));assert.equal(upload.headers['x-resource-id'],resourceId);assert.equal(upload.headers['idempotency-key'],key);assert.match(upload.body.toString(),/immutable file/);assert.match(upload.headers['content-type'],/^multipart\/form-data; boundary=/); + console.log('PASS asset receipt JSON and multipart resource identity cross the actual Next proxy'); const before=received.length,invalid=await fetch(`http://127.0.0.1:${port}/api/foundry/${path}/unexpected/extra`,{signal:AbortSignal.timeout(10000)});await invalid.text();assert.equal(invalid.status,404);assert.equal(received.length,before); console.log('PASS actual Next proxy preserves frozen request bytes/key/account and raw terminal receipt above 1 MiB'); console.log('PASS canonical route allowlist rejects unexpected paths without contacting the upstream'); diff --git a/tests/headless/resources.test.mjs b/tests/headless/resources.test.mjs new file mode 100644 index 0000000..1ab6417 --- /dev/null +++ b/tests/headless/resources.test.mjs @@ -0,0 +1,156 @@ +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 { decodeRecovery } from '../../src/lib/canvas-runtime/recovery.ts'; +import { resourcePins } from '../../src/lib/canvas-runtime/resources.ts'; + +const registry = createRegistry(builtinComponents()); +const scope = { accountId: 'account', workspaceId: 'workspace', projectId: 'project' }; +const geometry = { x: 0, y: 0, width: 100, height: 100, rotation: 0 }; +const node = (id, data = {}) => ({ id, incarnation: 1, structure: 'node', componentId: 'image', dataVersion: 1, ownerCanvasId: 'canvas', geometry, data: { title: id, text: '', ...data } }); +const original = createDocument('project', [{ id: 'canvas', parent: null }], [node('a')]); +const deferred = () => { let resolve; const promise = new Promise(yes => { resolve = yes; }); return { promise, resolve }; }; +async function until(predicate) { for (let i = 0; i < 100; i++) { if (predicate()) return; await new Promise(resolve => setTimeout(resolve, 1)); } assert.fail('condition did not become true'); } +function fixture() { + let image = null, owner, counter = 0, server = original, version = 1; + const files = new Map(), batches = [], uploads = [], writes = [], receipts = new Map(), assetReceipts = new Map(); + const storage = { + async claim(scope) { assert.equal(owner, undefined); const handle = { scope, fence: 1, active: true, async release() { owner = undefined; handle.active = false; } }; owner = handle; return handle; }, + async recover() { return structuredClone(image); }, + async commit(_leader, batch) { for (const file of batch.files ?? []) { assert.equal(files.has(file.id), false); files.set(file.id, file.bytes); } image = structuredClone(batch.image); batches.push({ ...batch, image }); return { durableRootRevision: image.rootRevision }; }, + async readFile(_leader, id) { return files.get(id); }, close() {}, + }; + const assets = { + sourceURL: id => `https://assets.example/${id}`, displayURL: id => `/assets/${id}`, + async upload(request, bytes) { + const payload = JSON.parse(request.body); assert.equal(await files.get(payload.resource_id).text(), await bytes.text()); + assert.ok(image.payload.runtime.ledger.requests.some(([id, row]) => id === request.id && row.phase === 'dispatch-marked')); + uploads.push(request); + const asset = { id: payload.resource_id, url: assets.sourceURL(payload.resource_id), width: 20, height: 10, bytes: bytes.size, mime_type: bytes.type }; + assetReceipts.set(request.id, asset); + return { outcome: 'applied', canonical: asset }; + }, + async query(request) { return assetReceipts.has(request.id) ? { outcome: 'applied', canonical: assetReceipts.get(request.id) } : { outcome: 'unresolved' }; }, + accept: (_request, status) => status.canonical, + }; + const persistence = { + compile(intent, id, snapshot) { return { payload: { intent, id }, canvasIds: intent.canvasIds, expectedVersions: Object.fromEntries(intent.canvasIds.map(id => [id, snapshot.runtime.versions.get(id)])) }; }, + async submit(request) { const { intent } = JSON.parse(request.body); writes.push(request); server = applyChanges(server, intent.changes); receipts.set(request.id, { document: server, versions: new Map([['canvas', ++version]]) }); return { outcome: 'applied' }; }, + async query(request) { return receipts.has(request.id) ? { outcome: 'applied' } : { outcome: 'unresolved' }; }, + canonical(request) { return receipts.get(request.id); }, + }; + return { storage, assets, persistence, files, batches, uploads, writes, assetReceipts, image: () => image, + open: extra => openProjectEngine({ scope, registry, document: original, versions: new Map([['canvas', 1]]), storage, assets, persistence, createId: () => `id-${++counter}`, ...extra }), + }; +} +function addImage(engine, id = 'picture') { + let resource; + const result = engine.stageFiles([{ name: 'image.png', bytes: new Blob(['image bytes'], { type: 'image/png' }) }], ([ref], tx) => { + resource = ref; + tx.dispatch({ type: 'element.create', element: node(id, { assetId: ref.id, src: ref.src }) }); + }); + assert.equal(result.status, 'applied'); return resource; +} + +test('pending images publish immediately, bytes share the durable cut, and independent canvas edits overtake upload', async () => { + const f = fixture(), gate = deferred(), upload = f.assets.upload; + f.assets.upload = async (...args) => { await gate.promise; return upload(...args); }; + const engine = await f.open(), ref = addImage(engine); + assert.equal(engine.getSnapshot().document.elements.get('picture').data.assetId, ref.id); + assert.equal(engine.getSnapshot().history.undo.length, 1); + assert.equal(engine.getSnapshot().runtime.ledger.operations.size, 2); + const sequences = [...engine.getSnapshot().runtime.ledger.operations.values()].map(record => record.intent.sequence); + assert.equal(new Set(sequences).size, 2); + const handle = await engine.displayResource(ref.id); assert.match(handle.url, /^blob:/); handle.release(); handle.release(); + engine.editor.dispatch({ type: 'element.data', target: { id: 'a', incarnation: 1 }, patch: { title: 'independent' } }); + await until(() => f.writes.length === 1); + assert.equal(f.uploads.length, 0); + const fileBatch = f.batches.find(batch => batch.files?.length); + assert.equal(fileBatch.image.payload.document.elements.some(element => element.id === 'picture'), true); + assert.equal(fileBatch.image.payload.runtime.resources.length, 1); + gate.resolve(); await engine.settle(); + assert.equal(f.uploads.length, 1); assert.equal(f.writes.length, 2); + assert.equal(engine.getSnapshot().history.undo.length, 2); + assert.ok(engine.getSnapshot().runtime.resources.get(ref.id).stored); + assert.equal(engine.getSnapshot().document.elements.get('a').data.title, 'independent'); + engine.editor.dispatch({ type: 'element.create', element: node('copy', { assetId: ref.id, src: ref.src }) }); + await engine.settle(); assert.equal(f.uploads.length, 1); + assert.equal(engine.editor.history.undo().status, 'applied'); await engine.settle(); + assert.equal(f.uploads.length, 1); assert.equal(engine.getSnapshot().document.elements.has('copy'), false); + const pins = resourcePins(engine.getSnapshot(), engine.getSnapshot().runtime.ledger, registry, new Map()); + assert.equal(pins.complete, true); assert.ok([...pins.pins.get(ref.id)].some(owner => owner.startsWith('history:'))); + await engine.close(); +}); + +test('lost upload acknowledgment recovers original byte identity and query receipt without recreating a node', async () => { + const f = fixture(), upload = f.assets.upload; + f.assets.upload = async (...args) => { await upload(...args); return { outcome: 'unresolved' }; }; + const engine = await f.open(), ref = addImage(engine); await engine.settle(); + const frozen = f.uploads[0]; assert.ok(frozen); assert.equal(f.writes.length, 0); + const decoded = decodeRecovery(f.image(), registry); + assert.equal(decoded.runtime.resources.get(ref.id).id, ref.id); assert.equal(decoded.operations.size, 1); assert.equal(decoded.runtime.ledger.operations.size, 2); + const bad = structuredClone(f.image()); bad.payload.runtime.resources[0][1].operationId = 'not-the-upload'; + assert.throws(() => decodeRecovery(bad, registry), /corrupt_recovery_image/); + await engine.close(); + const queries = [], query = f.assets.query; + f.assets.query = async request => { queries.push(request); return query(request); }; + const recovered = await f.open({ clock: { now: () => Date.now() + 100000, setTimer: (fn, ms) => setTimeout(fn, ms), clearTimer: id => clearTimeout(id) } }); + await new Promise(resolve => setTimeout(resolve, 1)); await recovered.settle(); + assert.equal(queries.length, 1); assert.equal(queries[0].id, frozen.id); assert.equal(queries[0].body, frozen.body); + assert.equal(f.uploads.length, 1); assert.equal(f.writes.length, 1); + assert.equal(recovered.getSnapshot().document.elements.size, 2); assert.equal(recovered.getSnapshot().history.undo.length, 1); + assert.ok(recovered.getSnapshot().runtime.resources.get(ref.id).stored); + await recovered.close(); +}); + +test('rejected and empty staging do not retain file bytes or create upload work', async () => { + const f = fixture(), engine = await f.open(); + const files = [{ name: 'image.png', bytes: new Blob(['image'], { type: 'image/png' }) }]; + const result = engine.stageFiles(files, ([ref], tx) => { tx.dispatch({ type: 'element.create', element: node('a', { assetId: ref.id }) }); }); + assert.equal(result.status, 'rejected'); + assert.equal(engine.stageFiles(files, () => {}).operationIds.length, 0); + await engine.settle(); + assert.equal(f.files.size, 0); assert.equal(f.uploads.length, 0); assert.equal(engine.getSnapshot().runtime.resources.size, 0); + await engine.close(); +}); + +test('unknown resource reference schemas prevent claiming complete garbage-collection coverage', async () => { + const f = fixture(), engine = await f.open(); + const opaque = { ...node('future'), componentId: 'future-image', data: { privateResource: 'unknown' } }; + const doc = createDocument('project', [...original.canvases.values()], [...original.elements.values(), opaque]); + const snapshot = { ...engine.getSnapshot(), document: doc }; + assert.equal(resourcePins(snapshot, snapshot.runtime.ledger, registry, new Map()).complete, false); + const pinned = resourcePins(engine.getSnapshot(), engine.getSnapshot().runtime.ledger, registry, new Map([['export-job', ['keep']]])); + assert.deepEqual([...pinned.pins.get('keep')], ['export-job']); + await engine.close(); +}); + +test('invalid applied asset metadata retains the receipt and pending document without releasing the dependency', async () => { + const f = fixture(), upload = f.assets.upload; + f.assets.upload = async (...args) => { const result = await upload(...args); return { ...result, canonical: { ...result.canonical, id: 'different-resource' } }; }; + const errors = [], engine = await f.open({ onError: error => errors.push(error) }); + const ref = addImage(engine); await engine.settle(); + const request = f.uploads[0], stored = engine.getSnapshot().runtime.ledger.requests.get(request.id); + assert.equal(stored.outcome, 'applied'); assert.equal(stored.phase, 'waiting'); assert.equal(f.writes.length, 0); + assert.equal(engine.getSnapshot().runtime.receipts.get(request.id).id, 'different-resource'); + assert.equal(engine.getSnapshot().document.elements.get('picture').data.assetId, ref.id); + assert.equal(engine.getSnapshot().runtime.resources.get(ref.id).stored, undefined); assert.equal(errors[0].message, 'asset_receipt_mismatch'); + await engine.close(); +}); + +test('a failed image durable cut keeps the visible element and sends neither bytes nor document', async () => { + const f = fixture(), errors = [], engine = await f.open({ onError: error => errors.push(error) }); + await engine.persist(); + const commit = f.storage.commit; + f.storage.commit = async () => { throw new Error('quota'); }; + const ref = addImage(engine); await new Promise(resolve => setTimeout(resolve, 5)); + await assert.rejects(() => engine.settle(), /quota/); + assert.equal(f.uploads.length, 0); assert.equal(f.writes.length, 0); assert.equal(f.files.size, 0); + assert.equal(engine.getSnapshot().document.elements.get('picture').data.assetId, ref.id); + await assert.rejects(() => engine.close(), /quota/); + f.storage.commit = commit; await engine.close(); + assert.equal(await f.files.get(ref.id).text(), 'image bytes'); + assert.equal(decodeRecovery(f.image(), registry).history.undo.length, 1); + assert.ok(errors.length); +}); diff --git a/tests/headless/scheduler.test.mjs b/tests/headless/scheduler.test.mjs index 6b31a8d..39fde89 100644 --- a/tests/headless/scheduler.test.mjs +++ b/tests/headless/scheduler.test.mjs @@ -24,7 +24,7 @@ function setup(overrides = {}, initial) { }); 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(); }, + async advance(ms, settle = true) { time += ms; for (const [id, timer] of [...timers]) if (timer.at <= time) { timers.delete(id); timer.callback(); } await tick(); if (settle) await scheduler.settle(); }, timerCount: () => timers.size, }; } @@ -105,3 +105,41 @@ test('permission/protocol rejection keeps unknown claims without an automatic re assert.equal(state.ledger().getSnapshot().requests.get('request-1').outcome,'unresolved');assert.equal(state.scheduler.getClaims().size,1);assert.equal(state.timerCount(),0); await state.advance(10000);assert.equal(calls.length,1);state.scheduler.stop(); }); + +const mixedCompile = (intent, requestId) => ({ kind: intent.operationType, protocolVersion: 2, canvasIds: intent.operationType === 'asset.upload' ? [] : intent.changes.map(change => change.id), expectedVersions: Object.fromEntries(intent.changes.map(change => [change.id, 1])), payload: { requestId, operationId: intent.id } }); +function addAsset(state, id) { + state.ledger().add({ id, operationType: 'asset.upload', operationSchemaVersion: 1, sequence: state.ledger().getSnapshot().operations.size + 1, origin: 'user', changes: [], canvasIds: [], dependencies: [], resourceId: `resource-${id}` }, {}); + state.scheduler.notify([id]); +} + +test('an upload burst yields without filling both network slots or starving independent edits', async () => { + const gate = deferred(), calls = []; + const state = setup({ compile: mixedCompile, submit: async request => { calls.push(request); if (request.operationIds[0] === 'asset-0') await gate.promise; return { outcome: 'applied' }; } }); + for (let index = 0; index < 80; index++) addAsset(state, `asset-${index}`); + state.add('independent', ['canvas']); state.scheduler.start(); await tick(); + await state.advance(0, false); + assert.deepEqual(calls.map(request => request.operationIds[0]), ['asset-0', 'independent']); + assert.equal(state.timerCount(), 0); + gate.resolve(); await state.scheduler.settle(); + assert.equal(calls.length, 81); assert.equal(state.errors.length, 0); state.scheduler.stop(); +}); + +test('a capacity-blocked upload retry does not spin or hide a later document retry deadline', async () => { + const gate = deferred(), calls = []; + const state = setup({ compile: mixedCompile, + submit: async request => { + const id = request.operationIds[0]; calls.push(['submit', id]); + if (id === 'asset-2') { await gate.promise; return { outcome: 'applied' }; } + return { outcome: 'unresolved', retryAfterMs: id === 'doc' ? 200 : 100 }; + }, + query: async request => { calls.push(['query', request.operationIds[0]]); return { outcome: 'applied' }; }, + }); + addAsset(state, 'asset-1'); addAsset(state, 'asset-2'); state.add('doc'); state.scheduler.start(); await tick(); + await state.advance(100, false); assert.equal(state.timerCount(), 1); + await state.advance(100, false); + assert.ok(calls.some(([action, id]) => action === 'query' && id === 'doc')); + assert.equal(calls.some(([action, id]) => action === 'query' && id === 'asset-1'), false); + assert.equal(state.timerCount(), 0); + gate.resolve(); await state.scheduler.settle(); + assert.ok(calls.some(([action, id]) => action === 'query' && id === 'asset-1')); state.scheduler.stop(); +});