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

Bound chunked source hashing with workspace worker pools

Browse Source
This commit is contained in:
Codex Bot
2026-11-08 20:44:29 +08:00
parent 4035956607
commit 9769f80cf5
15 changed files with 190 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/canvas-runtime/engine.ts
+13 -9
View File
@@ -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<Id, number>;
storage: StoragePort; persistence: PersistencePort; assets?: AssetPort; runs?: RunPort; createId?: () => Id;
storage: StoragePort; digests?:Pick<DigestPort,'digest'>; 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.