Zayden-Jung/Endoleap
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Persist staged image uploads alongside document operations

Browse Source
This commit is contained in:
Codex Bot
2026-11-07 23:24:00 +08:00
parent 8a19909716
commit 8a76e53a9f
20 changed files with 659 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/canvas-runtime/engine.ts
+125 -19
View File
@@ -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<Id, number>;
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<Id, Blob>(), digestTasks = new Map<Id, Promise<void>>();
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<Id, Id>();
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<Id, 'absent' | 'exists'> = 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<number, EngineSnapshot>();
@@ -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;
},