diff --git a/src/components/workspace/canvas-runtime-host.tsx b/src/components/workspace/canvas-runtime-host.tsx index 7bf4090..e920008 100644 --- a/src/components/workspace/canvas-runtime-host.tsx +++ b/src/components/workspace/canvas-runtime-host.tsx @@ -8,6 +8,7 @@ import { errorText } from "@/lib/api"; import { createRegistry, builtinComponents } from "@/lib/canvas-core/index"; import { createRuntimeHost, type RuntimeHost } from "@/lib/canvas-runtime/host"; import { createWorkspaceRuntime } from "@/lib/canvas-runtime/workspace"; +import { createBrowserDigests } from "@/lib/canvas-runtime/browser-digests"; import { createBrowserStorage } from "@/lib/canvas-runtime/storage"; import { createFoundryAssets, createFoundryPersistence, createFoundryTransport, createFoundryRuns } from "@/lib/canvas-service/index"; @@ -19,7 +20,7 @@ export function CanvasRuntimeHost({ children,extensions=[] }: { children: ReactN const registry = createRegistry([...builtinComponents(),...extensions.flatMap(extension=>extension.components??[])]),renderers=createReactRenderers([...defaultCanvasRenderers.definitions,...extensions.flatMap(extension=>extension.renderers??[])],registry); const host=createRuntimeHost({ registry, workspace(accountId, workspaceId, authEpoch, isCurrentIdentity) { - return createWorkspaceRuntime({ accountId, workspaceId, authEpoch, isCurrentIdentity, registry, storage: createBrowserStorage(), + return createWorkspaceRuntime({ accountId, workspaceId, authEpoch, isCurrentIdentity, registry, storage: createBrowserStorage(),digests:createBrowserDigests(), adapters(rootId, current) { const transport = createFoundryTransport({ accountId, isCurrentIdentity: current }); return { runs: createFoundryRuns({ workspaceId, rootId, registry, transport }), persistence: createFoundryPersistence({ workspaceId, rootId, registry, transport }), assets: createFoundryAssets({ workspaceId, transport, project: process.env.NEXT_PUBLIC_FOUNDRY_PROJECT || "endoleap", origin: process.env.NEXT_PUBLIC_FOUNDRY_API_ORIGIN }) }; diff --git a/src/lib/canvas-runtime/browser-digests.ts b/src/lib/canvas-runtime/browser-digests.ts new file mode 100644 index 0000000..c77f3d8 --- /dev/null +++ b/src/lib/canvas-runtime/browser-digests.ts @@ -0,0 +1,14 @@ +import {fault} from '../canvas-core/values.ts'; +import {createDigestPool} from './digest-pool.ts'; +/** Browser capability stays at the adapter; WorkspaceRuntime owns the pool's lifetime. */ +export function createBrowserDigests(){ + let next=0; + return createDigestPool({execute:typeof Worker==='undefined'?undefined:(bytes,signal)=>new Promise((resolve,reject)=>{ + if(signal.aborted){reject(fault('import_aborted'));return;} + let worker:Worker;try{worker=new Worker(new URL('./digest.worker.ts',import.meta.url),{type:'module'});}catch{reject(fault('digest_worker_unavailable'));return;} + const id=++next;let done=false; + const finish=(error?:unknown,sha256?:string)=>{if(done)return;done=true;signal.removeEventListener('abort',abort);worker.onmessage=null;worker.onerror=null;worker.onmessageerror=null;worker.terminate();if(error)reject(error);else resolve(sha256!);},abort=()=>finish(fault('import_aborted')); + signal.addEventListener('abort',abort,{once:true});worker.onmessage=event=>{const value=event.data;if(value?.version===1&&value.id===id&&value.byteLength===bytes.size&&value.error==='digest_worker_failed'){finish(fault('digest_worker_failed'));return;}if(!value||value.version!==1||value.id!==id||value.byteLength!==bytes.size||typeof value.sha256!=='string'||!(/^[a-f0-9]{64}$/.test(value.sha256)))finish(fault('invalid_digest_response'));else finish(undefined,value.sha256);};worker.onerror=event=>{event.preventDefault();finish(fault('digest_worker_failed'));};worker.onmessageerror=()=>finish(fault('invalid_digest_response')); + try{worker.postMessage({version:1,id,bytes});}catch{finish(fault('digest_worker_failed'));} + })}); +} diff --git a/src/lib/canvas-runtime/digest-pool.ts b/src/lib/canvas-runtime/digest-pool.ts new file mode 100644 index 0000000..9aaa599 --- /dev/null +++ b/src/lib/canvas-runtime/digest-pool.ts @@ -0,0 +1,24 @@ +import {fault} from '../canvas-core/values.ts'; +import {importSourceDigest} from './import-integrity.ts'; +export type DigestPort={digest(bytes:Blob,signal?:AbortSignal):Promise;close():void}; +/** Read-only compute scheduling: bounded active slots + FIFO Map, never durable operation facts. */ +export function createDigestPool(options:{execute?:(bytes:Blob,signal:AbortSignal)=>Promise;concurrency?:number;maxQueued?:number}={}):DigestPort{ + const concurrency=options.concurrency??2,maxQueued=options.maxQueued??64,execute=options.execute??importSourceDigest; + if(!Number.isSafeInteger(concurrency)||concurrency<1||concurrency>8||!Number.isSafeInteger(maxQueued)||maxQueued<1)throw fault('invalid_digest_budget'); + type Task={id:number;bytes:Blob;controller:AbortController;resolve:(value:string)=>void;reject:(error:unknown)=>void;cleanup:()=>void;done:boolean}; + const waiting=new Map(),active=new Map();let next=0,closed=false; + function finish(task:Task,error?:unknown,value?:string){if(task.done)return;task.done=true;task.cleanup();waiting.delete(task.id);if(error)task.reject(error);else task.resolve(value!);} + function pump(){ + if(closed)return; + while(active.size{if(task.controller.signal.aborted)throw fault('import_aborted');return execute(task.bytes,task.controller.signal);}).then(value=>{if(task.controller.signal.aborted)throw fault('import_aborted');if(!/^[a-f0-9]{64}$/.test(value))throw fault('invalid_digest_response');finish(task,undefined,value);},error=>finish(task,error)).catch(error=>finish(task,error)).finally(()=>{active.delete(task.id);pump();}); + } + } + return Object.freeze({ + digest(bytes:Blob,signal?:AbortSignal):Promise{ + if(closed)return Promise.reject(fault('digest_pool_closed'));if(signal?.aborted)return Promise.reject(fault('import_aborted'));if(!(bytes instanceof Blob))return Promise.reject(fault('invalid_digest_input'));if(active.size>=concurrency&&waiting.size>=maxQueued)return Promise.reject(fault('digest_capacity')); + return new Promise((resolve,reject)=>{const task:Task={id:++next,bytes,controller:new AbortController(),resolve,reject,cleanup:()=>signal?.removeEventListener('abort',abort),done:false},abort=()=>{task.controller.abort();finish(task,fault('import_aborted'));pump();};signal?.addEventListener('abort',abort,{once:true});waiting.set(task.id,task);pump();}); + }, + close(){if(closed)return;closed=true;for(const task of [...waiting.values(),...active.values()]){task.controller.abort();finish(task,fault('import_aborted'));}}, + }); +} diff --git a/src/lib/canvas-runtime/digest.worker.ts b/src/lib/canvas-runtime/digest.worker.ts new file mode 100644 index 0000000..016844b --- /dev/null +++ b/src/lib/canvas-runtime/digest.worker.ts @@ -0,0 +1,8 @@ +import {importSourceDigest} from './import-integrity.ts'; +// Dedicated to one request. Host terminates on success/failure/abort; no storage, model or credentials here. +const scope=globalThis as unknown as {onmessage:((event:MessageEvent)=>void)|null;postMessage(value:unknown):void}; +let accepted=false; +scope.onmessage=event=>{ + const request=event.data;if(accepted||!request||request.version!==1||!Number.isSafeInteger(request.id)||request.id<1||!(request.bytes instanceof Blob))return;accepted=true; + void importSourceDigest(request.bytes,undefined,{chunkBytes:1024*1024,yieldTask:()=>Promise.resolve()}).then(sha256=>scope.postMessage({version:1,id:request.id,byteLength:request.bytes.size,sha256}),()=>scope.postMessage({version:1,id:request.id,byteLength:request.bytes.size,error:'digest_worker_failed'})); +}; diff --git a/src/lib/canvas-runtime/engine.ts b/src/lib/canvas-runtime/engine.ts index eef4e94..4152040 100644 --- a/src/lib/canvas-runtime/engine.ts +++ b/src/lib/canvas-runtime/engine.ts @@ -2,6 +2,7 @@ import { encodeIndexedCopySnapshot, openIndexedCopySnapshot, type IndexedCopySna import { decodeCopySnapshot, copyRegistrySignature } from './copy-snapshot.ts'; import { createDuplicatePlanSource } from './duplicate-source.ts'; import { planDuplicateAsync, planDuplicateResumable, prepareDuplicateStream } from './duplicate.ts'; +import type { DigestPort } from './digest-pool.ts'; import { importSourceDigest } from './import-integrity.ts'; import { encodeDuplicateSource, prepareDuplicateImport, assertDuplicateExternal } from './import-duplicate.ts'; import { createImportJob, createImportRunner, createNdjsonImportSource, type ImportJob } from './imports.ts'; @@ -24,9 +25,10 @@ import { stageAppliedRecovery } from './applied-recovery.ts'; import { conflictsWith, maxResourceBytes, operationFootprint, referencedResources, type AssetPort, type ResourceRecord, type ResourceHandle, type StagedFile } from './resources.ts'; import type { StoragePort, StorageScope } from './storage.ts'; +const digestFailures=new Set(['digest_capacity','digest_worker_failed','digest_worker_unavailable','invalid_digest_response','digest_pool_closed']); export type EngineOptions = { scope: StorageScope; registry: Registry; document: Document; versions: ReadonlyMap; - storage: StoragePort; persistence: PersistencePort; assets?: AssetPort; runs?: RunPort; createId?: () => Id; + storage: StoragePort; digests?:Pick; persistence: PersistencePort; assets?: AssetPort; runs?: RunPort; createId?: () => Id; historyLimit?: number; policy?: EditorOptions['policy']; traversalPolicy?: EditorOptions['traversalPolicy']; clock?: Clock; onError?: (error: unknown) => void; }; @@ -52,6 +54,8 @@ export async function openProjectEngine(options: EngineOptions) { if (recovered && recovered.document.projectId !== options.scope.projectId) throw fault('project_mismatch'); for (const canvas of options.document.canvases.keys()) if (!Number.isSafeInteger(options.versions.get(canvas)) || options.versions.get(canvas)! < 1) throw fault('missing_server_version', [canvas]); let closing = false, closed = false, durableRootRevision = image?.rootRevision ?? -1; + const digestLifecycle=new AbortController(); + const sourceDigest=(bytes:Blob,signal?:AbortSignal)=>{const combined=signal?AbortSignal.any([signal,digestLifecycle.signal]):digestLifecycle.signal;return options.digests?options.digests.digest(bytes,combined):importSourceDigest(bytes,combined);}; const report = (error: unknown) => { try { options.onError?.(error); } catch { /* Error reporting cannot alter state. */ } }; const createId = options.createId ?? (() => crypto.randomUUID()); const network = createNetworkGate(2); @@ -227,7 +231,7 @@ export async function openProjectEngine(options: EngineOptions) { if (request.kind !== 'asset.upload') return network.run(signal, () => 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' }; + if (!bytes || bytes.size !== payload.byte_length || await digest(bytes,signal) !== payload.sha256) return { outcome: 'unresolved', blocked: true, error: 'resource_bytes_unavailable' }; return network.run(signal, () => options.assets!.upload(request, bytes, signal)); }, query: (request, signal) => network.run(signal, () => ['run.start','run.cancel','run.apply'].includes(request.kind??'') ? options.runs?.query(request,signal) ?? Promise.resolve({ outcome: 'unresolved', blocked: true, error: 'run_interpreter_unavailable' }) : request.kind === 'asset.upload' ? options.assets?.query(request, signal) ?? Promise.resolve({ outcome: 'unresolved', blocked: true, error: 'asset_interpreter_unavailable' }) : options.persistence.query(request, signal)), @@ -288,7 +292,7 @@ export async function openProjectEngine(options: EngineOptions) { const operationIds = [...commit.operationIds, ...staged.map(resource => resource.operationId)]; queueMicrotask(() => { if (!closing && !closed) { scheduler.notify(operationIds); if (operationIds.length) prepareDigests(); } }); }); - 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(''); } + async function digest(bytes:Blob,signal?:AbortSignal){return sourceDigest(bytes,signal);} function prepareDigests() { if (closing || closed) return; for (const resource of editor.getSnapshot().runtime.resources.values()) { @@ -309,7 +313,7 @@ export async function openProjectEngine(options: EngineOptions) { }); 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(); }); + }).catch(error => { if (!closing && !closed) { const code=(error as {code?:string}).code;mutate(ledger => ledger.block(resource.operationId, [code&&digestFailures.has(code)?code:'resource_bytes_unavailable'])); report(error); } }).finally(() => { digestTasks.delete(resource.id); prepareDigests(); }); digestTasks.set(resource.id, task); } } @@ -350,7 +354,7 @@ export async function openProjectEngine(options: EngineOptions) { publishImport(freeze({...job,generation:job.generation+1,preparation:{...previous,steps:Math.max(previous.steps,steps),allocationIds:[...allocationIds]}}),job.generation);await persist();check(); }}); check();if(preparation.version===1&&cursor!==allocationIds.length)throw fault('copy_interpreter_changed'); - const encoded=await encodeDuplicateSource(document,options.registry,plan,controller.signal),sha256=await importSourceDigest(encoded,controller.signal),job=check();assertWritable();assertNativeImportTarget(job,editor.getSnapshot().document); + const encoded=await encodeDuplicateSource(document,options.registry,plan,controller.signal),sha256=await sourceDigest(encoded,controller.signal),job=check();assertWritable();assertNativeImportTarget(job,editor.getSnapshot().document); const permission=options.policy?.({type:'changes.apply',changes:plan.changes},editor.getSnapshot().document);if(permission&&!permission.enabled)throw fault(permission.code,permission.targets); const fileId=createId(),runtime=editor.getSnapshot().runtime;if(!fileId||fileWrites.has(fileId)||runtime.resources.has(fileId)||[...(runtime.imports?.values()??[])].some(job=>job.source?.fileId===fileId))throw fault('import_identity_conflict'); const stableIds=Object.fromEntries([...Object.entries(plan.elementIds).map(([old,id])=>[`element:${old}`,id]),...Object.entries(plan.canvasIds).map(([old,id])=>[`canvas:${old}`,id])]); @@ -371,7 +375,7 @@ export async function openProjectEngine(options: EngineOptions) { const latest=getImport(id);if(latest.sourceFingerprint!==job.sourceFingerprint||latest.source?.fileId!==source.fileId||['completed','partial','cancelled'].includes(latest.status))return; if(!bytes)throw fault('import_source_unavailable');if(bytes.size!==source.byteLength)throw fault('import_source_changed'); if(source.integrityVersion!==1)throw fault('import_source_integrity_unavailable'); - const sha256=await importSourceDigest(bytes,preparationController.signal); + const sha256=await sourceDigest(bytes,preparationController.signal); if(closing||closed)return; const verified=getImport(id);if(verified.generation!==latest.generation||verified.sourceFingerprint!==job.sourceFingerprint||verified.source?.fileId!==source.fileId||['completed','partial','cancelled','paused'].includes(verified.status))return; if(importHolds.size){deferredImportLoads.add(id);return;} @@ -594,12 +598,12 @@ export async function openProjectEngine(options: EngineOptions) { getSnapshot: editor.getSnapshot, persist, cancelRecovery:async()=>{await recovery?.cancel();}, getDurability: () => Object.freeze({ rootRevision: editor.getSnapshot().rootRevision, durableRootRevision }), - retry: async () => { if(recovery)await recovery.cancel();assertWritable();await scheduler.resume();await runTracker.retry(); }, + retry: async () => { if(recovery)await recovery.cancel();assertWritable();const runtime=editor.getSnapshot().runtime,ids=[...runtime.resources.values()].filter(resource=>!resource.digest&&runtime.ledger.operations.get(resource.operationId)?.blockers.some(code=>digestFailures.has(code))).map(resource=>resource.operationId);if(ids.length){mutate(ledger=>{for(const id of ids)ledger.block(id,['waiting_digest']);});prepareDigests();}await scheduler.resume();await runTracker.retry(); }, async settle() { await Promise.resolve();await Promise.allSettled([...copyPreparations.values()]);while(importLoads.size)await Promise.all([...importLoads.values()]);await Promise.all([...importRunners.values()].map(runner=>runner.settle())); while (digestTasks.size) await Promise.all([...digestTasks.values()]); await scheduler.settle(); await runTracker.settle(); await persist(); }, async close() { if (closed) return; - closing = true;for(const controller of preparationControllers.values())controller.abort();for(const controller of copyPreparations.keys())controller.abort();await Promise.allSettled([...copyPreparations.values()]); sessions.dispose(); scheduler.stop(); runTracker.stop(); - await Promise.all([...importRunners.values()].map(runner=>runner.close()));while(importLoads.size)await Promise.all([...importLoads.values()]); + closing = true;digestLifecycle.abort();for(const controller of preparationControllers.values())controller.abort();for(const controller of copyPreparations.keys())controller.abort();await Promise.allSettled([...copyPreparations.values()]); sessions.dispose(); scheduler.stop(); runTracker.stop(); + await Promise.all([...importRunners.values()].map(runner=>runner.close()));while(importLoads.size)await Promise.all([...importLoads.values()]);while(digestTasks.size)await Promise.allSettled([...digestTasks.values()]); recovery?.controller.abort();recovery?.release(); for (const release of [...displays]) release(); // Failure keeps the leader/recovery material available for an explicit retry of close. diff --git a/src/lib/canvas-runtime/import-integrity.ts b/src/lib/canvas-runtime/import-integrity.ts index 1a39f1b..6cd1d8e 100644 --- a/src/lib/canvas-runtime/import-integrity.ts +++ b/src/lib/canvas-runtime/import-integrity.ts @@ -1,8 +1,10 @@ import { fault } from '../canvas-core/values.ts'; -/** Bind durable preparation/parse checkpoints to the complete immutable byte source. */ -export async function importSourceDigest(bytes:Blob,signal?:AbortSignal):Promise{ - if(signal?.aborted)throw fault('import_aborted'); - const buffer=await bytes.arrayBuffer();if(signal?.aborted)throw fault('import_aborted'); - const digest=await crypto.subtle.digest('SHA-256',buffer);if(signal?.aborted)throw fault('import_aborted'); - return [...new Uint8Array(digest)].map(byte=>byte.toString(16).padStart(2,'0')).join(''); +import {createSha256} from './sha256.ts'; +/** Complete-source SHA-256, with bounded chunk memory and cooperative fallback scheduling. */ +export async function importSourceDigest(bytes:Blob,signal?:AbortSignal,options:{chunkBytes?:number;yieldTask?:()=>Promise}={}):Promise{ + const chunk=options.chunkBytes??64*1024,yieldTask=options.yieldTask??(()=>new Promise(resolve=>setTimeout(resolve,0))); + if(!(bytes instanceof Blob)||!Number.isSafeInteger(chunk)||chunk<1||chunk>4*1024*1024)throw fault('invalid_digest_input'); + const check=()=>{if(signal?.aborted)throw fault('import_aborted');};check();const hash=createSha256(); + for(let cursor=0;cursor(n>>>bits)|(n<<(32-bits)); +/** Incremental bytes only; owns at most a partial block, state and message schedule. */ +export function createSha256(){ + const state=new Uint32Array([0x6a09e667,0xbb67ae85,0x3c6ef372,0xa54ff53a,0x510e527f,0x9b05688c,0x1f83d9ab,0x5be0cd19]),block=new Uint8Array(64),words=new Uint32Array(64);let pending=0,length=0,result:string|undefined; + function compress(bytes:Uint8Array,offset:number){ + const view=new DataView(bytes.buffer,bytes.byteOffset+offset,64);for(let i=0;i<16;i++)words[i]=view.getUint32(i*4,false); + for(let i=16;i<64;i++){const x=words[i-15],y=words[i-2],s0=rotate(x,7)^rotate(x,18)^(x>>>3),s1=rotate(y,17)^rotate(y,19)^(y>>>10);words[i]=(words[i-16]+s0+words[i-7]+s1)>>>0;} + let[a,b,c,d,e,f,g,h]=state; + for(let i=0;i<64;i++){const s1=rotate(e,6)^rotate(e,11)^rotate(e,25),ch=(e&f)^(~e&g),t1=(h+s1+ch+K[i]+words[i])>>>0,s0=rotate(a,2)^rotate(a,13)^rotate(a,22),maj=(a&b)^(a&c)^(b&c),t2=(s0+maj)>>>0;h=g;g=f;f=e;e=(d+t1)>>>0;d=c;c=b;b=a;a=(t1+t2)>>>0;} + const values=[a,b,c,d,e,f,g,h];for(let i=0;i<8;i++)state[i]=(state[i]+values[i])>>>0; + } + return Object.freeze({ + update(bytes:Uint8Array){ + if(result!==undefined)throw fault('digest_finalized');if(!(bytes instanceof Uint8Array)||!Number.isSafeInteger(length+bytes.byteLength))throw fault('invalid_digest_input');length+=bytes.byteLength;let offset=0; + if(pending){const count=Math.min(64-pending,bytes.byteLength);block.set(bytes.subarray(0,count),pending);pending+=count;offset=count;if(pending===64){compress(block,0);pending=0;}} + while(offset+64<=bytes.byteLength){compress(bytes,offset);offset+=64;} + if(offset56){block.fill(0,pending);compress(block,0);pending=0;} + block.fill(0,pending,56);new DataView(block.buffer).setBigUint64(56,BigInt(length)*8n,false);compress(block,0);result=[...state].map(word=>word.toString(16).padStart(8,'0')).join('');block.fill(0);words.fill(0);state.fill(0);return result; + }, + }); +} diff --git a/src/lib/canvas-runtime/workspace.ts b/src/lib/canvas-runtime/workspace.ts index 871c18f..e33d26c 100644 --- a/src/lib/canvas-runtime/workspace.ts +++ b/src/lib/canvas-runtime/workspace.ts @@ -1,11 +1,12 @@ import type { Document, Id, Registry } from '../canvas-core/types.ts'; import { fault, readMap } from '../canvas-core/values.ts'; import { openProjectEngine, type EngineOptions } from './engine.ts'; +import {createDigestPool,type DigestPort} from './digest-pool.ts'; import type { StoragePort } from './storage.ts'; export type ProjectEngine = Awaited>; export type WorkspaceOptions = { - accountId: Id; workspaceId: Id; authEpoch: number; registry: Registry; storage: StoragePort; + accountId: Id; workspaceId: Id; authEpoch: number; registry: Registry; storage: StoragePort; digests?:DigestPort; isCurrentIdentity(): boolean; adapters(projectId: Id, current: () => boolean): Pick; onError?: (error: unknown) => void; @@ -13,6 +14,7 @@ export type WorkspaceOptions = { /** Workspace owns persistent projects; a view only borrows one and never closes it. */ export function createWorkspaceRuntime(options: WorkspaceOptions) { + const digests=options.digests??createDigestPool(); const projects = new Map>(); let accepting = true, closed = false, closing: Promise | undefined; const current = () => accepting && !closed && options.isCurrentIdentity(); @@ -24,7 +26,7 @@ export function createWorkspaceRuntime(options: WorkspaceOptions) { if (!current()) return Promise.reject(fault('identity_changed')); const existing = projects.get(document.projectId); if (existing) return existing.then(engine => { if (!current()) throw fault('identity_changed'); engine.refresh(document, versions); return engine; }); - const opening = openProjectEngine({ scope: { accountId: options.accountId, workspaceId: options.workspaceId, projectId: document.projectId }, registry: options.registry, document, versions, storage: options.storage, ...options.adapters(document.projectId, current), onError: options.onError, + const opening = openProjectEngine({ scope: { accountId: options.accountId, workspaceId: options.workspaceId, projectId: document.projectId }, registry: options.registry, document, versions, storage: options.storage, digests, ...options.adapters(document.projectId, current), onError: options.onError, traversalPolicy: () => current() ? { enabled: true } : { enabled: false, code: 'identity_changed', targets: [] }, policy: () => current() ? { enabled: true } : { enabled: false, code: 'identity_changed', targets: [] }, }); @@ -44,7 +46,7 @@ export function createWorkspaceRuntime(options: WorkspaceOptions) { })); const errors = results.flatMap(result => result.status === 'rejected' ? [result.reason] : []); if (errors.length) throw new AggregateError(errors, 'workspace_close_failed'); - options.storage.close(); projects.clear(); closed = true; + digests.close();options.storage.close(); projects.clear(); closed = true; })(); closing = task.finally(() => { closing = undefined; }); return closing; diff --git a/tests/headless/bindings.test.mjs b/tests/headless/bindings.test.mjs index 814e6ee..9d821a5 100644 --- a/tests/headless/bindings.test.mjs +++ b/tests/headless/bindings.test.mjs @@ -68,20 +68,20 @@ test('late connection changes cancel preview without restoring stale path or unr test('workspace retains a failed close for retry and stops edits before releasing its persistent owner', async () => { const { createWorkspaceRuntime } = await import('../../src/lib/canvas-runtime/workspace.ts'); - let image = null, leader, broken = false, closed = false, server = document, version = 1; + let image = null, leader, broken = false, closed = false, digestClosed=false, server = document, version = 1; const receipts = new Map(); const storage = { async claim(scope) { leader = { scope, fence: 1, active: true, async release() { leader.active = false; } }; return leader; }, async recover() { return image; }, async commit(_leader, batch) { if (broken) throw new Error('quota'); image = structuredClone(batch.image); return { durableRootRevision: image.rootRevision }; }, async readFile() {}, close() { closed = true; } }; const persistence = { compile: (intent, requestId, snapshot) => ({ canvasIds: intent.canvasIds, expectedVersions: Object.fromEntries(intent.canvasIds.map(id => [id, snapshot.runtime.versions.get(id)])), payload: { intent, requestId } }), async submit(request) { const { applyChanges } = await import('../../src/lib/canvas-core/document.ts'); server = applyChanges(server, JSON.parse(request.body).intent.changes); version++; receipts.set(request.id, { document: server, versions: new Map([['root', version], ['child', 1]]) }); return { outcome: 'applied' }; }, async query() { return { outcome: 'unresolved' }; }, canonical: request => receipts.get(request.id) }; - const workspace = createWorkspaceRuntime({ accountId: 'account', workspaceId: 'workspace', authEpoch: 1, registry, storage, isCurrentIdentity: () => true, adapters: () => ({ persistence }) }); + const workspace = createWorkspaceRuntime({ accountId: 'account', workspaceId: 'workspace', authEpoch: 1, registry, storage, isCurrentIdentity: () => true,digests:{digest:async()=> 'a'.repeat(64),close(){digestClosed=true;}}, adapters: () => ({ persistence }) }); const [a, b] = await Promise.all([workspace.open(document, new Map([['root', 1], ['child', 1]])), workspace.open(document, new Map([['root', 1], ['child', 1]]))]); assert.equal(a, b); const detach = a.sessions.attach('first'); a.sessions.attach('second'); detach(); a.editor.dispatch({ type: 'element.data', target: { id: 'outside', incarnation: 1 }, patch: { title: 'survives view detach' } }); await a.settle(); broken = true; a.editor.dispatch({ type: 'element.data', target: { id: 'outside', incarnation: 1 }, patch: { title: 'not durable yet' } }); - await assert.rejects(() => workspace.close(), /workspace_close_failed/); assert.equal(leader.active, true); assert.equal(closed, false); + await assert.rejects(() => workspace.close(), /workspace_close_failed/); assert.equal(leader.active, true); assert.equal(closed, false);assert.equal(digestClosed,false); assert.equal(a.editor.dispatch({ type: 'element.data', target: { id: 'outside', incarnation: 1 }, patch: { title: 'late' } }).status, 'rejected'); assert.equal(workspace.getProjects().size, 1); - broken = false; await workspace.close(); assert.equal(leader.active, false); assert.equal(closed, true); + broken = false; await workspace.close(); assert.equal(leader.active, false); assert.equal(closed, true);assert.equal(digestClosed,true); }); test('future curve formats remain readable and opaque while appearance edits preserve their path', () => { diff --git a/tests/headless/browser-digests.test.mjs b/tests/headless/browser-digests.test.mjs new file mode 100644 index 0000000..b049d76 --- /dev/null +++ b/tests/headless/browser-digests.test.mjs @@ -0,0 +1,13 @@ +import {test} from 'node:test'; +import assert from 'node:assert/strict'; +import {createBrowserDigests} from '../../src/lib/canvas-runtime/browser-digests.ts'; + +test('browser digest adapter rejects mismatched worker identity and terminates the worker',async()=>{ + const previous=globalThis.Worker;let terminated=0; + globalThis.Worker=class{postMessage(value){queueMicrotask(()=>this.onmessage({data:{version:1,id:value.id+1,byteLength:value.bytes.size,sha256:'a'.repeat(64)}}));}terminate(){terminated++;}}; + const pool=createBrowserDigests();try{await assert.rejects(pool.digest(new Blob(['bytes'])),/invalid_digest_response/);assert.equal(terminated,1);}finally{pool.close();if(previous===undefined)delete globalThis.Worker;else globalThis.Worker=previous;} +}); +test('browser digest adapter exposes worker startup and runtime failures without returning a false hash',async()=>{ + const previous=globalThis.Worker;let pool; + try{globalThis.Worker=class{constructor(){throw Error('denied');}};pool=createBrowserDigests();await assert.rejects(pool.digest(new Blob()),/digest_worker_unavailable/);pool.close();let terminated=0;globalThis.Worker=class{postMessage(){queueMicrotask(()=>this.onerror({preventDefault(){}}));}terminate(){terminated++;}};pool=createBrowserDigests();await assert.rejects(pool.digest(new Blob()),/digest_worker_failed/);assert.equal(terminated,1);}finally{pool?.close();if(previous===undefined)delete globalThis.Worker;else globalThis.Worker=previous;} +}); diff --git a/tests/headless/browser-storage-cases.mjs b/tests/headless/browser-storage-cases.mjs index 0ba2892..e1332e1 100644 --- a/tests/headless/browser-storage-cases.mjs +++ b/tests/headless/browser-storage-cases.mjs @@ -1,3 +1,4 @@ +import {createBrowserDigests} from '../../src/lib/canvas-runtime/browser-digests.ts'; import {sealRecoveryImage} from '../../src/lib/canvas-runtime/recovery-encoding.ts'; import {planDuplicateAsync} from '../../src/lib/canvas-runtime/duplicate.ts'; import {replayHistoryEvidence,sameHistory} from '../../src/lib/canvas-runtime/history-replay.ts'; @@ -8,6 +9,16 @@ import { createFoundryRuns } from '../../src/lib/canvas-service/runs.ts'; import { createFoundryAssets } from '../../src/lib/canvas-service/assets.ts'; import { createFoundryPersistence } from '../../src/lib/canvas-service/foundry.ts'; +async function digestWorkerCase(assert){ + const NativeWorker=globalThis.Worker;let active=0,peak=0,created=0;const replies=[]; + globalThis.Worker=class extends NativeWorker{constructor(...args){super(...args);created++;active++;peak=Math.max(peak,active);this.live=true;this.addEventListener('message',event=>replies.push(event.data));}terminate(){if(this.live){this.live=false;active--;}super.terminate();}}; + const pool=createBrowserDigests();try{ + const bytes=new Blob([new Uint8Array(4*1024*1024).fill(97)]),expected=[...new Uint8Array(await crypto.subtle.digest('SHA-256',await bytes.arrayBuffer()))].map(value=>value.toString(16).padStart(2,'0')).join(''),cancel=new AbortController(),a=pool.digest(bytes),b=pool.digest(bytes),queued=pool.digest(bytes,cancel.signal),cancelled=queued.then(()=>{throw Error('queued digest ran');},error=>assert(error.code==='import_aborted','queued digest cancellation changed')); + cancel.abort();await cancelled;const results=await Promise.all([a,b]);assert(results.every(value=>value===expected),'worker digest differs from native SHA256');assert(peak===2&&created===2&&active===0,'worker pool exceeded bounds or leaked a worker');assert(replies.length===2&&replies.every(reply=>reply.version===1&&reply.byteLength===bytes.size),'worker protocol did not actually run'); + const late=pool.digest(bytes),closed=late.then(()=>{throw Error('closed worker published');},error=>assert(error.code==='import_aborted','pool close did not cancel work'));await Promise.resolve();pool.close();await closed;assert(active===0,'pool close retained a worker'); + }finally{pool.close();globalThis.Worker=NativeWorker;} +} + async function ownedImageStorageCase(name,assert){ const storage=createBrowserStorage({name:`${name}-owned-image`}),scope={accountId:'owner',workspaceId:'workspace',projectId:'project'};let leader; try{ @@ -597,6 +608,8 @@ export async function run() { 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 digestWorkerCase(assert); + results.push('bounded real module workers hash chunked Blobs with native SHA256 parity, queued abort and close terminate work without a late result'); await ownedImageStorageCase(name,assert); results.push('owned immutable recovery images survive IndexedDB checksums; mutable images and file descriptors are captured before asynchronous work'); await cooperativeCopyCase(assert); diff --git a/tests/headless/digest-pool.test.mjs b/tests/headless/digest-pool.test.mjs new file mode 100644 index 0000000..aa9493f --- /dev/null +++ b/tests/headless/digest-pool.test.mjs @@ -0,0 +1,13 @@ +import {test} from 'node:test'; +import assert from 'node:assert/strict'; +import {createDigestPool} from '../../src/lib/canvas-runtime/digest-pool.ts'; +const hash='a'.repeat(64),tick=()=>new Promise(resolve=>setTimeout(resolve,0)); +test('digest pool bounds execution and FIFO queue, rejects overload and drops queued cancellation',async()=>{ + const started=[],release=[];const pool=createDigestPool({concurrency:2,maxQueued:2,execute:async bytes=>{started.push(await bytes.text());return new Promise(resolve=>release.push(()=>resolve(hash)));}}),cancel=new AbortController(),a=pool.digest(new Blob(['A'])),b=pool.digest(new Blob(['B'])),c=pool.digest(new Blob(['C']),cancel.signal),d=pool.digest(new Blob(['D']));const rejected=assert.rejects(c,/import_aborted/);await assert.rejects(pool.digest(new Blob(['overflow'])),/digest_capacity/);cancel.abort();await rejected;await tick();assert.deepEqual(started,['A','B']);release[0]();await a;await tick();assert.deepEqual(started,['A','B','D']);release[1]();release[2]();await Promise.all([b,d]);pool.close();await assert.rejects(pool.digest(new Blob()),/digest_pool_closed/); +}); +test('active cancellation cannot free a compute slot before its executor stops; late output is ignored',async()=>{ + let release;const started=[],controller=new AbortController(),pool=createDigestPool({concurrency:1,execute:async bytes=>{started.push(await bytes.text());if(started.length===1)await new Promise(resolve=>release=resolve);return hash;}}),first=pool.digest(new Blob(['first']),controller.signal),rejected=assert.rejects(first,/import_aborted/),second=pool.digest(new Blob(['second']));await tick();controller.abort();await rejected;await tick();assert.deepEqual(started,['first']);release();assert.equal(await second,hash);pool.close(); +}); +test('closing digest pool aborts active/queued tasks once and malformed executor output is rejected',async()=>{ + const signals=[],pool=createDigestPool({concurrency:1,execute:(_bytes,signal)=>{signals.push(signal);return new Promise((_,reject)=>signal.addEventListener('abort',()=>reject(Error('stopped')),{once:true}));}}),a=pool.digest(new Blob()),b=pool.digest(new Blob()),failures=[assert.rejects(a,/import_aborted/),assert.rejects(b,/import_aborted/)];await tick();pool.close();pool.close();await Promise.all(failures);assert.equal(signals.length,1);assert.equal(signals[0].aborted,true);const invalid=createDigestPool({execute:async()=> 'bad'});await assert.rejects(invalid.digest(new Blob()),/invalid_digest_response/);invalid.close(); +}); diff --git a/tests/headless/imports-engine.test.mjs b/tests/headless/imports-engine.test.mjs index 43aafe9..cf83f02 100644 --- a/tests/headless/imports-engine.test.mjs +++ b/tests/headless/imports-engine.test.mjs @@ -187,7 +187,7 @@ test('a changed unparsed tail cannot pass native import recovery or alter its pr test('late source digest cannot revive a cancelled job or publish into a held recovery root',async()=>{ for(const cancel of [true,false]){ - const f=fixture(),engine=await f.open(),entered=deferred(),gate=deferred(),bytes=new class extends Blob{async arrayBuffer(){entered.resolve();await gate.promise;return super.arrayBuffer();}}([JSON.stringify(node('one'))]); + const f=fixture(),engine=await f.open(),entered=deferred(),gate=deferred(),bytes=new class extends Blob{slice(...args){return new class extends Blob{async arrayBuffer(){entered.resolve();await gate.promise;return super.arrayBuffer();}}([super.slice(...args)]);}}([JSON.stringify(node('one'))]); try{const id=engine.imports.startNdjson(bytes,'canvas');await entered.promise;if(cancel){await engine.imports.cancel(id);gate.resolve();await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'cancelled');assert.equal(engine.getSnapshot().document.elements.size,0);assert.equal(engine.imports.getJobs().get(id).source.sha256,undefined);}else{const preparing=engine.prepareRecovery();gate.resolve();const session=await preparing,revision=engine.getSnapshot().rootRevision;await new Promise(resolve=>setTimeout(resolve,10));assert.equal(engine.getSnapshot().rootRevision,revision);assert.equal((await session.apply()).status,'applied');await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(f.server().elements.size,1);}assert.equal(f.errors.length,0);}finally{gate.resolve();await engine.close();} } }); @@ -219,7 +219,7 @@ test('late planned-source decoding respects cancellation and canonical recovery const f=fixture();let engine=await f.open();const entered=deferred(),gate=deferred();try{ await seedCopy(engine);const id=await engine.imports.duplicate([{id:'F',incarnation:1}]);await until(engine,snapshot=>snapshot.runtime.imports.get(id).committedSourceKeys.length===1);await engine.imports.pause(id);await engine.settle();const before=engine.imports.getJobs().get(id),size=engine.getSnapshot().document.elements.size;await engine.close(); class DelayedChunk extends Blob{async arrayBuffer(){entered.resolve();await gate.promise;return super.arrayBuffer();}} - class DelayedSource extends Blob{slice(...args){return new DelayedChunk([super.slice(...args)]);}} + class DelayedSource extends Blob{slice(start,end,...rest){const part=super.slice(start,end,...rest);return end-start===16*1024?new DelayedChunk([part]):part;}} f.files.set(before.source.fileId,new DelayedSource([f.files.get(before.source.fileId)]));engine=await f.open();const resuming=engine.imports.resume(id);await entered.promise; if(cancel){await engine.imports.cancel(id);gate.resolve();await resuming;await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'cancelled');assert.equal(engine.getSnapshot().document.elements.size,size);}else{const preparing=engine.prepareRecovery();gate.resolve();await resuming;const session=await preparing,revision=engine.getSnapshot().rootRevision;await new Promise(resolve=>setTimeout(resolve,10));assert.equal(engine.getSnapshot().rootRevision,revision);assert.equal(engine.getSnapshot().document.elements.size,size);assert.equal((await session.apply()).status,'applied');await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');}assert.equal(f.errors.length,0); }finally{gate.resolve();await engine.close();} @@ -231,7 +231,13 @@ test('indexed copy displays a prefix while a later raw payload read is still blo const id=await engine.imports.duplicate([{id:'F',incarnation:1}]);await until(engine,snapshot=>!!snapshot.runtime.imports.get(id).source.sha256);await engine.imports.pause(id);await engine.settle();const before=engine.imports.getJobs().get(id);await engine.close();const bytes=f.files.get(before.source.fileId),rows=(await bytes.text()).trimEnd().split('\n');let payload=0; for(const text of rows){const row=JSON.parse(text);payload+=new TextEncoder().encode(text+'\n').byteLength;if(row.kind==='payload')break;} class DelayedChunk extends Blob{async arrayBuffer(){entered.resolve();await gate.promise;return super.arrayBuffer();}} - class DelayedPayload extends Blob{slice(start,end,...rest){const part=super.slice(start,end,...rest);return start>=payload+64*1024?new DelayedChunk([part]):part;}} + class DelayedPayload extends Blob{slice(start,end,...rest){const part=super.slice(start,end,...rest);return start>=payload+64*1024&&(start-payload)%(64*1024)===0?new DelayedChunk([part]):part;}} f.files.set(before.source.fileId,new DelayedPayload([bytes]));engine=await f.open();await engine.imports.resume(id);await entered.promise;const cut=engine.getSnapshot(),job=cut.runtime.imports.get(id);assert.equal(job.source.snapshotVersion,2);assert.equal(job.status,'running');assert.ok(job.committedSourceKeys.length>0&&job.committedSourceKeys.length<102);assert.ok(cut.document.elements.size>source.elements.size&&cut.document.elements.size<202,'raw tail read gated the entire copy');gate.resolve();await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(engine.getSnapshot().document.elements.size,202);assert.equal(f.errors.length,0); }finally{gate.resolve();await engine.close();} }); + + +test('closing a project aborts its injected digest without closing another project sharing the pool',async()=>{ + const {createDigestPool}=await import('../../src/lib/canvas-runtime/digest-pool.ts'),entered=deferred();let first=true;const pool=createDigestPool({concurrency:1,execute:async(bytes,signal)=>{if(first){first=false;entered.resolve();await new Promise((_,reject)=>signal.addEventListener('abort',()=>reject(Error('stopped')),{once:true}));}return importSourceDigest(bytes,signal);}}),a=fixture(),b=fixture();let ea=await a.open({digests:pool}),eb=await b.open({digests:pool}); + try{const aid=ea.imports.startNdjson(file([node('A')]),'canvas');await entered.promise;const bid=eb.imports.startNdjson(file([node('B')]),'canvas');await ea.close();await eb.settle();assert.equal(eb.imports.getJobs().get(bid).status,'completed');assert.equal(eb.getSnapshot().document.elements.size,1);assert.equal(a.errors.length,0);assert.equal(b.errors.length,0);ea=await a.open();await ea.settle();assert.equal(ea.imports.getJobs().get(aid).status,'completed');assert.equal(ea.getSnapshot().document.elements.size,1);}finally{await ea.close();await eb.close();pool.close();} +}); diff --git a/tests/headless/resources.test.mjs b/tests/headless/resources.test.mjs index 6352fad..cd70bca 100644 --- a/tests/headless/resources.test.mjs +++ b/tests/headless/resources.test.mjs @@ -164,3 +164,9 @@ test('known successful upload coordination recovers stored identity and releases assert.equal(f.uploads.length,1);assert.equal(f.writes.length,1);assert.equal(engine.getSnapshot().history,history);assert.equal(engine.getSnapshot().runtime.resources.get(resource.id).stored.id,resource.id);assert.equal(engine.getSnapshot().runtime.ledger.requests.get(request.id).body,request.body); assert.equal(await f.files.get(resource.id).text(),'image bytes');assert.equal(decodeRecovery(f.image(),registry).runtime.resources.get(resource.id).stored.id,resource.id);await engine.close(); }); + + +test('digest worker failure remains distinct from missing bytes and explicit retry resumes the original resource',async()=>{ + const {importSourceDigest}=await import('../../src/lib/canvas-runtime/import-integrity.ts'),f=fixture(),errors=[];let calls=0;const engine=await f.open({digests:{async digest(bytes,signal){if(++calls===1)throw Object.assign(Error('worker failed'),{code:'digest_worker_failed'});return importSourceDigest(bytes,signal);}},onError:error=>errors.push(error)}); + try{const ref=addImage(engine);await engine.settle();const resource=engine.getSnapshot().runtime.resources.get(ref.id);assert.deepEqual(engine.getSnapshot().runtime.ledger.operations.get(resource.operationId).blockers,['digest_worker_failed']);assert.equal(f.files.size,1);assert.equal(f.uploads.length,0);await new Promise(resolve=>setTimeout(resolve,10));assert.equal(calls,1,'failed worker retried in a busy loop');await engine.retry();await engine.settle();assert.equal(f.uploads.length,1);assert.equal(engine.getSnapshot().runtime.resources.get(ref.id).stored.id,ref.id);assert.equal(engine.getSnapshot().history.undo.length,1);assert.equal(errors.length,1);}finally{await engine.close();} +}); diff --git a/tests/headless/sha256.test.mjs b/tests/headless/sha256.test.mjs new file mode 100644 index 0000000..a3ccf2e --- /dev/null +++ b/tests/headless/sha256.test.mjs @@ -0,0 +1,20 @@ +import {test} from 'node:test'; +import assert from 'node:assert/strict'; +import {createHash,randomBytes} from 'node:crypto'; +import {createSha256} from '../../src/lib/canvas-runtime/sha256.ts'; +import {importSourceDigest} from '../../src/lib/canvas-runtime/import-integrity.ts'; +const expected=bytes=>createHash('sha256').update(bytes).digest('hex'); +test('incremental SHA-256 matches known vectors, every padding boundary and varied chunk splits',()=>{ + for(const bytes of [Buffer.alloc(0),Buffer.from('abc'),Buffer.from('abcdbcdecdefdefgefghfghighijhijkijkljklmklmnlmnomnopnopq'),Buffer.alloc(1000000,97),...Array.from({length:260},(_,n)=>randomBytes(n))]){ + for(const chunk of [1,7,63,64,65,8192]){const hash=createSha256();for(let offset=0;offsethash.update(new Uint8Array()),/digest_finalized/);} + } +}); +test('source digest reads bounded slices, yields and never materializes the entire Blob',async()=>{ + const data=randomBytes(1024*1024+13),reads=[];let yields=0; + class SlicedBlob extends Blob{arrayBuffer(){throw Error('whole Blob read');}slice(start,end){reads.push(end-start);return super.slice(start,end);}} + assert.equal(await importSourceDigest(new SlicedBlob([data]),undefined,{chunkBytes:65536,yieldTask:async()=>{yields++;}}),expected(data));assert.equal(reads.length,17);assert.ok(reads.every(size=>size<=65536));assert.equal(yields,16); +}); +test('source digest cancellation fences a late chunk and a scheduler yield',async()=>{ + let release,entered;const gate=new Promise(resolve=>release=resolve),ready=new Promise(resolve=>entered=resolve),controller=new AbortController();class Chunk extends Blob{async arrayBuffer(){entered();await gate;return super.arrayBuffer();}}class Source extends Blob{slice(...args){return new Chunk([super.slice(...args)]);}} + const pending=importSourceDigest(new Source(['abc']),controller.signal);await ready;controller.abort();release();await assert.rejects(pending,/import_aborted/);const second=new AbortController();await assert.rejects(importSourceDigest(new Blob(['abcdef']),second.signal,{chunkBytes:2,yieldTask:async()=>second.abort()}),/import_aborted/); +});