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

Resolve default image sizes atomically with creation history and recovery

Browse Source
This commit is contained in:
Codex Bot
2026-11-09 00:25:18 +08:00
parent fa7dad44e7
commit 5919eb0453
20 changed files with 456 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/canvas-runtime/engine.ts
+71 -10
View File
@@ -1,5 +1,7 @@
import {validImageDimensions,type ImageDecoderPort} from './image-decoder.ts';
import {unitScale} from '../canvas-core/units.ts';
import { validateCreationSizeEffect } from '../canvas-core/creation-size-plan.ts';
import { captureDefaultImageSizing, prepareDefaultImageSizing } from './default-image-size.ts';
import {orphanCoverage,prepareOrphanFiles,abortableFileRead,type OrphanFile} from './orphan-files.ts';
import {importFiles,terminalImport,markImportFiles,prepareImportFileRetention,type FileRetirement} from './import-file-retention.ts';
import {resourceRetentionPins,prepareResourceRetention} from './resource-retention.ts';
@@ -75,8 +77,10 @@ export async function openProjectEngine(options: EngineOptions) {
for(const [id,entry]of importFiles(recovered?.runtime.imports))if(entry.file.retirement?.purgedAt!==undefined)durablyPurged.add(id);
let collection:Promise<CollectionReport>|undefined;
const imageTasks=new Map<Id,Promise<void>>();
let sizingTask: Promise<void> | undefined;
const fileWrites = new Map<Id, Blob>(), digestTasks = new Map<Id, Promise<void>>();
let staged: readonly ResourceRecord[] = [];
const stagedDefaultSizes = new Set<Id>();
let applying: RunApplyInput | undefined;
let importing: { job: ImportJob; expectedGeneration: number } | undefined;
const preparationControllers=new Map<Id,AbortController>();
@@ -91,7 +95,8 @@ export async function openProjectEngine(options: EngineOptions) {
traversalPolicy(entry, direction) { return closing || closed || recovery ? { enabled: false, code: closing||closed?'engine_closing':'recovery_in_progress', targets: [] } : options.traversalPolicy?.(entry, direction) ?? { enabled: true }; },
onListenerError: report,
prepareRuntime:(previous,next)=>{
const runtime=prepareResourceRetention(next,options.registry,displayPins,collectionReservations,admissionPins(next));
const sizing=prepareDefaultImageSizing(previous,next);
const runtime=prepareResourceRetention({...next,runtime:sizing},options.registry,displayPins,collectionReservations,admissionPins(next));
const changed=previous.document!==next.document||previous.history!==next.history||previous.runtime.ledger!==runtime.ledger||previous.runtime.resourcePins!==runtime.resourcePins;
const imports=prepareImportFileRetention(previous.runtime,runtime,()=>resourceRetentionPins({...next,runtime},options.registry,displayPins,admissionPins({...next,runtime})),collectionReservations,changed);
return prepareOrphanFiles(previous.runtime,imports,()=>orphanCoverage({...next,runtime:imports},options.registry,displayPins,admissionPins({...next,runtime:imports})),collectionReservations,changed);
@@ -104,12 +109,20 @@ export async function openProjectEngine(options: EngineOptions) {
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);
resources.set(resource.id, stagedDefaultSizes.has(resource.id) ? captureDefaultImageSizing(resource, next, commit, options.registry) : 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)!, footprint = operationFootprint(intent, next.document);
if (intent.historyEffect) {
const effect = intent.historyEffect;
for (const [resourceId, resource] of resources) if (resource.sizing?.version === 1) {
const grants = resource.sizing.grants.map(grant => grant.status === 'pending' && grant.entryId === effect.entryId && grant.target.id === effect.target.id && grant.target.incarnation === effect.target.incarnation
? freeze({ ...grant, status: 'applied' as const, code: undefined }) : grant);
if (grants.some((grant, index) => grant !== resource.sizing!.grants[index])) resources.set(resourceId, freeze({ ...resource, sizing: { version: 1 as const, grants } }));
}
}
const dependencies = new Set(intent.dependencies);
for (const record of ledger.getSnapshot().operations.values()) if (record.phase !== 'resolved' && (record.intent.operationType === 'run.start' || ['document.change','run.apply'].includes(record.intent.operationType) && conflictsWith(operationFootprint(record.intent, next.document), footprint))) dependencies.add(record.intent.id);
for (const id of referencedResources(intent.changes, options.registry)) {
@@ -371,7 +384,7 @@ 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),
participants: intent => { if(intent.historyEffect)validateCreationSizeEffect(intent.historyEffect);return operationCanvases(intent, editor.getSnapshot().document); },
compile(intent, id) {
if (intent.operationType === 'document.change') return options.persistence.compile(intent, id, editor.getSnapshot());
if (['run.start','run.cancel','run.apply'].includes(intent.operationType)) { if (!options.runs) throw fault('run_interpreter_unavailable');return options.runs.compile(intent,id,editor.getSnapshot()); }
@@ -444,6 +457,7 @@ export async function openProjectEngine(options: EngineOptions) {
poll:(run,signal)=>options.runs ? network.run(signal,()=>options.runs!.poll(run,signal)) : Promise.reject(fault('run_interpreter_unavailable')),onError:report});
const trackedRuns=new Set(initial.runs.keys());
const sessions = createSessions(editor);
const unsubscribeSizing = sessions.subscribe(() => { if (!sessions.getSnapshot().gesture) queueDefaultSizes(); });
const unsubscribe = editor.subscribe((commit, snapshot) => {
pending.set(snapshot.rootRevision, snapshot);
const addedRuns=[...snapshot.runtime.runs.keys()].filter(id=>!trackedRuns.has(id));for(const id of addedRuns)trackedRuns.add(id);
@@ -479,6 +493,7 @@ export async function openProjectEngine(options: EngineOptions) {
}
}
function prepareImages(){
queueDefaultSizes();
if(closing||closed||recovery||!options.images)return;
for(const resource of editor.getSnapshot().runtime.resources.values()){
if(imageTasks.size>=2)break;
@@ -494,6 +509,52 @@ export async function openProjectEngine(options: EngineOptions) {
})().catch(error=>{if(closing||closed||recovery)return;const code=(error as{code?:string}).code??'image_decode_failed';editor.updateRuntime(previous=>{const current=previous.resources.get(resource.id);if(!current||current.operationId!==resource.operationId||current.image!==resource.image)return previous;const resources=new Map(previous.resources);resources.set(resource.id,freeze({...current,image:{version:1 as const,status:'failed' as const,code}}));return Object.freeze({...previous,resources:readMap(resources)});});}).finally(()=>{imageTasks.delete(resource.id);prepareImages();});imageTasks.set(resource.id,task);
}
}
function queueDefaultSizes() {
if (sizingTask || closing || closed || recovery || ![...editor.getSnapshot().runtime.resources.values()].some(resource => resource.sizing?.version === 1 && resource.sizing.grants.some(grant => grant.status === 'pending'))) return;
// A scheduling yield, not a logical retry deadline: avoid a self-extending
// microtask chain when one resource supplies many newly created elements.
sizingTask = new Promise(resolve => { setTimeout(() => { sizingTask = undefined; try { prepareDefaultSizes(); } catch (error) { report(error); } finally { resolve(); } }, 0); });
}
function prepareDefaultSizes() {
if (closing || closed || recovery || sessions.getSnapshot().gesture) return;
const root = editor.getSnapshot();
for (const [id, resource] of root.runtime.resources) {
if (resource.sizing?.version !== 1 || resource.image && resource.image.version !== 1) continue;
const size = resource.image?.version === 1 && resource.image.status === 'ready' ? resource.image : !options.images ? resource.stored : undefined;
if (!size || !validImageDimensions(size)) continue;
for (const grant of resource.sizing.grants) {
if (grant.status !== 'pending') continue;
const element = root.document.elements.get(grant.target.id);
if (element?.structure !== 'node') continue;
const units = unitScale(element.data.presentation), min = units, max = 100000 * units;
const scale = Math.min(element.geometry.width / size.width, max / size.width, max / size.height), grow = Math.max(scale, min / size.width, min / size.height);
const width = size.width * grow, height = size.height * grow;
const result = !Number.isFinite(width) || !Number.isFinite(height) || width > max || height > max ? reject(fault('image_fit_constraint'))
: editor.resolveCreationSize({ version: 1, kind: 'creation-size', entryId: grant.entryId, target: grant.target, size: { width: width / units, height: height / units } }, root.rootRevision);
if (result.status === 'rejected' || !result.operationIds.length) {
editor.updateRuntime(previous => {
const current = previous.resources.get(id);
if (current?.sizing?.version !== 1) return previous;
const grants = current.sizing.grants.map(value => value !== grant ? value : freeze({ ...value, status: result.status === 'rejected' ? 'blocked' as const : 'applied' as const, code: result.status === 'rejected' ? result.code : undefined }));
const resources = new Map(previous.resources); resources.set(id, freeze({ ...current, sizing: { version: 1 as const, grants } }));
return Object.freeze({ ...previous, resources: readMap(resources) });
});
if (result.status === 'rejected') report(fault(result.code, result.targets));
}
return; // One publication per pump; its notification schedules further grants.
}
}
}
function retryDefaultSizes() {
editor.updateRuntime(previous => {
const resources = new Map(previous.resources); let changed = false;
for (const [id, resource] of resources) if (resource.sizing?.version === 1 && resource.sizing.grants.some(grant => grant.status === 'blocked')) {
resources.set(id, freeze({ ...resource, sizing: { version: 1 as const, grants: resource.sizing.grants.map(grant => grant.status === 'blocked' ? freeze({ ...grant, status: 'pending' as const, code: undefined }) : grant) } })); changed = true;
}
return changed ? Object.freeze({ ...previous, resources: readMap(resources) }) : previous;
});
queueDefaultSizes();
}
function queueRunAction(operationType:'run.cancel'|'run.apply',action:RunCancelInput|RunApplyInput):Result {
const root=editor.getSnapshot();let sequence=1;for(const record of root.runtime.ledger.operations.values())sequence=Math.max(sequence,record.intent.sequence+1);
const id=createId(),intent={id,operationType,action,operationSchemaVersion:1,sequence,origin:'user',canvasIds:operationType==='run.apply'?[(action as RunApplyInput).targetCanvasId]:[],changes:[],dependencies:operationType==='run.cancel'?[]:[...root.runtime.ledger.operations.values()].filter(record=>record.phase!=='resolved').map(record=>record.intent.id)} as OperationIntent;
@@ -757,17 +818,17 @@ export async function openProjectEngine(options: EngineOptions) {
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');
if (!(file.bytes instanceof Blob) || !file.bytes.size || file.bytes.size > maxResourceBytes || !file.bytes.type.startsWith('image/') || typeof file.name !== 'string' || !file.name || file.defaultSize !== undefined && typeof file.defaultSize !== 'boolean') throw fault('invalid_staged_file');
return freeze({ id: createId(), operationId: createId(), name: file.name, mimeType: file.bytes.type, byteLength: file.bytes.size,...(options.images?{image:{version:1 as const,status:'pending' as const}}:{}) });
});
if (new Set(resources.flatMap(resource => [resource.id, resource.operationId])).size !== resources.length * 2 || resources.some(resource => editor.getSnapshot().runtime.resources.has(resource.id) || editor.getSnapshot().runtime.orphanFiles?.has(resource.id) || fileWrites.has(resource.id) || importFiles(editor.getSnapshot().runtime.imports).has(resource.id))) throw fault('resource_identity_conflict');
staged = resources; resources.forEach((resource, index) => fileWrites.set(resource.id, files[index].bytes));
staged = resources; resources.forEach((resource, index) => { fileWrites.set(resource.id, files[index].bytes);if(files[index].defaultSize)stagedDefaultSizes.add(resource.id); });
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); }
} finally { staged = []; stagedDefaultSizes.clear();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');
@@ -810,12 +871,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();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();}if(options.images){editor.updateRuntime(previous=>{const resources=new Map(previous.resources);let changed=false;for(const [id,resource]of resources)if(resource.image?.version===1&&resource.image.status==='failed'){resources.set(id,freeze({...resource,image:{version:1 as const,status:'pending' as const}}));changed=true;}return changed?Object.freeze({...previous,resources:readMap(resources)}):previous;});prepareImages();}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()]);while(imageTasks.size)await Promise.all([...imageTasks.values()]); await scheduler.settle(); await runTracker.settle(); await persist(); },
retry: async () => { if(recovery)await recovery.cancel();assertWritable();retryDefaultSizes();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();}if(options.images){editor.updateRuntime(previous=>{const resources=new Map(previous.resources);let changed=false;for(const [id,resource]of resources)if(resource.image?.version===1&&resource.image.status==='failed'){resources.set(id,freeze({...resource,image:{version:1 as const,status:'pending' as const}}));changed=true;}return changed?Object.freeze({...previous,resources:readMap(resources)}):previous;});prepareImages();}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()]);while(imageTasks.size)await Promise.all([...imageTasks.values()]);while(sizingTask)await sizingTask; await scheduler.settle(); await runTracker.settle(); await persist(); },
async close() {
if (closed) return;
closing = true;orphanController?.abort();await Promise.allSettled([collection,sourceCollection,orphanCollection].filter(Boolean));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()]);while(imageTasks.size)await Promise.allSettled([...imageTasks.values()]);
closing = true;unsubscribeSizing();orphanController?.abort();await Promise.allSettled([collection,sourceCollection,orphanCollection].filter(Boolean));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()]);while(imageTasks.size)await Promise.allSettled([...imageTasks.values()]);while(sizingTask)await sizingTask;
recovery?.controller.abort();recovery?.release();
for (const release of [...displays]) release();
// Failure keeps the leader/recovery material available for an explicit retry of close.