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

Stream native imports before full source verification with durable send seals

Browse Source
This commit is contained in:
Codex Bot
2026-11-09 23:24:53 +08:00
parent f11ea36b48
commit 0ce3abbe04
17 changed files with 230 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/components/canvas/headless-imports.tsx
+3 -1
View File
@@ -7,7 +7,7 @@ import { IconButton } from "@/components/ui/icon-button";
import { errorText } from "@/lib/api";
import styles from "./canvas.module.css";
const reason = (code: string) => ({ import_asset_authorization_failed: "资源访问授权失败，此记录未加入", import_asset_capacity: "本组资源过大，无法一次准备", asset_depot_unavailable: "资源仓库暂不可用", import_resource_authorization_required: "资源需要重新授权，此记录尚未加入", invalid_document_import: "导出文档结构不合法", invalid_json_array: "JSON 记录数组格式错误", incomplete_json_array: "JSON 数组尾部不完整", json_nesting_limit: "JSON 嵌套层数过多", invalid_json: "记录格式错误", invalid_json_value: "记录包含不支持的数据", invalid_utf8: "文件编码不是 UTF-8", record_too_large: "单条记录过大", unresolved_dependency: "引用的记录未加入", invalid_native_import_record: "不是画布记录", import_component_unavailable: "元素类型暂不支持", invalid_geometry: "元素尺寸不合法", invalid_endpoint: "连线端点已不存在", duplicate_import_source_key: "记录标识重复", import_source_unavailable: "导入源文件不可用", import_source_changed: "源文件内容与保存时不一致，任务已暂停", import_source_integrity_unavailable: "旧任务缺少源内容校验，无法安全继续；已加入内容保留", import_source_integrity_missing: "源校验记录缺失，任务已暂停", import_source_integrity_interpreter_unavailable: "源校验版本暂不支持", import_target_missing: "目标分区已不存在", import_target_changed: "目标分区已改变", import_dependency_changed: "引用对象已删除或恢复为新版本", dependency_capacity: "等待引用过多，请调整文件顺序", copy_external_changed: "原图引用已删除、移动或恢复为新版本", invalid_duplicate_source: "复制源数据不可用", invalid_copy_snapshot: "复制快照不可用", invalid_copy_preparation: "复制准备记录不可用", copy_interpreter_changed: "复制规则已变化，需要重新复制", invalid_import_source: "文件过大或格式不支持" } as Record<string,string>)[code] ?? "这条记录暂时无法加入";
const reason = (code: string) => ({ import_asset_authorization_failed: "资源访问授权失败，此记录未加入", import_asset_capacity: "本组资源过大，无法一次准备", asset_depot_unavailable: "资源仓库暂不可用", import_resource_authorization_required: "资源需要重新授权，此记录尚未加入", invalid_document_import: "导出文档结构不合法", invalid_json_array: "JSON 记录数组格式错误", incomplete_json_array: "JSON 数组尾部不完整", json_nesting_limit: "JSON 嵌套层数过多", invalid_json: "记录格式错误", invalid_json_value: "记录包含不支持的数据", invalid_utf8: "文件编码不是 UTF-8", record_too_large: "单条记录过大", unresolved_dependency: "引用的记录未加入", invalid_native_import_record: "不是画布记录", import_component_unavailable: "元素类型暂不支持", invalid_geometry: "元素尺寸不合法", invalid_endpoint: "连线端点已不存在", duplicate_import_source_key: "记录标识重复", import_source_read_failed: "暂时无法读取源文件，可重试；已加入内容保留", import_source_verification_failed: "源文件校验失败，可重试；已加入内容保留", invalid_import_read_proof: "源文件读取证据不一致，任务已暂停", import_source_unavailable: "导入源文件不可用", import_source_changed: "源文件内容与保存时不一致，任务已暂停", import_source_integrity_unavailable: "旧任务缺少源内容校验，无法安全继续；已加入内容保留", import_source_integrity_missing: "源校验记录缺失，任务已暂停", import_source_integrity_interpreter_unavailable: "源校验版本暂不支持", import_target_missing: "目标分区已不存在", import_target_changed: "目标分区已改变", import_dependency_changed: "引用对象已删除或恢复为新版本", dependency_capacity: "等待引用过多，请调整文件顺序", copy_external_changed: "原图引用已删除、移动或恢复为新版本", invalid_duplicate_source: "复制源数据不可用", invalid_copy_snapshot: "复制快照不可用", invalid_copy_preparation: "复制准备记录不可用", copy_interpreter_changed: "复制规则已变化，需要重新复制", invalid_import_source: "文件过大或格式不支持" } as Record<string,string>)[code] ?? "这条记录暂时无法加入";
export function HeadlessImports({ engine, canvasId }: { engine: ProjectEngine; canvasId: string }) {
const subscribe = useCallback((notify: () => void) => engine.editor.subscribe(notify), [engine]);
@@ -26,9 +26,11 @@ export function HeadlessImports({ engine, canvasId }: { engine: ProjectEngine; c
{jobs.map((job, index) => <section className={styles.importJob} key={job.id} aria-label={`导入任务 ${index + 1}`}>
<strong>{job.source?.name || `导入任务 ${index + 1}`}</strong>
<p role="status">{["running","preparing"].includes(job.status) && job.source?.integrityVersion === 1 && !job.source.sha256 ? "正在校验源文件" : job.status === "preparing" ? "正在准备复制" : job.status === "running" ? "正在加入" : job.status === "paused" ? "已暂停" : job.status === "cancelled" ? "已停止" : job.status === "partial" ? "部分加入" : "已加入画布"} · {job.committedSourceKeys.length} 项{job.pending.length > 0 && `，${job.pending.length} 项等待`}</p>
{!(job.status==='cancelled'&&!job.committedSourceKeys.length)&&job.source?.integrityVersion===2&&!job.source.sha256&&!job.source.integrityError&&<small>内容可继续编辑，源文件校验完成后同步。</small>}
{job.source?.retirement?.purgedAt !== undefined && <small>本地源文件已清理；画布内容和撤销记录保留。</small>}
{job.error && <p>{reason(job.error)}</p>}
<div className={styles.importActions}>
{job.source?.integrityError&&<button onClick={()=>control(engine.retry())}>重试源文件校验</button>}
{["running","preparing"].includes(job.status) && <button onClick={() => control(engine.imports.pause(job.id))}>暂停导入</button>}
{job.status === "paused" && <button onClick={() => control(engine.imports.resume(job.id))}>继续导入</button>}
{["running","preparing","paused"].includes(job.status) && <button onClick={() => control(engine.imports.cancel(job.id))}>停止导入</button>}
src/lib/canvas-runtime/engine.ts
+69 -23
View File
@@ -1,3 +1,6 @@
import {createImportReadProof,IMPORT_PROOF_BLOCK_BYTES} from './import-read-proof.ts';
import {isImportByteFailure} from './import-byte-reader.ts';
import {assertImportEvidence,importSeals,importsMaySend} from './import-verification.ts';
import {createAssetDepot,sharedAssetRef,type AssetDepot,type SharedAsset} from './asset-depot.ts';
import {prepareDepotCut} from './depot-cut.ts';
import {loadingAssetURLs} from './loading-assets.ts';
@@ -21,7 +24,7 @@ import { encodeDuplicateSource, prepareDuplicateImport, assertDuplicateExternal
import {openDocumentImportSource} from './import-document.ts';
import {openStreamingDocumentImportSource} from './import-document-stream.ts';
import {createJsonArrayImportSource} from './import-json-array.ts';
import { createImportJob, createImportRunner, createNdjsonImportSource, type ImportJob } from './imports.ts';
import { createImportJob, importNeedsSeal, createImportRunner, createNdjsonImportSource, type ImportJob } from './imports.ts';
import { assertNativeImportDependencies, assertNativeImportTarget, prepareNativeImport } from './import-native.ts';
import { elisionBatchLimit, findElision } from './elision.ts';
import type { Document, Id, Registry, Result, RunInput, RunApplyInput, RunCancelInput, Target, Point, Json, OperationIntent, Data } from '../canvas-core/types.ts';
@@ -107,6 +110,9 @@ export async function openProjectEngine(options: EngineOptions) {
const stagedDefaultSizes = new Set<Id>();
let applying: RunApplyInput | undefined;
let importing: { job: ImportJob; expectedGeneration: number } | undefined;
let integrityOwner:Id|undefined;
const proofReaders=new Map<Id,{fileId:Id;bytes:Blob;reader:ReturnType<typeof createImportReadProof>}>();
const verificationTasks=new Map<Id,{controller:AbortController;promise:Promise<void>}>();
const preparationControllers=new Map<Id,AbortController>();
const copyPreparations=new Map<AbortController,Promise<Id>>();
const importRunners=new Map<Id,ReturnType<typeof createImportRunner>>(),importLoads=new Map<Id,Promise<void>>(),importHolds=new Map<symbol,Map<Id,()=>void>>();
@@ -116,6 +122,7 @@ export async function openProjectEngine(options: EngineOptions) {
function assetURLs(){const root=editor.getSnapshot(),urls=loadingAssetURLs(root.runtime);return staged.length||activeAdoptions().length?new Map([...urls,...staged.map(resource=>[resource.id,options.assets!.sourceURL(resource.id)] as const),...activeAdoptions().map(ref=>[ref.asset.id,ref.asset.url] as const)]):urls;}
const initial: RuntimeState = recovered?.runtime ?? Object.freeze({ imports:readMap(new Map()), recoveries:readMap(new Map()), runs: readMap(new Map()), acknowledged: createDocument(options.document.projectId, [...options.document.canvases.values()], [...options.document.elements.values()], options.document.tombstones), versions: openingVersions, ...(openingLoading?{loading:openingLoading,...(openingLoading.version>=3?{loadingRead:Object.freeze({document:createDocument(options.document.projectId,[...options.document.canvases.values()],[...options.document.elements.values()],options.document.tombstones),versions:openingVersions})}:{})}:{}), ledger: createLedger().getSnapshot(), receipts: readMap(new Map()), conflicts: readMap(new Map()), resources: readMap(new Map()), resourcePins: readMap(new Map()) });
let durableImportSeals=importSeals(initial.imports);
validateLoadingRuntime(initial,options.registry);
for(const [id,binding]of initial.assetBindings??[])if(!depot||!options.assets?.authority||binding.authority!==options.assets.authority.scope||binding.asset.url!==options.assets.sourceURL(id))throw fault('asset_binding_interpreter_unavailable',[id]);
if(initial.loading?.version===6&&options.assets)for(const resource of initial.resources.values())if(resource.loadingAdmission?.sourceURL!==options.assets.sourceURL(resource.id))throw fault('asset_source_changed',[resource.id]);
@@ -125,6 +132,7 @@ export async function openProjectEngine(options: EngineOptions) {
onListenerError: report,
prepareRuntime:(previous,next)=>{
if(!leader.active)throw fault('storage_leader_lost');
assertImportEvidence(previous.runtime.imports,next.runtime.imports,integrityOwner??importing?.job.id);
if(!equal([...(previous.runtime.assetBindings??[])],[...(next.runtime.assetBindings??[])])){
const before=new Map(previous.runtime.assetBindings??[]),after=new Map(next.runtime.assetBindings??[]);
const permitted=new Map(activeAdoptions().map(ref=>[ref.asset.id,ref]));if(after.size<before.size||[...before].some(([id,binding])=>!equal(after.get(id),binding))||[...after].some(([id,binding])=>!before.has(id)&&(!permitted.has(id)||!equal(sharedAssetRef(binding),sharedAssetRef(permitted.get(id)!)))))throw fault('asset_bindings_owned');
@@ -202,7 +210,7 @@ export async function openProjectEngine(options: EngineOptions) {
const depot=options.assets?.authority&&options.storage.assetDepot?await prepareDepotCut({snapshot,registry:options.registry,storage:options.storage,leader,authority:options.assets.authority.scope,files:fileWrites,adoptedFiles,displays:displayPins,temporary:admissionPins(snapshot)}):undefined;
const receipt = await options.storage.commit(leader, { image: encodeRecovery(snapshot), files,deleteFiles,...(depot?{depot}:{}) });
if (receipt.durableRootRevision !== revision) throw fault('invalid_durable_receipt');
durableRootRevision = revision; pending.delete(revision);notifyDurability();
durableRootRevision = revision;durableImportSeals=importSeals(snapshot.runtime.imports); pending.delete(revision);notifyDurability();
for(const id of deleteFiles){durablyPurged.add(id);collectionReservations.delete(id);}
for (const file of files) fileWrites.delete(file.id);
for(const [id,entry]of adoptedFiles)if(entry.rootRevision<=snapshot.rootRevision&&depot?.writes.some(write=>write.ref.asset.id===id&&write.bytes===entry.bytes))adoptedFiles.delete(id);
@@ -434,7 +442,7 @@ export async function openProjectEngine(options: EngineOptions) {
report(error); return { blocked: true };
}
}
const scheduler = createScheduler({ admit:intent=>{const root=editor.getSnapshot();if(root.runtime.loading&&root.runtime.loading.version<3)return true;if(closing||closed||loadingCompletionRevision!==undefined&&durableRootRevision<loadingCompletionRevision)return false;return (root.runtime.loading?.version===6&&intent.operationType==='asset.upload'||loadingCanSend(root.runtime,operationCanvases(intent,root.document),root.document))&&(!root.runtime.loading||durableRootRevision>=root.rootRevision);},getLedger: () => editor.getSnapshot().runtime.ledger, mutate, persist,
const scheduler = createScheduler({ admit:intent=>{const root=editor.getSnapshot();if(!importsMaySend(root.runtime.imports,durableImportSeals))return false;if(root.runtime.loading&&root.runtime.loading.version<3)return true;if(closing||closed||loadingCompletionRevision!==undefined&&durableRootRevision<loadingCompletionRevision)return false;return (root.runtime.loading?.version===6&&intent.operationType==='asset.upload'||loadingCanSend(root.runtime,operationCanvases(intent,root.document),root.document))&&(!root.runtime.loading||durableRootRevision>=root.rootRevision);},getLedger: () => editor.getSnapshot().runtime.ledger, mutate, persist,
createId: options.createId ?? (() => crypto.randomUUID()),
participants: intent => { if(intent.historyEffect)validateCreationSizeEffect(intent.historyEffect);return operationCanvases(intent, editor.getSnapshot().document); },
compile(intent, id) {
@@ -631,10 +639,41 @@ export async function openProjectEngine(options: EngineOptions) {
function getImport(id:Id){const job=editor.getSnapshot().runtime.imports?.get(id);if(!job)throw fault('import_missing',[id]);return job;}
function publishImport(job:ImportJob,expectedGeneration:number){
assertImportWritable();const permission=options.policy?.({type:'changes.apply',changes:[]},editor.getSnapshot().document);if(permission&&!permission.enabled)throw fault(permission.code,permission.targets);
const result=editor.updateRuntime(previous=>{const imports=new Map(previous.imports??[]);if(imports.get(job.id)?.generation!==expectedGeneration)throw fault('stale_import_job');imports.set(job.id,job.version===3?job:freeze({...job,version:3 as const}));return Object.freeze({...previous,imports:readMap(imports)});});if(result.status==='rejected')throw fault(result.code,result.targets);
integrityOwner=job.id;try{const result=editor.updateRuntime(previous=>{const imports=new Map(previous.imports??[]);if(imports.get(job.id)?.generation!==expectedGeneration)throw fault('stale_import_job');imports.set(job.id,job.version===3?job:freeze({...job,version:3 as const}));return Object.freeze({...previous,imports:readMap(imports)});});if(result.status==='rejected')throw fault(result.code,result.targets);}finally{integrityOwner=undefined;}
}
function checkpointImportSource(job:ImportJob):ImportJob{
if(job.source?.integrityVersion!==2)return job;
const context=proofReaders.get(job.id),current=editor.getSnapshot().runtime.imports?.get(job.id)?.source;
if(!context||context.fileId!==job.source.fileId)throw fault('import_source_unavailable');
const {sha256:_,integrityError:__,...source}=job.source;
return {...job,source:{...source,readProof:context.reader.getProof(),...(current?.sha256?{sha256:current.sha256}:{}),...(current?.integrityError?{integrityError:current.integrityError}:{})}};
}
function failImportVerification(job:ImportJob,error:unknown){
const code=(error as {code?:string}).code??'import_source_verification_failed';
publishImport(freeze({...job,generation:job.generation+1,status:['cancelled','partial'].includes(job.status)?job.status:'paused',source:{...job.source!,integrityError:code},error:code}),job.generation);
const release=importRunners.get(job.id)?.hold();release?.();report(error);
}
function verifyImportSource(id:Id,context:{fileId:Id;bytes:Blob;reader:ReturnType<typeof createImportReadProof>}):Promise<void>{
const active=verificationTasks.get(id);if(active)return active.controller.signal.aborted?active.promise.then(()=>{if(!closing&&!closed&&!importHolds.size&&proofReaders.get(id)===context&&sourceNeedsSeal(getImport(id)))return verifyImportSource(id,context);}):active.promise;
const admitted=getImport(id),controller=new AbortController(),signal=AbortSignal.any([controller.signal,digestLifecycle.signal]);
const current=()=>{signal.throwIfAborted();if(closing||closed||importHolds.size)throw fault('import_aborted');const job=getImport(id);if(job.sourceFingerprint!==admitted.sourceFingerprint||job.source?.fileId!==context.fileId||proofReaders.get(id)!==context)throw fault('import_identity_changed');return job;};
const promise=(async()=>{
let sha256:string;
try{await context.reader.verify(signal);sha256=await abortableFileRead(sourceDigest(context.bytes,signal),signal);const job=current();if(job.source?.sha256&&job.source.sha256!==sha256)throw fault('import_source_changed');}
catch(error){if(signal.aborted||closing||closed||importHolds.size)return;failImportVerification(current(),error);try{await persist();}catch{}return;}
const job=current();if(job.status==='cancelled'&&!job.committedSourceKeys.length)return;const{integrityError:oldError,...source}=job.source!,next={...job,source:{...source,sha256,readProof:context.reader.getProof()}};if(oldError&&next.error===oldError)delete next.error;
publishImport(freeze(next),job.generation);await persist();
})().catch(error=>{if(!signal.aborted&&!closing&&!closed)report(error);}).finally(()=>{if(verificationTasks.get(id)?.controller===controller)verificationTasks.delete(id);});
verificationTasks.set(id,{controller,promise});return promise;
}
async function resetImportReader(id:Id){
const active=verificationTasks.get(id);active?.controller.abort();if(active)await active.promise;
const runner=importRunners.get(id);if(runner){await runner.close();importRunners.delete(id);}
proofReaders.get(id)?.reader.close();proofReaders.delete(id);
}
function sourceNeedsSeal(job:ImportJob){return importNeedsSeal(job);}
const deferredImportLoads=new Set<Id>();
function holdImports(){for(const id of importLoads.keys())if(!importRunners.has(id))deferredImportLoads.add(id);const token=Symbol(),releases=new Map<Id,()=>void>();importHolds.set(token,releases);for(const controller of preparationControllers.values())controller.abort();for(const[id,runner]of importRunners)releases.set(id,runner.hold());return()=>{if(!importHolds.delete(token))return;for(const release of releases.values())release();if(!closing&&!closed)for(const job of editor.getSnapshot().runtime.imports?.values()??[])if(['running','preparing'].includes(job.status)&&!importRunners.has(job.id))void loadImport(job.id);};}
function holdImports(){for(const id of importLoads.keys())if(!importRunners.has(id))deferredImportLoads.add(id);const token=Symbol(),releases=new Map<Id,()=>void>();importHolds.set(token,releases);for(const controller of preparationControllers.values())controller.abort();for(const {controller}of verificationTasks.values())controller.abort();for(const[id,runner]of importRunners)releases.set(id,runner.hold());return()=>{if(!importHolds.delete(token))return;for(const release of releases.values())release();if(!closing&&!closed)for(const job of editor.getSnapshot().runtime.imports?.values()??[])if(sourceNeedsSeal(job)||['running','preparing'].includes(job.status)&&!importRunners.has(job.id))void loadImport(job.id);};}
async function materializeCopy(id:Id,bytes:Blob,controller:AbortController){
const admitted=getImport(id),preparation=admitted.preparation;
if(!preparation||admitted.source?.kind!=='duplicate-snapshot')throw fault('invalid_copy_preparation');
@@ -670,16 +709,22 @@ export async function openProjectEngine(options: EngineOptions) {
}
function loadImport(id:Id):Promise<void>{
if(importLoads.has(id))return importLoads.get(id)!;
if(importRunners.has(id)){void importRunners.get(id)!.start();return Promise.resolve();}
if(importRunners.has(id)){const context=proofReaders.get(id);if(context&&sourceNeedsSeal(getImport(id)))void verifyImportSource(id,context);if(getImport(id).status==='running')void importRunners.get(id)!.start();return Promise.resolve();}
deferredImportLoads.delete(id);
let expectedFingerprint=getImport(id).sourceFingerprint;
const preparationController=new AbortController();if(['native-document','duplicate-snapshot','duplicate-plan'].includes(getImport(id).source?.kind??''))preparationControllers.set(id,preparationController);
const preparationController=new AbortController();if(getImport(id).source?.integrityVersion===2||['native-document','duplicate-snapshot','duplicate-plan'].includes(getImport(id).source?.kind??''))preparationControllers.set(id,preparationController);
const task=(async()=>{
let job=getImport(id),source=job.source;if(!source||!['native-document','native-json-array','native-ndjson','duplicate-ndjson','duplicate-snapshot','duplicate-plan'].includes(source.kind)||(source.kind==='native-document'?![4,5].includes(job.parserVersion):job.parserVersion!==(source.kind==='native-json-array'?3:source.kind==='duplicate-plan'?2:1)))throw fault('import_interpreter_unavailable');
let bytes=fileWrites.get(source.fileId)??await options.storage.readFile(leader,source.fileId);
let bytes=fileWrites.get(source.fileId)??await abortableFileRead(options.storage.readFile(leader,source.fileId),preparationController.signal);
if(closing||closed)return;
const latest=getImport(id);if(latest.sourceFingerprint!==job.sourceFingerprint||latest.source?.fileId!==source.fileId||['completed','partial','cancelled'].includes(latest.status))return;
const latest=getImport(id);if(latest.sourceFingerprint!==job.sourceFingerprint||latest.source?.fileId!==source.fileId||['completed','partial','cancelled'].includes(latest.status)&&(source.integrityVersion!==2||!sourceNeedsSeal(latest)))return;
if(!bytes)throw fault('import_source_unavailable');if(bytes.size!==source.byteLength)throw fault('import_source_changed');
let readBytes:ReturnType<typeof createImportReadProof>['port']|undefined;
if(source.integrityVersion===2){
let context=proofReaders.get(id);if(!context||context.fileId!==source.fileId){context={fileId:source.fileId,bytes,reader:createImportReadProof({bytes,fileId:source.fileId,proof:source.readProof})};proofReaders.set(id,context);}
await context.reader.verify(preparationController.signal);const checking=verifyImportSource(id,context);if(source.sha256)await checking;
job=getImport(id);source=job.source!;if(preparationController.signal.aborted||['completed','partial','cancelled','paused'].includes(job.status)||importHolds.size)return;readBytes=context.reader.port;
}else{
if(source.integrityVersion!==1)throw fault('import_source_integrity_unavailable');
const sha256=await sourceDigest(bytes,preparationController.signal);
if(closing||closed)return;
@@ -687,15 +732,16 @@ export async function openProjectEngine(options: EngineOptions) {
if(importHolds.size){deferredImportLoads.add(id);return;}
if(source.sha256&&source.sha256!==sha256)throw fault('import_source_changed');
if(!source.sha256){publishImport(freeze({...verified,generation:verified.generation+1,source:{...source,sha256}}),verified.generation);job=getImport(id);source=job.source!;}else job=verified;
}
let copyDocument:Document|undefined,indexed:IndexedCopySnapshot|undefined;
if(source.kind==='duplicate-snapshot'){if(preparationController.signal.aborted||importHolds.size||latest.status!=='preparing')return;const materialized=await materializeCopy(id,bytes,preparationController);job=materialized.job;source=job.source!;bytes=materialized.bytes;copyDocument=materialized.document;indexed=materialized.indexed;expectedFingerprint=job.sourceFingerprint;}
if(source.kind==='duplicate-plan'&&source.snapshotVersion===2){indexed??=await openIndexedCopySnapshot(bytes,options.scope.projectId,options.registry,preparationController.signal);copyDocument=indexed.document;}
if(source.kind==='duplicate-plan'&&copyDocument)validateLoadingCopySource(editor.getSnapshot(),job,copyDocument);
const importSource=source.kind==='duplicate-plan'?createDuplicatePlanSource(copyDocument??await decodeCopySnapshot(bytes,options.scope.projectId,options.registry,preparationController.signal),options.registry,source.plan!,job.sourceFingerprint,indexed?.readElement):source.kind==='native-document'?await (job.parserVersion===5?openStreamingDocumentImportSource:openDocumentImportSource)({bytes,fingerprint:job.sourceFingerprint,signal:preparationController.signal}):source.kind==='native-json-array'?createJsonArrayImportSource({bytes,fingerprint:job.sourceFingerprint}):createNdjsonImportSource({bytes,fingerprint:job.sourceFingerprint});
const importSource=source.kind==='duplicate-plan'?createDuplicatePlanSource(copyDocument??await decodeCopySnapshot(bytes,options.scope.projectId,options.registry,preparationController.signal),options.registry,source.plan!,job.sourceFingerprint,indexed?.readElement):source.kind==='native-document'?await (job.parserVersion===5?openStreamingDocumentImportSource:openDocumentImportSource)({bytes,readBytes,fingerprint:job.sourceFingerprint,signal:preparationController.signal}):source.kind==='native-json-array'?createJsonArrayImportSource({bytes,readBytes,fingerprint:job.sourceFingerprint}):createNdjsonImportSource({bytes,readBytes,fingerprint:job.sourceFingerprint});
if(closing||closed||preparationController.signal.aborted||getImport(id).generation!==job.generation)return;
if(importHolds.size){deferredImportLoads.add(id);return;}
const authorizedImportAssets=new Map<Id,{ref:SharedAsset;bytes:Blob}>(),assetPreparationSignals=new WeakSet<AbortSignal>();
const runner=createImportRunner({optimisticLocal:true,awaitDurabilityBudget:stage=>pending.size>=32||stage!=='parsed'&&persistenceFailed?persist():undefined,source:importSource,getJob:()=>getImport(id),createId,
const runner=createImportRunner({optimisticLocal:true,checkpointSource:source.integrityVersion===2?checkpointImportSource:undefined,awaitDurabilityBudget:stage=>pending.size>=32||stage!=='parsed'&&persistenceFailed?persist():undefined,source:importSource,getJob:()=>getImport(id),createId,
publish:publishImport,prepare:(line,job,reserve)=>['duplicate-ndjson','duplicate-plan'].includes(job.source?.kind??'')?prepareDuplicateImport(line,job,reserve,options.registry):prepareNativeImport(line,job,reserve,editor.getSnapshot().document,options.registry,depot&&options.assets?.authority?options.assets.sourceURL:undefined),persist,onError:report,
async beforeCommit(units,job,signal){
authorizedImportAssets.clear();if(job.source?.kind!=='native-document')return{enabled:true};
@@ -716,7 +762,7 @@ export async function openProjectEngine(options: EngineOptions) {
});
importRunners.set(id,runner);for(const releases of importHolds.values())releases.set(id,runner.hold());
if(getImport(id).status==='running')void runner.start();
})().catch(error=>{if(preparationController.signal.aborted)return;if(!closing&&!closed){try{const job=getImport(id);if(job.sourceFingerprint!==expectedFingerprint||['completed','partial','cancelled','paused'].includes(job.status))return;publishImport(freeze({...job,generation:job.generation+1,status:'paused',error:(error as {code?:string}).code??'import_failed'}),job.generation);}catch{/* Recovery/identity fences retain the original job. */}report(error);}}).finally(()=>{preparationControllers.delete(id);importLoads.delete(id);if(!['running','preparing'].includes(getImport(id).status))deferredImportLoads.delete(id);if(!closing&&!closed&&!importHolds.size&&deferredImportLoads.delete(id)&&!importRunners.has(id)&&['running','preparing'].includes(getImport(id).status))queueMicrotask(()=>{if(!closing&&!closed)void loadImport(id);});});
})().catch(error=>{if(preparationController.signal.aborted)return;if(!closing&&!closed){try{const job=getImport(id);if(job.source?.integrityVersion===2&&!importRunners.has(id)&&!verificationTasks.has(id)){proofReaders.get(id)?.reader.close();proofReaders.delete(id);}if(job.sourceFingerprint!==expectedFingerprint)return;if(job.source?.integrityVersion===2&&(isImportByteFailure(error)||(error as {code?:string}).code==='import_source_unavailable')){failImportVerification(job,error);void persist().catch(report);return;}if(['completed','partial','cancelled','paused'].includes(job.status))return;publishImport(freeze({...job,generation:job.generation+1,status:'paused',error:(error as {code?:string}).code??'import_failed'}),job.generation);}catch{/* Recovery/identity fences retain the original job. */}report(error);}}).finally(()=>{preparationControllers.delete(id);importLoads.delete(id);if(!['running','preparing'].includes(getImport(id).status))deferredImportLoads.delete(id);if(!closing&&!closed&&!importHolds.size&&deferredImportLoads.delete(id)&&!importRunners.has(id)&&['running','preparing'].includes(getImport(id).status))queueMicrotask(()=>{if(!closing&&!closed)void loadImport(id);});});
importLoads.set(id,task);return task;
}
function startNativeImport(bytes:Blob,targetCanvasId:Id,format:'ndjson'|'json-array'|'document'):Id{
@@ -724,10 +770,10 @@ export async function openProjectEngine(options: EngineOptions) {
assertImportWritable();const loadingAdmission=loadingImportAdmission(editor.getSnapshot(),targetCanvasId);const permission=options.policy?.({type:'changes.apply',changes:[]},editor.getSnapshot().document);if(permission&&!permission.enabled)throw fault(permission.code,permission.targets);const canvas=editor.getSnapshot().document.canvases.get(targetCanvasId);if(!canvas)throw fault('canvas_not_loaded',[targetCanvasId]);if(!(bytes instanceof Blob)||bytes.size>64*1024*1024)throw fault('invalid_import_source');
const id=createId(),fileId=createId(),current=editor.getSnapshot().runtime;if(!id||!fileId||id===fileId||current.imports?.has(id)||current.resources.has(fileId)||current.orphanFiles?.has(fileId)||fileWrites.has(fileId)||importFiles(current.imports).has(fileId))throw fault('import_identity_conflict');
const entry=canvas.parent?editor.getSnapshot().document.elements.get(canvas.parent.entryElementId)!:undefined;
const job=freeze({...createImportJob({id,projectId:options.scope.projectId,sourceFingerprint:`local-file:${fileId}:${bytes.size}`,parserVersion:format==='document'?5:format==='json-array'?3:1,source:{...(loadingAdmission?{loadingAdmission}:{}),kind:format==='document'?'native-document':format==='json-array'?'native-json-array':'native-ndjson',name:((bytes as Blob&{name?:string}).name||'画布数据').slice(0,255),fileId,byteLength:bytes.size,targetCanvasId,...(entry?{targetEntry:{id:entry.id,incarnation:entry.incarnation}}:{})}}),status:'running' as const});
const job=freeze({...createImportJob({id,projectId:options.scope.projectId,sourceFingerprint:`local-file:${fileId}:${bytes.size}`,parserVersion:format==='document'?5:format==='json-array'?3:1,source:{...(loadingAdmission?{loadingAdmission}:{}),integrityVersion:2,readProof:{version:1,fileId,byteLength:bytes.size,blockBytes:IMPORT_PROOF_BLOCK_BYTES,blocks:[]},kind:format==='document'?'native-document':format==='json-array'?'native-json-array':'native-ndjson',name:((bytes as Blob&{name?:string}).name||'画布数据').slice(0,255),fileId,byteLength:bytes.size,targetCanvasId,...(entry?{targetEntry:{id:entry.id,incarnation:entry.incarnation}}:{})}}),status:'running' as const});
fileWrites.set(fileId,bytes);
const result=editor.updateRuntime(previous=>{const jobs=new Map(previous.imports??[]);jobs.set(id,job);return Object.freeze({...previous,imports:readMap(jobs)});});
if(result.status==='rejected'){fileWrites.delete(fileId);throw fault(result.code,result.targets);}void loadImport(id);return id;
integrityOwner=id;try{const result=editor.updateRuntime(previous=>{const jobs=new Map(previous.imports??[]);jobs.set(id,job);return Object.freeze({...previous,imports:readMap(jobs)});});
if(result.status==='rejected'){fileWrites.delete(fileId);throw fault(result.code,result.targets);}void loadImport(id);return id;}finally{integrityOwner=undefined;}
}
const imports=Object.freeze({
getJobs:()=>readMap(editor.getSnapshot().runtime.imports??new Map()),
@@ -761,12 +807,12 @@ export async function openProjectEngine(options: EngineOptions) {
})();
copyPreparations.set(controller,task);void task.finally(()=>{copyPreparations.delete(controller);signal?.removeEventListener('abort',abort);}).catch(()=>{});return task;
},
async pause(id:Id){assertImportWritable();const runner=importRunners.get(id);if(runner)await runner.pause();else{const job=getImport(id);if(['running','preparing'].includes(job.status)){preparationControllers.get(id)?.abort();publishImport(freeze({...job,generation:job.generation+1,status:'paused'}),job.generation);await persist();}}},
async resume(id:Id){assertImportWritable();const job=getImport(id);if(['completed','partial','cancelled'].includes(job.status))return;if(['running','preparing'].includes(job.status)){await loadImport(id);return;}await importLoads.get(id);await importRunners.get(id)?.settle();await persist();await scheduler.resume();assertImportWritable();const current=getImport(id);if(current.status!=='paused')return;const next={...current,generation:current.generation+1,status:current.source?.kind==='duplicate-snapshot'?'preparing' as const:'running' as const};delete next.error;publishImport(freeze(next),current.generation);await loadImport(id);},
async cancel(id:Id){assertImportWritable();const runner=importRunners.get(id);if(runner)await runner.cancel();else{const job=getImport(id);if(['running','preparing','paused'].includes(job.status)){preparationControllers.get(id)?.abort();const next={...job,generation:job.generation+1,status:'cancelled' as const};delete next.error;publishImport(freeze(next),job.generation);await persist();}}},
async pause(id:Id){assertImportWritable();const runner=importRunners.get(id);if(runner)await runner.pause();else{const job=getImport(id);if(['running','preparing'].includes(job.status)){if(job.source?.integrityVersion!==2)preparationControllers.get(id)?.abort();publishImport(freeze({...job,generation:job.generation+1,status:'paused'}),job.generation);await persist();}}},
async resume(id:Id){assertImportWritable();const job=getImport(id);if(['completed','partial','cancelled'].includes(job.status))return;if(['running','preparing'].includes(job.status)){await loadImport(id);return;}await importLoads.get(id);await importRunners.get(id)?.settle();if(getImport(id).source?.integrityVersion===2)await resetImportReader(id);await persist();await scheduler.resume();assertImportWritable();const current=getImport(id);if(current.status!=='paused')return;const next={...current,generation:current.generation+1,status:current.source?.kind==='duplicate-snapshot'?'preparing' as const:'running' as const};delete next.error;publishImport(freeze(next),current.generation);await loadImport(id);},
async cancel(id:Id){assertImportWritable();const runner=importRunners.get(id);if(runner)await runner.cancel();else{const job=getImport(id);if(['running','preparing','paused'].includes(job.status)){if(job.source?.integrityVersion!==2)preparationControllers.get(id)?.abort();const next={...job,generation:job.generation+1,status:'cancelled' as const};delete next.error;publishImport(freeze(next),job.generation);await persist();}}if(!sourceNeedsSeal(getImport(id)))verificationTasks.get(id)?.controller.abort();},
});
scheduler.start(); prepareDigests();prepareImages(); runTracker.notify([...initial.runs.keys()]);
for(const job of initial.imports?.values()??[])if(['running','preparing'].includes(job.status))void loadImport(job.id);
for(const job of initial.imports?.values()??[])if(sourceNeedsSeal(job)||['running','preparing'].includes(job.status))void loadImport(job.id);
return Object.freeze({
editor, sessions, imports,
loading:Object.freeze({getState:()=>editor.getSnapshot().runtime.loading,
@@ -1004,12 +1050,12 @@ export async function openProjectEngine(options: EngineOptions) {
cancelRecovery:async()=>{await recovery?.cancel();},
getDurability,
subscribeDurability(listener:()=>void){if(closed)return()=>{};durabilityListeners.add(listener);return()=>{durabilityListeners.delete(listener);};},
retry: async () => { if(recovery)await recovery.cancel();assertWritable(false);await persist();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([...adoptionTasks.values()]);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();await scheduler.settle();await persist(); },
retry: async () => { if(recovery)await recovery.cancel();assertWritable(false);await persist();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();}for(const job of editor.getSnapshot().runtime.imports?.values()??[])if(sourceNeedsSeal(job)){await importLoads.get(job.id);await resetImportReader(job.id);await loadImport(job.id);}await scheduler.resume();await runTracker.retry(); },
async settle() { await Promise.resolve();await Promise.allSettled([...adoptionTasks.values()]);await Promise.allSettled([...copyPreparations.values()]);while(importLoads.size)await Promise.all([...importLoads.values()]);await Promise.all([...importRunners.values()].map(runner=>runner.settle())); while(verificationTasks.size)await Promise.allSettled([...verificationTasks.values()].map(task=>task.promise));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();await scheduler.settle();await persist(); },
async close() {
if (closed) return;
closing = true;notifyDurability();loadingReadController?.abort();await loadingRead?.catch(()=>{});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()]);for(const controller of adoptionTasks.keys())controller.abort();await Promise.allSettled([...adoptionTasks.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;
closing = true;notifyDurability();loadingReadController?.abort();await loadingRead?.catch(()=>{});unsubscribeSizing();orphanController?.abort();await Promise.allSettled([collection,sourceCollection,orphanCollection].filter(Boolean));digestLifecycle.abort();for(const {controller}of verificationTasks.values())controller.abort();for(const controller of preparationControllers.values())controller.abort();for(const controller of copyPreparations.keys())controller.abort();await Promise.allSettled([...copyPreparations.values()]);for(const controller of adoptionTasks.keys())controller.abort();await Promise.allSettled([...adoptionTasks.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(verificationTasks.size)await Promise.allSettled([...verificationTasks.values()].map(task=>task.promise));for(const {reader}of proofReaders.values())reader.close();proofReaders.clear();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();displayCache.releaseOwner(displayOwner);if(!options.displays)displayCache.close();
// Failure keeps the leader/recovery material available for an explicit retry of close.
src/lib/canvas-runtime/import-verification.ts
+25
View File
@@ -0,0 +1,25 @@
import type {Id} from '../canvas-core/types.ts';
import {equal,fault} from '../canvas-core/values.ts';
import {importNeedsSeal,validateImportJob,type ImportJob} from './imports.ts';
import {validateImportReadProof} from './import-read-proof.ts';
/** A derived durable-cut index, never an in-memory permission boolean. */
export function importSeals(jobs:ReadonlyMap<Id,ImportJob>|undefined){
const seals=new Map<Id,string>();for(const job of jobs?.values()??[])if(job.source?.integrityVersion===2&&job.source.sha256&&!job.source.integrityError)seals.set(job.source.fileId,job.source.sha256);return seals;
}
export function importsMaySend(jobs:ReadonlyMap<Id,ImportJob>|undefined,seals:ReadonlyMap<Id,string>){
for(const job of jobs?.values()??[])if(job.source?.integrityVersion===2&&!(job.status==='cancelled'&&!job.committedSourceKeys.length)&&(importNeedsSeal(job)||seals.get(job.source.fileId)!==job.source.sha256))return false;return true;
}
/** Only engine-owned publication paths may advance read evidence or source verification. */
export function assertImportEvidence(previous:ReadonlyMap<Id,ImportJob>|undefined,next:ReadonlyMap<Id,ImportJob>|undefined,owner:Id|undefined){
if(previous===next)return;
for(const[id,job]of previous??[])if(job.source?.integrityVersion===2&&!next?.has(id))throw fault('import_integrity_owned',[id]);
const execution=(job:ImportJob|undefined)=>{if(!job)return job;const{generation:_,source,...body}=job;if(!source)return body;const{retirement:__,...identity}=source;return {...body,source:identity};};
for(const[id,job]of next??[]){const source=job.source,old=previous?.get(id)?.source;if(source?.integrityVersion!==2){if(old?.integrityVersion===2)throw fault('import_integrity_owned',[id]);continue;}
validateImportJob(job);const proof=validateImportReadProof(source.readProof,source.fileId,source.byteLength);
if(proof.blocks.some((block,index)=>block.index!==index))throw fault('invalid_import_read_proof');
const changed=!equal(old?.readProof,proof)||old?.sha256!==source.sha256||old?.integrityError!==source.integrityError;
if(owner!==id&&(changed||!equal(execution(previous?.get(id)),execution(job))))throw fault('import_integrity_owned',[id]);
if(old?.integrityVersion===2){if(old.fileId!==source.fileId||old.byteLength!==source.byteLength||old.sha256&&old.sha256!==source.sha256)throw fault('import_source_changed',[id]);const hashes=new Map(proof.blocks.map(block=>[block.index,block.sha256]));for(const block of old.readProof!.blocks)if(hashes.get(block.index)!==block.sha256)throw fault('import_source_changed',[id]);}
}
}
src/lib/canvas-runtime/imports.ts
+22 -5
View File
@@ -1,3 +1,4 @@
import {validateImportReadProof,type ImportReadProof} from './import-read-proof.ts';
import {readImportRange,type ImportByteReader} from './import-byte-reader.ts';
import {createImportBudget,type ImportScheduling} from './import-budget.ts';
import {validateImportFileRetention,type FileRetirement,type PreviousImportFile} from './import-file-retention.ts';
@@ -12,7 +13,7 @@ export type ImportUnit = { readonly key: string; readonly dependencies: readonly
export type ImportLocator = { readonly key: string; readonly cursor: number; readonly nextCursor: number; readonly dependencies: readonly string[]; readonly digest: string; readonly byteLength: number; readonly changeCount: number };
export type ImportFailure = { readonly key: string; readonly cursor: number; readonly code: string; readonly dependencies?: readonly string[] };
export type LoadingImportAdmission={readonly version:1;readonly sourceToken:string;readonly operationSequence:number;readonly requiredCanvasIds:readonly Id[]};
export type NativeImportSource = { readonly loadingAdmission?:LoadingImportAdmission; readonly kind: 'native-document' | 'native-json-array' | 'native-ndjson' | 'duplicate-ndjson' | 'duplicate-snapshot' | 'duplicate-plan'; readonly retirement?:FileRetirement;readonly previousFiles?:readonly PreviousImportFile[];readonly snapshotVersion?:2;readonly plan?:DuplicateCheckpoint; readonly copy?: {readonly sourceRevision:number; readonly roots:readonly Id[]}; readonly integrityVersion?:1;readonly sha256?:string; readonly name?:string; readonly fileId: Id; readonly byteLength: number; readonly targetCanvasId: Id; readonly targetEntry?: {readonly id:Id;readonly incarnation:number} };
export type NativeImportSource = { readonly loadingAdmission?:LoadingImportAdmission; readonly kind: 'native-document' | 'native-json-array' | 'native-ndjson' | 'duplicate-ndjson' | 'duplicate-snapshot' | 'duplicate-plan'; readonly retirement?:FileRetirement;readonly previousFiles?:readonly PreviousImportFile[];readonly snapshotVersion?:2;readonly plan?:DuplicateCheckpoint; readonly copy?: {readonly sourceRevision:number; readonly roots:readonly Id[]}; readonly integrityVersion?:1|2;readonly readProof?:ImportReadProof;readonly integrityError?:string;readonly sha256?:string; readonly name?:string; readonly fileId: Id; readonly byteLength: number; readonly targetCanvasId: Id; readonly targetEntry?: {readonly id:Id;readonly incarnation:number} };
export type ImportBinding = {readonly kind:'element';readonly incarnation:number}|{readonly kind:'canvas';readonly entryElementId?:Id};
export type ImportJob = {
readonly version: 1 | 2 | 3; readonly id: Id; readonly projectId: Id; readonly sourceFingerprint: string; readonly parserVersion: number;
@@ -22,6 +23,7 @@ export type ImportJob = {
readonly bindings?: Readonly<Record<Id,ImportBinding>>;
readonly stableIds: Readonly<Record<string, Id>>; readonly pending: readonly ImportLocator[]; readonly committedSourceKeys: readonly string[]; readonly failures: readonly ImportFailure[]; readonly error?: string;
};
export const importNeedsSeal=(job:ImportJob)=>job.source?.integrityVersion===2&&!(job.status==='cancelled'&&!job.committedSourceKeys.length)&&(!job.source.sha256||!!job.source.integrityError);
export function createImportJob(input: Pick<ImportJob, 'id' | 'projectId' | 'sourceFingerprint' | 'parserVersion' | 'source'>): ImportJob {
if (![input.id,input.projectId,input.sourceFingerprint].every(value=>typeof value==='string'&&!!value) || !Number.isSafeInteger(input.parserVersion) || input.parserVersion<1) throw fault('invalid_import_job');
return freeze({ id: input.id, projectId: input.projectId, sourceFingerprint: input.sourceFingerprint, parserVersion: input.parserVersion, ...(input.source?{source:{...clone(input.source),integrityVersion:input.source.integrityVersion??1}}:{}), version: 3, generation: 0, status: 'paused', cursor: 0, eof: false, bindings: {}, stableIds: {}, pending: [], committedSourceKeys: [], failures: [] });
@@ -38,7 +40,12 @@ export function validateImportJob(job: ImportJob): void {
if(job.source){
validateImportFileRetention(job.source,job.status);
if(job.source.snapshotVersion!==undefined&&(job.source.snapshotVersion!==2||!['duplicate-snapshot','duplicate-plan'].includes(job.source.kind)))throw fault('copy_snapshot_interpreter_unavailable');
if(job.source.integrityVersion!==undefined&&job.source.integrityVersion!==1)throw fault('import_source_integrity_interpreter_unavailable');
if(job.source.integrityVersion!==undefined&&![1,2].includes(job.source.integrityVersion))throw fault('import_source_integrity_interpreter_unavailable');
if(job.source.integrityVersion===2){
if(!['native-document','native-json-array','native-ndjson'].includes(job.source.kind)||job.source.kind==='native-document'&&job.parserVersion!==5)throw fault('import_source_integrity_interpreter_unavailable');
const proof=validateImportReadProof(job.source.readProof,job.source.fileId,job.source.byteLength);if(proof.blocks.some((block,index)=>block.index!==index)||job.source.byteLength>0&&!proof.blocks.length&&(job.cursor>0||job.eof||job.pending.length||job.committedSourceKeys.length||Object.keys(job.stableIds).length))throw fault('import_source_integrity_missing');
if(job.source.integrityError!==undefined&&(typeof job.source.integrityError!=='string'||!job.source.integrityError))throw fault('invalid_import_source');
}else if(job.source.readProof!==undefined||job.source.integrityError!==undefined)throw fault('import_source_integrity_interpreter_unavailable');
if(job.source.sha256!==undefined&&(typeof job.source.sha256!=='string'||!(/^[a-f0-9]{64}$/.test(job.source.sha256))))throw fault('invalid_import_source');
if(job.source.integrityVersion===1&&!job.source.sha256&&(job.cursor!==0||job.eof||job.pending.length||job.committedSourceKeys.length||Object.keys(job.stableIds).length||(job.preparation?.allocationIds.length??0)>0||(job.preparation?.steps??0)>0))throw fault('import_source_integrity_missing');
}
@@ -98,6 +105,7 @@ export function createNdjsonImportSource(options:{bytes:Blob;fingerprint:string;
}
export type ImportRunnerPorts={
checkpointSource?:(job:ImportJob)=>ImportJob;
source:ImportSource;getJob():ImportJob;createId():Id;
/** Synchronous compare-and-publish; never write a second private job store. */
publish(next:ImportJob,expectedGeneration:number):void;
@@ -120,13 +128,22 @@ export function createImportRunner(ports:ImportRunnerPorts){
const initial=ports.getJob(),identity={id:initial.id,projectId:initial.projectId,sourceFingerprint:initial.sourceFingerprint,parserVersion:initial.parserVersion},budget=ports.frameBudgetMs??6,maxPending=ports.maxPending??4096,maxBytes=ports.maxPendingBytes??1024*1024,maxBatch=ports.maxBatch??32,now=ports.now??(()=>performance.now());
validateImportJob(initial);
if(initial.source?.kind==='duplicate-snapshot')throw fault('import_interpreter_unavailable');
if(initial.source&&(!initial.source.sha256||initial.source.integrityVersion!==1))throw fault('import_source_integrity_unavailable');
if(initial.source&&(initial.source.integrityVersion===2?!ports.checkpointSource:!initial.source.sha256||initial.source.integrityVersion!==1))throw fault('import_source_integrity_unavailable');
if(!Number.isFinite(budget)||budget<=0||![maxPending,maxBytes,maxBatch].every(value=>Number.isSafeInteger(value)&&value>0))throw fault('invalid_import_budget');
let costPerUnit:number|undefined;
let active:Promise<void>|undefined,controller:AbortController|undefined,closed=false,restart=false;const holds=new Set<symbol>();
const report=(error:unknown)=>{try{ports.onError(error);}catch{/* Reporting cannot change imported content. */}};
function current(){const job=ports.getJob();if(job.id!==identity.id||job.projectId!==identity.projectId||job.sourceFingerprint!==identity.sourceFingerprint||job.parserVersion!==identity.parserVersion)throw fault('import_identity_changed');return job;}
function next(job:ImportJob,patch:Partial<ImportJob>):ImportJob{const value={...job,...patch,version:3 as const,generation:job.generation+1};if(value.error===undefined)delete value.error;return freeze(value);}
function checkpointEvidence(job:ImportJob):ImportJob{
if(!ports.checkpointSource)return job;
const value=ports.checkpointSource(job);if(!value||!equal({...job,source:null},{...value,source:null}))throw fault('invalid_import_checkpoint');
const strip=(source:ImportJob['source'])=>{if(!source)return source;const{readProof:_,sha256:__,integrityError:___,...identity}=source;return identity;};
if(!equal(strip(job.source),strip(value.source)))throw fault('invalid_import_checkpoint');
if(value.source?.integrityVersion===2){validateImportReadProof(value.source.readProof,value.source.fileId,value.source.byteLength);if(value.source.sha256!==undefined&&(typeof value.source.sha256!=='string'||!/^[a-f0-9]{64}$/.test(value.source.sha256)))throw fault('invalid_import_checkpoint');if(value.source.integrityError!==undefined&&(typeof value.source.integrityError!=='string'||!value.source.integrityError))throw fault('invalid_import_checkpoint');}
else if(!equal(job.source,value.source))throw fault('invalid_import_checkpoint');
return value;
}
function next(job:ImportJob,patch:Partial<ImportJob>):ImportJob{const value={...job,...patch,version:3 as const,generation:job.generation+1};if(value.error===undefined)delete value.error;return freeze(checkpointEvidence(value));}
function publish(job:ImportJob,patch:Partial<ImportJob>){const value=next(job,patch);ports.publish(value,job.generation);if(!equal(current(),value))throw fault('invalid_import_publish');return value;}
function prepare(line:ImportLine,job:ImportJob){
const ids:Record<string,Id>=Object.assign(Object.create(null),job.stableIds),used=new Set(Object.values(ids));
@@ -144,7 +161,7 @@ export function createImportRunner(ports:ImportRunnerPorts){
for(const unit of units)for(const change of unit.changes)if(change.kind==='element'&&change.before===null&&change.after)bindings[change.id]={kind:'element',incarnation:change.after.incarnation};else if(change.kind==='canvas'&&change.before===null&&change.after)bindings[change.id]={kind:'canvas',...(change.after.parent?{entryElementId:change.after.parent.entryElementId}:{})};
return next(job,{pending,committedSourceKeys:[...job.committedSourceKeys,...units.map(unit=>unit.key)],bindings:{...bindings}});
}
function commit(units:readonly ImportUnit[],job:ImportJob,value:ImportJob){const start=now(),result=ports.commit(units,value,job.generation),cost=Math.max(.01,(now()-start)/units.length);costPerUnit=costPerUnit===undefined?cost:Math.max(cost,costPerUnit*.8+cost*.2);if(!['applied','rejected'].includes(result?.status)||!equal(current(),result.status==='applied'?value:job))throw fault('invalid_import_commit');return result;}
function commit(units:readonly ImportUnit[],job:ImportJob,value:ImportJob){const base=current();if(base.generation!==job.generation)throw fault('stale_import_job');value=freeze(checkpointEvidence(value));const start=now(),result=ports.commit(units,value,job.generation),cost=Math.max(.01,(now()-start)/units.length);costPerUnit=costPerUnit===undefined?cost:Math.max(cost,costPerUnit*.8+cost*.2);if(!['applied','rejected'].includes(result?.status)||!equal(current(),result.status==='applied'?value:base))throw fault('invalid_import_commit');return result;}
async function pump(signal:AbortSignal){
let frame=now(),didWork=false;
async function attempt(units:readonly ImportUnit[],job:ImportJob,proposed:ImportJob):Promise<Result|undefined>{
src/lib/canvas-runtime/recovery-encoding.ts
+2 -2
View File
@@ -52,8 +52,8 @@ export function encodeRecoveryValue(input:unknown):Json {
}
/** Only codec-owned, validated immutable images qualify for the storage copy fast path. */
export function sealRecoveryImage(formatVersion:1|2|3|4|5|6|7|8|9|10|11|12|13|14|15|16,rootRevision:number,localRevision:number,payload:unknown):RecoveryImage {
if(![1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16].includes(formatVersion)||!Number.isSafeInteger(rootRevision)||rootRevision<0||!Number.isSafeInteger(localRevision)||localRevision<0||localRevision>rootRevision)throw fault('corrupt_recovery_image');
export function sealRecoveryImage(formatVersion:1|2|3|4|5|6|7|8|9|10|11|12|13|14|15|16|17,rootRevision:number,localRevision:number,payload:unknown):RecoveryImage {
if(![1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17].includes(formatVersion)||!Number.isSafeInteger(rootRevision)||rootRevision<0||!Number.isSafeInteger(localRevision)||localRevision<0||localRevision>rootRevision)throw fault('corrupt_recovery_image');
const image=Object.freeze({formatVersion,rootRevision,localRevision,payload:encodeRecoveryValue(payload)});sealed.add(image);return image;
}
export function isSealedRecoveryImage(image:RecoveryImage):boolean{return sealed.has(image);}
src/lib/canvas-runtime/recovery.ts
+4 -3
View File
@@ -30,7 +30,7 @@ export function encodeRecovery(snapshot: EngineSnapshot): RecoveryImage {
operations: [...snapshot.operations],
runtime: { ...(snapshot.runtime.assetBindings?.size?{assetBindings:[...snapshot.runtime.assetBindings]}:{}), ...(snapshot.runtime.loadingRead?{loadingRead:{document:documentValue(snapshot.runtime.loadingRead.document),versions:[...snapshot.runtime.loadingRead.versions]}}:{}), ...(snapshot.runtime.loading?{loading:snapshot.runtime.loading}:{}), ...(snapshot.runtime.orphanFiles?.size?{orphanFiles:[...snapshot.runtime.orphanFiles]}:{}), imports:[...(snapshot.runtime.imports??[])], recoveries:[...(snapshot.runtime.recoveries??[])], runs: [...snapshot.runtime.runs], resources: [...snapshot.runtime.resources], resourcePins: [...snapshot.runtime.resourcePins], acknowledged: documentValue(snapshot.runtime.acknowledged), versions: [...snapshot.runtime.versions], receipts: [...snapshot.runtime.receipts], conflicts: [...snapshot.runtime.conflicts], ledger: { ...snapshot.runtime.ledger, operations: [...snapshot.runtime.ledger.operations], requests: [...snapshot.runtime.ledger.requests] } },
};
return sealRecoveryImage([...(snapshot.runtime.imports?.values()??[])].some(job=>job.source?.kind==='native-document'&&job.parserVersion===5)?16:[...(snapshot.runtime.imports?.values()??[])].some(job=>job.source?.kind==='native-document')?15:[...(snapshot.runtime.imports?.values()??[])].some(job=>job.source?.kind==='native-json-array')?14:snapshot.runtime.assetBindings?.size?13:snapshot.runtime.loading?(snapshot.runtime.loading.version===6?12:snapshot.runtime.loading.version===5?11:snapshot.runtime.loading.version===4?10:snapshot.runtime.loading.version===3?9:snapshot.runtime.loading.version===2?8:7):[...snapshot.operations.values()].some(intent=>intent.historyEffect)||[...snapshot.runtime.resources.values()].some(resource=>resource.sizing)?6:snapshot.runtime.orphanFiles?.size?5:[...(snapshot.runtime.imports?.values()??[])].some(job=>job.source?.retirement||job.source?.previousFiles?.length)?4:[...snapshot.runtime.resources.values()].some(resource=>resource.retirement)?3:snapshot.runtime.imports?.size ? 2 : 1,snapshot.rootRevision,snapshot.revision,payload);
return sealRecoveryImage([...(snapshot.runtime.imports?.values()??[])].some(job=>job.source?.integrityVersion===2)?17:[...(snapshot.runtime.imports?.values()??[])].some(job=>job.source?.kind==='native-document'&&job.parserVersion===5)?16:[...(snapshot.runtime.imports?.values()??[])].some(job=>job.source?.kind==='native-document')?15:[...(snapshot.runtime.imports?.values()??[])].some(job=>job.source?.kind==='native-json-array')?14:snapshot.runtime.assetBindings?.size?13:snapshot.runtime.loading?(snapshot.runtime.loading.version===6?12:snapshot.runtime.loading.version===5?11:snapshot.runtime.loading.version===4?10:snapshot.runtime.loading.version===3?9:snapshot.runtime.loading.version===2?8:7):[...snapshot.operations.values()].some(intent=>intent.historyEffect)||[...snapshot.runtime.resources.values()].some(resource=>resource.sizing)?6:snapshot.runtime.orphanFiles?.size?5:[...(snapshot.runtime.imports?.values()??[])].some(job=>job.source?.retirement||job.source?.previousFiles?.length)?4:[...snapshot.runtime.resources.values()].some(resource=>resource.retirement)?3:snapshot.runtime.imports?.size ? 2 : 1,snapshot.rootRevision,snapshot.revision,payload);
}
type ObjectValue = Record<string, unknown>;
function object(value: unknown): ObjectValue { if (!value || Array.isArray(value) || typeof value !== 'object') throw fault('corrupt_recovery_image'); return value as ObjectValue; }
@@ -94,7 +94,7 @@ function readHistory(history:ObjectValue,operations:ReadonlyMap<Id,OperationInte
return Object.freeze({undo:freeze(undo),redo:freeze(redo),entries});
}
export function decodeRecovery(image: RecoveryImage, registry: Registry): CoreSnapshot & { readonly rootRevision: number; readonly runtime: RuntimeState } {
if (![1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16].includes(image.formatVersion)) throw fault('unsupported_recovery_version');
if (![1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17].includes(image.formatVersion)) throw fault('unsupported_recovery_version');
integer(image.rootRevision); integer(image.localRevision);
if (image.localRevision > image.rootRevision || !validJson(image.payload)) throw fault('corrupt_recovery_image');
// Own the parsed records; mutating a caller's RecoveryImage must not mutate live state.
@@ -250,7 +250,8 @@ export function decodeRecovery(image: RecoveryImage, registry: Registry): CoreSn
});
const imports=table<ImportJob>(runtime.imports??[],(value,id)=>{const job=value as ImportJob;validateImportJob(job);if(job.id!==id||job.projectId!==document.projectId)throw fault('corrupt_recovery_image');return freeze(job);});
const hasJsonArrays=[...imports.values()].some(job=>job.source?.kind==='native-json-array');if(hasJsonArrays&&image.formatVersion<14||image.formatVersion===14&&!hasJsonArrays)throw fault('unsupported_recovery_version');
const hasDocuments=[...imports.values()].some(job=>job.source?.kind==='native-document'),hasStreamingDocuments=[...imports.values()].some(job=>job.source?.kind==='native-document'&&job.parserVersion===5);if(hasDocuments&&image.formatVersion!==(hasStreamingDocuments?16:15)||image.formatVersion>=15&&!hasDocuments)throw fault('unsupported_recovery_version');
const hasReadProof=[...imports.values()].some(job=>job.source?.integrityVersion===2);if(hasReadProof!==(image.formatVersion===17))throw fault('unsupported_recovery_version');
const hasDocuments=[...imports.values()].some(job=>job.source?.kind==='native-document'),hasStreamingDocuments=[...imports.values()].some(job=>job.source?.kind==='native-document'&&job.parserVersion===5);if(hasDocuments&&image.formatVersion!==(hasReadProof?17:hasStreamingDocuments?16:15)||image.formatVersion>=15&&!hasDocuments&&!hasReadProof)throw fault('unsupported_recovery_version');
for(const [id,entry]of importFiles(imports)){if(resources.has(id)||assetBindings.has(id))throw fault('import_file_identity_conflict',[id]);if((entry.file.retirement||imports.get(entry.jobId)?.source?.previousFiles?.length)&&image.formatVersion<4)throw fault('unsupported_recovery_version');}
const orphanFiles=table<OrphanFile>(runtime.orphanFiles??[],(file,id)=>{validateOrphanFile(file,id);return freeze(file);});
if(orphanFiles.size&&image.formatVersion<5)throw fault('unsupported_recovery_version');for(const id of orphanFiles.keys())if(resources.has(id)||assetBindings.has(id)||importFiles(imports).has(id))throw fault('local_file_identity_conflict',[id]);
src/lib/canvas-runtime/resource-retention.ts
+2 -1
View File
@@ -1,3 +1,4 @@
import {importNeedsSeal} from './imports.ts';
import type {Id,Registry} from '../canvas-core/types.ts';
import {fault,readMap,readSet,freeze} from '../canvas-core/values.ts';
import type {EngineSnapshot,RuntimeState} from './contracts.ts';
@@ -14,7 +15,7 @@ export function resourceRetentionPins(snapshot:EngineSnapshot,registry:Registry,
for(const [id,owners]of base.pins)for(const owner of owners)if(owner.startsWith('element:'))add(id,`acknowledged:${owner}`);
// Sources and server run envelopes are not all interpreted resource schemas yet.
// Pin the whole known local depot rather than claiming opaque task data contains no references.
for(const job of runtime.imports?.values()??[])if(!['completed','partial','cancelled'].includes(job.status)){if(job.source)add(job.source.fileId,`import-source:${job.id}`);for(const [id,resource]of runtime.resources)if(resource.retirement?.purgedAt===undefined)add(id,`import:${job.id}`);}
for(const job of runtime.imports?.values()??[])if(!['completed','partial','cancelled'].includes(job.status)||importNeedsSeal(job)){if(job.source)add(job.source.fileId,`import-source:${job.id}`);for(const [id,resource]of runtime.resources)if(resource.retirement?.purgedAt===undefined)add(id,`import:${job.id}`);}
for(const run of runtime.runs.values())for(const [id,resource]of runtime.resources)if(resource.retirement?.purgedAt===undefined)add(id,`run:${run.id}`);
for(const resource of runtime.resources.values())if(resource.image&&(resource.image.version!==1||resource.image.status==='pending'))add(resource.id,resource.image.version!==1?'image-decoder-unavailable':'image-decode');
for(const [owner,ids]of temporary)for(const id of ids)add(id,owner);
src/lib/canvas-runtime/storage-port.ts
+1 -1
View File
@@ -2,7 +2,7 @@ import type { Json } from '../canvas-core/types.ts';
/** Portable durable-store contract. Browser database types belong to the adapter. */
export type StorageScope = { readonly accountId: string; readonly workspaceId: string; readonly projectId: string };
export type RecoveryImage = { readonly formatVersion: 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8 | 9 | 10 | 11 | 12 | 13 | 14 | 15 | 16; readonly rootRevision: number; readonly localRevision: number; readonly payload: Json };
export type RecoveryImage = { readonly formatVersion: 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8 | 9 | 10 | 11 | 12 | 13 | 14 | 15 | 16 | 17; readonly rootRevision: number; readonly localRevision: number; readonly payload: Json };
export type DurableBatch = { readonly image: RecoveryImage; readonly deleteFiles?:readonly string[]; readonly depot?:import('./asset-depot.ts').DepotCut; readonly files?: readonly { readonly id: string; readonly bytes: Blob }[] };
export type StorageLeader = { readonly scope: StorageScope; readonly fence: number; readonly active: boolean; release(): Promise<void> };
export type StoredFileInfo={readonly id:string;readonly byteLength:number;readonly mimeType:string};
src/lib/canvas-runtime/storage.ts
+1 -1
View File
@@ -31,7 +31,7 @@ function purgedFileIds(image:RecoveryImage):Set<string>{
return result;
}
function validateImage(image: RecoveryImage) {
if (![1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16].includes(image.formatVersion)) throw fault('unsupported_recovery_version');
if (![1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17].includes(image.formatVersion)) throw fault('unsupported_recovery_version');
if (!Number.isSafeInteger(image.rootRevision) || image.rootRevision < 0 || !Number.isSafeInteger(image.localRevision) || image.localRevision < 0 || image.localRevision > image.rootRevision || !validJson(image.payload)) throw fault('corrupt_recovery_image');
}
const requestValue = <T>(request: IDBRequest<T>) => new Promise<T>((resolve, reject) => { request.onsuccess = () => resolve(request.result); request.onerror = () => reject(request.error); });