diff --git a/src/lib/canvas-runtime/copy-budget.ts b/src/lib/canvas-runtime/copy-budget.ts index ee66a52..04f9008 100644 --- a/src/lib/canvas-runtime/copy-budget.ts +++ b/src/lib/canvas-runtime/copy-budget.ts @@ -1,11 +1,18 @@ +/** + * 复用导入的合作调度机制,同时维持copy专用取消/配置错误码。 + * 它不分配ID或落盘,单个原生encode/解码步骤的大小仍必须由调用者另外限制。 + */ import {fault} from '../canvas-core/values.ts'; import {createImportBudget,type ImportScheduling} from './import-budget.ts'; /** Share import scheduling while retaining the copy cancellation/error contract. */ +/** 验证copy预算后包装ImportBudget;调用者只有真正await返回Promise才会交出执行权。 */ export function createCopyBudget(signal:AbortSignal,options:ImportScheduling={}){ if(signal.aborted)throw fault('copy_aborted'); if(options.frameBudgetMs!==undefined&&(!Number.isFinite(options.frameBudgetMs)||options.frameBudgetMs<=0))throw fault('invalid_copy_budget'); const budget=createImportBudget(signal,options); + /** signal已取消时将同步/异步预算失败统一为copy_aborted,其它错误原样抛给宿主。 */ const fail=(error:unknown):never=>{if(signal.aborted)throw fault('copy_aborted');throw error;}; + /** 每次step先查copy取消,保持底层同步返回undefined或异步让出的原调度语义。 */ return Object.freeze({step(units=1):Promise|undefined{ if(signal.aborted)throw fault('copy_aborted'); try{return budget.step(units)?.catch(fail);}catch(error){return fail(error);} diff --git a/src/lib/canvas-runtime/copy-snapshot-index.ts b/src/lib/canvas-runtime/copy-snapshot-index.ts index e334cd3..29229c8 100644 --- a/src/lib/canvas-runtime/copy-snapshot-index.ts +++ b/src/lib/canvas-runtime/copy-snapshot-index.ts @@ -1,3 +1,8 @@ +/** + * 格式2的复制源:前段保存轻量topology和payload字节定位,后段保存原完整元素记录。 + * 准备阶段只扫描索引/结构;真正复制某记录前才按range读取和校验payload,支持边转换边加入。 + * 编码仍生成完整Blob,索引/定位仍占O(records)内存,不能据lazy解码宣称总内存常数。 + */ import type {CanvasRecord,Document,ElementRecord,Id,Json,Registry,Data} from '../canvas-core/types.ts'; import {createDocument,validateDocumentTopologySteps} from '../canvas-core/document.ts'; import {equal,fault,freeze,validJson} from '../canvas-core/values.ts'; @@ -5,6 +10,8 @@ import {createNdjsonImportSource} from './imports.ts'; import {createCopyBudget} from './copy-budget.ts'; import type {ImportScheduling} from './import-budget.ts'; /** Only ownership, endpoint and geometry fields needed to decide scope/coordinates. No component content. */ +/** 保留ID/owner/端点/几何及contentScale/unitScale,清组件内容,边path置null用于范围/坐标规划。 + * 返回结构记录不能直接作为用户元素发布,也不证明组件payload符合schema。 */ export function copyTopology(record:ElementRecord):ElementRecord{ const raw=record.data.presentation,presentation=raw&&typeof raw==='object'&&!Array.isArray(raw)?Object.fromEntries(['contentScale','unitScale'].filter(key=>Object.hasOwn(raw,key)).map(key=>[key,raw[key]])):{}; const data:Data=Object.keys(presentation).length?{presentation}:{}; @@ -12,20 +19,26 @@ export function copyTopology(record:ElementRecord):ElementRecord{ return record.structure==='node'?{...common,structure:'node',geometry:record.geometry}:{...common,structure:'connection',source:record.source,target:record.target,path:null}; } /** Format 2 places a compact topology/byte-locator index before the uninterpreted payload region. */ +/** 输出完整format2源:header/canvas/element-index/tombstone/payload标记,然后原元素NDJSON区域。 */ export async function encodeIndexedCopySnapshot(document:Document,signal:AbortSignal,scheduling:ImportScheduling={}):Promise{ const index:BlobPart[]=[],payload:BlobPart[]=[];const budget=createCopyBudget(signal,scheduling);let indexBytes=0,payloadBytes=0; + /** 单记录序列化、UTF8计长与256KiB上限,保持原始记录/索引locator成本可核验。 */ function encode(value:unknown){if(signal.aborted)throw fault('copy_aborted');const text=JSON.stringify(value)+'\n',size=new TextEncoder().encode(text).byteLength;if(size>256*1024)throw fault('record_too_large');return{text,size};} + /** 添加索引记录并检查索引+payload总源上限,offset按payload区域相对字节计。 */ function append(value:unknown){const part=encode(value);index.push(part.text);indexBytes+=part.size;if(indexBytes+payloadBytes>64*1024*1024)throw fault('invalid_import_source');return part.size;} append({kind:'header',version:2,projectId:document.projectId}); for(const record of document.canvases.values()){const size=append({kind:'canvas',record});await budget.step(Math.max(1,Math.ceil(size/1024)));} + // 每个元素同时写小topology索引和完整payload,locator保存精确offset/长度而非JS字符位置。 for(const record of document.elements.values()){ const part=encode({kind:'element',record}),indexSize=append({kind:'element-index',record:copyTopology(record),offset:payloadBytes,byteLength:part.size});payload.push(part.text);payloadBytes+=part.size;if(indexBytes+payloadBytes>64*1024*1024)throw fault('invalid_import_source');await budget.step(Math.max(1,Math.ceil((part.size+indexSize)/1024))); } for(const record of document.tombstones){const size=append({kind:'tombstone',record});await budget.step(Math.max(1,Math.ceil(size/1024)));} append({kind:'payload',byteLength:payloadBytes});if(signal.aborted)throw fault('copy_aborted');return new Blob([...index,...payload],{type:'application/x-ndjson'}); } +/** document只含结构投影,readElement取得完整record;调用方必须明确使用topologyOnly规划模式。 */ export type IndexedCopySnapshot={readonly document:Document;readElement(id:Id,signal:AbortSignal):Promise}; /** Scan only the small index. Component JSON/schema checks occur just before each record enters the runner. */ +/** 只扫描前段索引,验证locator连续/总长/项目与结构,再返回有界窗口读取器。 */ export async function openIndexedCopySnapshot(bytes:Blob,projectId:Id,registry:Registry,signal:AbortSignal,scheduling:ImportScheduling={}):Promise{ if(!(bytes instanceof Blob)||bytes.size>64*1024*1024)throw fault('invalid_import_source'); const budget=createCopyBudget(signal,scheduling),source=createNdjsonImportSource({bytes,fingerprint:'indexed-copy-snapshot'}),canvases:CanvasRecord[]=[],elements:ElementRecord[]=[],tombstones=new Map(),locators=new Map();let cursor=0,header=false,payloadOffset=-1,expectedOffset=0; @@ -33,20 +46,25 @@ export async function openIndexedCopySnapshot(bytes:Blob,projectId:Id,registry:R if(signal.aborted)throw fault('copy_aborted');const start=cursor,line=await source.read(cursor,signal);if(!line||line.error||line.skip||!line.value)throw fault('invalid_copy_snapshot');cursor=line.nextCursor; const value=line.value as Record; if(!header){if(value.kind!=='header'||value.version!==2||value.projectId!==projectId)throw fault('invalid_copy_snapshot');header=true;continue;} + // 标记声明长度必须等于所有locator连续总长,且恰好延伸到Blob EOF,不允许藏额外尾部。 if(value.kind==='payload'){if(value.byteLength!==expectedOffset||cursor+expectedOffset!==bytes.size)throw fault('invalid_copy_snapshot');payloadOffset=cursor;break;} if(value.kind==='canvas')canvases.push(value.record as unknown as CanvasRecord); else if(value.kind==='element-index'){ const record=value.record as unknown as ElementRecord,offset=value.offset as number,byteLength=value.byteLength as number; + // 每条offset严格等于上条结束,ID唯一且record确实是最小topology表示,不能把未知内容塞进结构索引。 if(!record||!Number.isSafeInteger(offset)||offset!==expectedOffset||!Number.isSafeInteger(byteLength)||byteLength<1||byteLength>256*1024||locators.has(record.id)||!equal(copyTopology(record),record))throw fault('invalid_copy_snapshot'); locators.set(record.id,{offset,byteLength});elements.push(record);expectedOffset+=byteLength;if(expectedOffset>bytes.size)throw fault('invalid_copy_snapshot'); }else if(value.kind==='tombstone'){const row=value.record as [Id,number];if(!Array.isArray(row)||row.length!==2||tombstones.has(row[0]))throw fault('invalid_copy_snapshot');tombstones.set(row[0],row[1]);}else throw fault('invalid_copy_snapshot'); await budget.step(Math.max(1,Math.ceil((cursor-start)/1024))); } + // 这里只validateDocumentTopologySteps,不把缺payload的投影误送完整组件schema。 const document=createDocument(projectId,canvases,elements,tombstones); for(const _ of validateDocumentTopologySteps(document,registry))await budget.step(); if(signal.aborted)throw fault('copy_aborted'); // Cache raw byte windows, never decoded component objects. Preserve parser batching across Blob task boundaries. + // 缓存原始byte窗口而非解码对象,最多4窗口/1MiB,减少每记录Blob读取任务又限制常驻字节。 const windows=new Map(),windowBytes=64*1024;let cachedBytes=0; + /** readElement按locator截取原payload,fatal UTF8/JSON、topology一致及已知schema全部通过才返回冻结record。 */ return Object.freeze({document,async readElement(id:Id,signal:AbortSignal){ if(signal.aborted)throw fault('copy_aborted');const locator=locators.get(id);if(!locator)throw fault('element_missing',[id]); const start=Math.floor(locator.offset/windowBytes)*windowBytes,required=locator.offset+locator.byteLength-start;let window=windows.get(start); @@ -54,6 +72,7 @@ export async function openIndexedCopySnapshot(bytes:Blob,projectId:Id,registry:R const raw=window.subarray(locator.offset-start,required);let value:{kind:string;record:ElementRecord}; try{value=JSON.parse(new TextDecoder('utf-8',{fatal:true}).decode(raw));}catch{throw fault('invalid_copy_snapshot');} if(!value||value.kind!=='element'||!validJson(value.record)||!value.record.data||typeof value.record.data!=='object'||Array.isArray(value.record.data)||!equal(copyTopology(value.record),document.elements.get(id)))throw fault('invalid_copy_snapshot'); + // 当前registry已知版本才执行其schema;后续复制转换还会要求对应copy解释器存在。 const record=value.record,definition=registry.components.get(record.componentId);if(registry.resolve(record)==='known'&&(!definition||definition.structure!==record.structure||!definition.validate(record.data)))throw fault('invalid_component_data',[id]); return freeze(record); }}); diff --git a/src/lib/canvas-runtime/copy-snapshot.ts b/src/lib/canvas-runtime/copy-snapshot.ts index 8d03725..f2447ef 100644 --- a/src/lib/canvas-runtime/copy-snapshot.ts +++ b/src/lib/canvas-runtime/copy-snapshot.ts @@ -1,3 +1,7 @@ +/** + * 格式1的完整复制源快照与复制声明签名。源字节在任务准入前从一个不可变文档捕获, + * 后续规划/重开不读取当前用户正在改的live文档;源SHA与持久字节由外层完整性流程绑定。 + */ import type { CanvasRecord, Document, ElementRecord, Registry } from '../canvas-core/types.ts'; import { createDocument, validateDocumentSteps } from '../canvas-core/document.ts'; import { fault } from '../canvas-core/values.ts'; @@ -6,13 +10,16 @@ import { createNdjsonImportSource } from './imports.ts'; import {createCopyBudget} from './copy-budget.ts'; import type {ImportScheduling} from './import-budget.ts'; /** Capture immutable source bytes before a preparation job is admitted. No mutable live getters. */ +/** 逐项编码header/canvas/element/tombstone为NDJSON,检查记录/总源上限并合作让出,最终返回完整Blob。 */ export async function encodeCopySnapshot(document:Document,signal:AbortSignal,scheduling:ImportScheduling={}):Promise{ const budget=createCopyBudget(signal,scheduling),parts:BlobPart[]=[];let bytes=0; + /** 生成一条记录并累计实际UTF8字节,单条≤256KiB、总源≤64MiB;不省略未知组件payload。 */ const append=(value:unknown)=>{if(signal.aborted)throw fault('copy_aborted');const text=JSON.stringify(value)+'\n',size=new TextEncoder().encode(text).byteLength;if(size>256*1024)throw fault('record_too_large');bytes+=size;if(bytes>64*1024*1024)throw fault('invalid_import_source');parts.push(text);return size;}; append({kind:'header',version:1,projectId:document.projectId}); for(const[kind,records]of [['canvas',document.canvases.values()],['element',document.elements.values()],['tombstone',document.tombstones.entries()]] as const)for(const record of records){const size=append({kind,record});await budget.step(Math.max(1,Math.ceil(size/1024)));} if(signal.aborted)throw fault('copy_aborted');return new Blob(parts,{type:'application/x-ndjson'}); } +/** 旧源完整解码,验证同project/header、墓碑唯一,重建并分片校验完整Document后才返回。 */ export async function decodeCopySnapshot(bytes:Blob,projectId:string,registry:Registry,signal:AbortSignal,scheduling:ImportScheduling={}):Promise{ const budget=createCopyBudget(signal,scheduling),source=createNdjsonImportSource({bytes,fingerprint:'captured-copy-snapshot'}),canvases:CanvasRecord[]=[],elements:ElementRecord[]=[],tombstones=new Map();let cursor=0,header=false; while(true){ @@ -23,6 +30,7 @@ export async function decodeCopySnapshot(bytes:Blob,projectId:string,registry:Re await budget.step(Math.max(1,Math.ceil((cursor-start)/1024))); } if(!header||signal.aborted)throw fault(signal.aborted?'copy_aborted':'invalid_copy_snapshot'); + // 不把合法header/记录数量当成完整图合法;parent/owner/端点/schema仍由核心校验。 const document=createDocument(projectId,canvases,elements,tombstones); for(const _ of validateDocumentSteps(document,registry))await budget.step(); if(signal.aborted)throw fault('copy_aborted'); @@ -30,6 +38,7 @@ export async function decodeCopySnapshot(bytes:Blob,projectId:string,registry:Re } /** Copy declarations are a versioned input; never silently reinterpret a persisted preparation. */ +/** 仅对源涉及组件的数据版本/structure/copy声明生成稳定签名,防持久准备被新声明静默重解释。 */ export function copyRegistrySignature(document:Document,registry:Registry):string{ const ids=new Set([...document.elements.values()].map(element=>element.componentId)); return JSON.stringify([...ids].sort().map(id=>{const definition=registry.components.get(id);return definition?{id,dataVersion:definition.dataVersion,structure:definition.structure,copy:definition.copy??null}: {id,missing:true};})); diff --git a/src/lib/canvas-runtime/duplicate-source.ts b/src/lib/canvas-runtime/duplicate-source.ts index 339df4d..3332d87 100644 --- a/src/lib/canvas-runtime/duplicate-source.ts +++ b/src/lib/canvas-runtime/duplicate-source.ts @@ -1,3 +1,8 @@ +/** + * 由原SHA绑定快照与稳定emit检查点提供可seek的lazy复制记录流(parserVersion=2)。 + * 没有第二份完整映射Blob;节点/引用/曲线在read某ordinal时才转换,可用readElement从索引源按需解码payload。 + * 这只是ImportSource,真正的记录摘要、依赖等待、提交和失败报告由统一runner负责。 + */ import type { Document, ElementRecord, CanvasRecord, Id, Registry, Json } from '../canvas-core/types.ts'; import type { DuplicateCheckpoint } from '../canvas-core/duplicate-checkpoint.ts'; import { createDuplicatePlanner } from '../canvas-core/duplicate-checkpoint.ts'; @@ -8,22 +13,30 @@ import { duplicateImportLine } from './import-duplicate.ts'; import type { ImportSource } from './imports.ts'; /** A seekable record cursor over the original SHA-bound snapshot, not a second mapped Blob. */ +/** 要求未输出的emit起点、当前registry签名及有效绑定,ID工厂在恢复重校验中禁止意外再分配。 */ export function createDuplicatePlanSource(document:Document,registry:Registry,state:DuplicateCheckpoint,fingerprint:string,readElement?:(id:Id,signal:AbortSignal)=>Promise):ImportSource{ if(state.phase!=='emit'||state.emitted!==0||state.emittedEdges!==0||state.schemaToken!==copyRegistrySignature(document,registry))throw fault('copy_interpreter_changed'); // Structural/source binding checks; the decoded source has already passed document validation. createDuplicatePlanner(document,registry,state.selection,{createId:()=>{throw fault('unstable_import_identity');},worldOffset:state.offset,...(state.mapping?{mapping:state.mapping}:{})},{sourceToken:state.sourceToken,schemaToken:state.schemaToken,checkpoint:state,topologyOnly:!!readElement}); + // 固定旧→新和新→旧映射;source key仍用原ID,输出change使用已经预留的新ID。 const elements=new Map(state.elementIds),canvases=new Map(state.canvasIds),copiedCanvases=new Set(state.canvases),keys=new Map(),reverseElements=new Map(state.elementIds.map(([id,mapped])=>[mapped,id])),reverseCanvases=new Map(state.canvasIds.map(([id,mapped])=>[mapped,id])); for(const[id,mapped]of elements)keys.set(mapped,`element:${id}`);for(const[id,mapped]of canvases)keys.set(mapped,`canvas:${id}`); // Edge geometry needs mapped endpoint/owner records. Compute these on demand; retain a bounded cache. + /** 给边计算提供新端点/owner视图,get按需转换并只保留32条缓存,不批量生成所有副本对象。 */ function projection(base:ReadonlyMap,reverse:ReadonlyMap,kind:'element'|'canvas'):ReadonlyMap{ + // 未映射ID沿用原表,新ID转换源记录;缓存淘汰只影响计算成本,不改变稳定映射/源数据。 const cache=new Map();const get=(id:Id):T|undefined=>{const old=reverse.get(id);if(!old)return base.get(id);const found=cache.get(id);if(found)return found;const original=base.get(old);if(!original)throw fault('invalid_copy_stream_checkpoint');const record=duplicateTreeChange(document,registry,kind,original,elements,canvases,copiedCanvases,state.offset).after as T;cache.set(id,record);if(cache.size>32)cache.delete(cache.keys().next().value!);return record;}; + /** 迭代先原记录再映射记录,keys/values/forEach均复用同一逻辑投影,不能作为发布文档表。 */ const entries=function*():Generator<[Id,T],undefined>{yield* base;for(const id of reverse.keys())yield[id,get(id)!];return undefined;}; const view:ReadonlyMap={get,has:id=>base.has(id)||reverse.has(id),size:base.size+reverse.size,entries,keys:function*():Generator{for(const[id]of entries())yield id;return undefined;},values:function*():Generator{for(const[,value]of entries())yield value;return undefined;},[Symbol.iterator]:entries,forEach(fn,thisArg){for(const[id,value]of entries())fn.call(thisArg,value,id,view);}};return Object.freeze(view); } + // projected只服务边端点和公共owner的坐标计算;length是tree order+edges的固定ordinal总数。 const projected:Document={...document,elements:projection(document.elements,reverseElements,'element'),canvases:projection(document.canvases,reverseCanvases,'canvas')},length=state.order.length+state.edges.length; + /** read按ordinal选择共享tree/edge转换器;单记录错误返回error,源指纹/完整性与调度另由宿主约束。 */ return Object.freeze({fingerprint,parserVersion:2,async read(cursor:number,signal:AbortSignal){ if(signal.aborted)throw fault('import_aborted');if(!Number.isSafeInteger(cursor)||cursor<0||cursor>length)throw fault('invalid_import_cursor');if(cursor===length)return null; const task=state.order[cursor],key=task?`${task.kind}:${task.id}`:`element:${state.edges[cursor-state.order.length]}`; + // 索引源只在此请求完整元素payload;cancel后重查signal,再生成映射/guard行并检查单条大小。 try{const original=task?.kind==='canvas'?document.canvases.get(task.id)!:readElement?await readElement(task?.id??state.edges[cursor-state.order.length],signal):document.elements.get(task?.id??state.edges[cursor-state.order.length])!; if(signal.aborted)throw fault('import_aborted');const change=cursor256*1024)throw fault('record_too_large'); diff --git a/src/lib/canvas-runtime/duplicate.ts b/src/lib/canvas-runtime/duplicate.ts index 7f14596..8f9d339 100644 --- a/src/lib/canvas-runtime/duplicate.ts +++ b/src/lib/canvas-runtime/duplicate.ts @@ -1,8 +1,14 @@ +/** + * 复制规划器的合作调度适配,不读live editor getter、不发布文档、不访问网络。 + * 完整同步生成器、可恢复语义状态机、lazy输出准备分别提供显式入口,不能把准备完成当成内容已加入。 + * 所有时间片都是软目标,持久源/映射/检查点与提交前外部身份复核由调用者负责。 + */ import { createDuplicatePlanner, type DuplicateCheckpoint } from '../canvas-core/duplicate-checkpoint.ts'; import type { Document, Registry, Target } from '../canvas-core/types.ts'; import { planDuplicateSteps, type DuplicateOptions, type DuplicatePlan } from '../canvas-core/duplicate.ts'; import { clone, fault, freeze } from '../canvas-core/values.ts'; +/** 可注入取消、每片时间/步数和让出端口;checkpoint在完整规划适配中只报告累计语义步数。 */ export type DuplicateScheduling = { readonly signal?: AbortSignal; readonly checkpoint?: (steps:number) => Promise; @@ -17,13 +23,16 @@ export type DuplicateScheduling = { * No document publication or network effects. The caller must persist source/mapping and * revalidate external identities before applying a plan; this is not the import executor. */ +/** 分片消费完整plan迭代器,每片默认6ms/512步;返回前全部记录已生成并校验,但不执行导入。 */ export async function planDuplicateAsync(document: Document, registry: Registry, selection: readonly Target[], options: DuplicateOptions, scheduling: DuplicateScheduling = {}): Promise { const budget = scheduling.frameBudgetMs ?? 6, limit = scheduling.maxStepsPerSlice ?? 512; if (!Number.isFinite(budget) || budget <= 0 || !Number.isSafeInteger(limit) || limit < 1) throw fault('invalid_copy_budget'); const signal = scheduling.signal, now = scheduling.now ?? (() => performance.now()), yieldTask = scheduling.yieldTask ?? (() => new Promise(resolve => setTimeout(resolve, 0))); + /** 在迭代、checkpoint和yield边界检查取消,避免迟到准备结果被宿主继续采用。 */ const check = () => { if (signal?.aborted) throw fault('copy_aborted'); }; check(); // Capture caller-owned small inputs before the first await. Document/registry are immutable core snapshots. + // 首次await前捕获选区/偏移/预留映射,源document与registry按不可变快照合同使用。 const targets = freeze(clone(selection)), settings: DuplicateOptions = Object.freeze({ createId: options.createId, ...(options.worldOffset ? {worldOffset: freeze(clone(options.worldOffset))} : {}), ...(options.mapping ? {mapping: freeze(clone(options.mapping))} : {}) }); const steps = planDuplicateSteps(document, registry, targets, settings); let completedSteps=0; @@ -41,6 +50,7 @@ export async function planDuplicateAsync(document: Document, registry: Registry, if(scheduling.checkpoint)await scheduling.checkpoint(completedSteps); check();await yieldTask(); } + // 成功/失败/取消都关闭生成器,释放其持有的源快照/临时索引引用。 } finally { // Release snapshot references after cancellation or host scheduling failures. steps.return(undefined as never); @@ -48,6 +58,7 @@ export async function planDuplicateAsync(document: Document, registry: Registry, } /** Resume semantic planning cursors. Source/output validation and derived output caches are rebuilt. */ +/** 以完整checkpoint恢复语义游标,仍重做源验证/派生输出缓存,不声称从任意字节零重算。 */ export async function planDuplicateResumable(document:Document,registry:Registry,selection:readonly Target[],options:DuplicateOptions,binding:{sourceToken:string;schemaToken:string;checkpoint?:DuplicateCheckpoint},scheduling:Omit&{checkpoint?:(state:DuplicateCheckpoint)=>Promise}={}):Promise{ const budget=scheduling.frameBudgetMs??6,limit=scheduling.maxStepsPerSlice??512,now=scheduling.now??(()=>performance.now()),yieldTask=scheduling.yieldTask??(()=>new Promise(resolve=>setTimeout(resolve,0))); if(!Number.isFinite(budget)||budget<=0||!Number.isSafeInteger(limit)||limit<1)throw fault('invalid_copy_budget'); @@ -62,18 +73,22 @@ export async function planDuplicateResumable(document:Document,registry:Registry /** Stop before record conversion: the stable order/mappings are sufficient for seekable lazy output. */ +/** 只推进到emit起点:范围/顺序/全部新ID已固定,但尚未转换全部payload,可交lazy source逐条输出。 */ export async function prepareDuplicateStream(document:Document,registry:Registry,selection:readonly Target[],options:DuplicateOptions,binding:{sourceToken:string;schemaToken:string;checkpoint?:DuplicateCheckpoint;topologyOnly?:boolean},scheduling:Omit&{checkpoint?:(state:DuplicateCheckpoint)=>Promise;checkpointIntervalMs?:number;maxStepsPerCheckpoint?:number}={}):Promise{ const budget=scheduling.frameBudgetMs??6,limit=scheduling.maxStepsPerSlice??512,now=scheduling.now??(()=>performance.now()),yieldTask=scheduling.yieldTask??(()=>new Promise(resolve=>setTimeout(resolve,0))); if(!Number.isFinite(budget)||budget<=0||!Number.isSafeInteger(limit)||limit<1)throw fault('invalid_copy_budget'); const check=()=>{if(scheduling.signal?.aborted)throw fault('copy_aborted');};check(); + // 检查点保存频率与UI时间片分开:阶段变化、首次分配、工作量/时间阈值或时钟异常均触发保存。 const interval=scheduling.checkpointIntervalMs??0,maxCheckpointSteps=scheduling.maxStepsPerCheckpoint??8192; if(!Number.isFinite(interval)||interval<0||!Number.isSafeInteger(maxCheckpointSteps)||maxCheckpointSteps<1)throw fault('invalid_copy_budget'); const planner=createDuplicatePlanner(document,registry,selection,options,binding); + // savedSteps等只记录上次已等待checkpoint回调成功的位置,保存耗时不冒充新增规划工作。 let savedPhase:DuplicateCheckpoint['phase']|undefined,savedSteps=binding.checkpoint?.steps??0,savedAllocated=(binding.checkpoint?.elementIds.length??0)+(binding.checkpoint?.canvasIds.length??0),savedAt=now(); while(true){ const start=now();let count=0; do{check();if(planner.ready()){const state=planner.checkpoint();if(interval&&scheduling.checkpoint){await scheduling.checkpoint(state);check();}return state;}if(planner.step())throw fault('invalid_copy_stream_checkpoint');count++;}while(count0||!Number.isFinite(elapsed)||elapsed<0||elapsed>=interval||progress.steps-savedSteps>=maxCheckpointSteps){ await scheduling.checkpoint(planner.checkpoint()); diff --git a/src/lib/canvas-runtime/import-duplicate.ts b/src/lib/canvas-runtime/import-duplicate.ts index 426685b..ff39d31 100644 --- a/src/lib/canvas-runtime/import-duplicate.ts +++ b/src/lib/canvas-runtime/import-duplicate.ts @@ -1,3 +1,8 @@ +/** + * 已映射复制记录与导入runner之间的桥:保留新ID,记录副本依赖及仍指向原图的外部身份guard。 + * 完整plan可编码NDJSON,lazy source也复用同一逐记录转换;不在此上传图片或启动生成任务。 + * 外部guard必须在同步commit边界重新查,不能只信解析时原图还没变。 + */ import type { CanvasRecord, Change, Document, Id, Json, Registry } from '../canvas-core/types.ts'; import { copyReferenceIds } from '../canvas-core/copy-paths.ts'; import type { DuplicatePlan } from '../canvas-core/duplicate.ts'; @@ -7,11 +12,15 @@ import type { ImportJob, ImportLine, ImportUnit } from './imports.ts'; import {createCopyBudget} from './copy-budget.ts'; import type {ImportScheduling} from './import-budget.ts'; +/** 保留外部端点/引用节点的ID、incarnation、owner,允许内容变化但不允许身份被替换。 */ type ExternalElement = {id: Id; incarnation: number; ownerCanvasId: Id}; +/** 保留外部拥有链及入口代次,防重挂/删除恢复后把副本接到另一个分区关系。 */ type ExternalCanvas = {id: Id; parent: CanvasRecord['parent']; entryIncarnation?: number}; +/** 一条预映射创建change,内部引用用source key依赖,外部引用用明确guard。 */ type CopyLine = {key: string; change: Exclude; dependencies: string[]; externalElements: ExternalElement[]; externalCanvases: ExternalCanvas[]}; /** Internal source: already mapped records, with explicit guards for every retained external reference. */ +/** 兼容完整计划路径:逐change生成有限记录NDJSON,单条≤256KiB、全源≤64MiB;仍返回完整Blob。 */ export async function encodeDuplicateSource(document: Document, registry: Registry, plan: DuplicatePlan, signal: AbortSignal, scheduling:ImportScheduling={}): Promise { const keys = new Map(),parts: BlobPart[] = []; for (const [id,mapped] of Object.entries(plan.elementIds)) keys.set(mapped,`element:${id}`); @@ -27,9 +36,11 @@ export async function encodeDuplicateSource(document: Document, registry: Regist if(signal.aborted)throw fault('copy_aborted');return new Blob(parts,{type:'application/x-ndjson'}); } +/** 共享逐记录转换器:将新映射ID关联到source key,收集owner/端点/声明引用的必要外部闭包。 */ export function duplicateImportLine(document:Document,registry:Registry,change:Exclude,keys:ReadonlyMap):CopyLine { if(!change.after||change.before!==null)throw fault('invalid_duplicate_plan'); const dependencies=new Set(),elements=new Map(),canvases=new Map(); + /** 内部canvas转成依赖;外部canvas沿父链收集guard和入口节点身份,不猜其数据字段。 */ const canvas=(id:Id)=>{ const key=keys.get(id);if(key){dependencies.add(key);return;} let next:Id|undefined=id; @@ -41,6 +52,7 @@ export function duplicateImportLine(document:Document,registry:Registry,change:E next=original.parent?.canvasId; } }; + /** 内部元素作为依赖,外部元素保留同生命周期与owner,同时补其外部canvas链。 */ const element=(id:Id)=>{const key=keys.get(id);if(key){dependencies.add(key);return;}const original=document.elements.get(id);if(!original)throw fault('copy_external_changed',[id]);elements.set(id,{id,incarnation:original.incarnation,ownerCanvasId:original.ownerCanvasId});canvas(original.ownerCanvasId);}; if(change.kind==='canvas'){if(!change.after.parent)throw fault('invalid_duplicate_plan');canvas(change.after.parent.canvasId);element(change.after.parent.entryElementId);} else { @@ -53,6 +65,7 @@ export function duplicateImportLine(document:Document,registry:Registry,change:E return {key,change,dependencies:[...dependencies],externalElements:[...elements.values()],externalCanvases:[...canvases.values()]}; } +/** 校验旧durable预映射行与当前job.stableIds一致,元素新incarnation必须为1,不重新分配ID。 */ function copyLine(line:ImportLine,job:ImportJob):CopyLine { const value=line.value as unknown as CopyLine; if(!['duplicate-ndjson','duplicate-plan'].includes(job.source?.kind??'')||!value||!validJson(value)||typeof value.key!=='string'||!Array.isArray(value.dependencies)||!Array.isArray(value.externalElements)||!Array.isArray(value.externalCanvases)||!value.change||value.change.kind!=='element'&&value.change.kind!=='canvas'||value.change.before!==null||!value.change.after||!Object.hasOwn(job.stableIds,value.key)||job.stableIds[value.key]!==value.change.id||value.change.after.id!==value.change.id||!value.key.startsWith(`${value.change.kind}:`))throw fault('invalid_duplicate_source'); @@ -60,6 +73,7 @@ function copyLine(line:ImportLine,job:ImportJob):CopyLine { return value; } /** Re-check the current reference contract when decoding old durable prepared sources. */ +/** 对当前复制schema重验引用:内部映射须声明依赖,外部引用须explicit preserve且有对应guard。 */ function assertCopyReferences(value:CopyLine,job:ImportJob,registry:Registry){ if(value.change.kind!=='element')return; const record=value.change.after!,schema=registry.components.get(record.componentId)?.copy; @@ -70,12 +84,14 @@ function assertCopyReferences(value:CopyLine,job:ImportJob,registry:Registry){ else if(ref.external!=='preserve'||!(ref.target==='element'?value.externalElements:value.externalCanvases).some(guard=>guard.id===id))throw fault('copy_external_reference',[record.id,id]); } } +/** 形成runner的纯ImportUnit并保留原行作为guard,reserve只核对已有映射,不修改原图。 */ export function prepareDuplicateImport(line:ImportLine,job:ImportJob,reserve:(key:string)=>Id,registry:Registry):ImportUnit { const value=copyLine(line,job);assertCopyReferences(value,job,registry);if(reserve(value.key)!==value.change.id)throw fault('invalid_duplicate_source'); for(const key of value.dependencies)if(!Object.hasOwn(job.stableIds,key))throw fault('invalid_duplicate_source'); return{key:value.key,dependencies:value.dependencies,changes:[value.change],guard:clone(line.value!)}; } /** Check at the synchronous commit boundary, never only when the source was first parsed. */ +/** 提交前按当前文档核对外部节点代次/owner、canvas parent/入口代次;不符即拒绝这一单元/组。 */ export function assertDuplicateExternal(line:ImportLine,job:ImportJob,document:Document,registry:Registry){ const value=copyLine(line,job);assertCopyReferences(value,job,registry); for(const guard of value.externalElements){const record=document.elements.get(guard.id);if(!record||record.incarnation!==guard.incarnation||record.ownerCanvasId!==guard.ownerCanvasId)throw fault('copy_external_changed',[guard.id]);} diff --git a/src/lib/canvas-runtime/import-file-retention.ts b/src/lib/canvas-runtime/import-file-retention.ts index 3669367..ab8854f 100644 --- a/src/lib/canvas-runtime/import-file-retention.ts +++ b/src/lib/canvas-runtime/import-file-retention.ts @@ -1,35 +1,49 @@ +/** + * 导入原源与转换旧源文件的保留/回收事实。文件ID一旦归某job,不能从记录中消失、换成upload资源或复用内容。 + * 软退休可因重pin/覆盖不完整撤销,purge后身份保留且不允许恢复执行;文件删除仍由存储原子批次完成。 + */ import type {Id} from '../canvas-core/types.ts'; import {equal,fault,freeze,readMap} from '../canvas-core/values.ts'; import type {ImportJob,NativeImportSource} from './imports.ts'; import type {RuntimeState} from './contracts.ts'; +/** 版本化保留时钟:首次无引用时间与可选永久清理时间,后者不得早于前者。 */ export type FileRetirement={readonly version:1;readonly unreferencedAt:number;readonly purgedAt?:number}; +/** 源转换后仍登记的旧文件元数据,保留其摘要/退休身份,而不是直接丢旧fileId。 */ export type PreviousImportFile={readonly fileId:Id;readonly byteLength:number;readonly sha256?:string;readonly retirement?:FileRetirement}; +/** 由所有job派生的文件归属,current区分当前源与历史转换源,不另存可漂移归属表。 */ export type ImportFile={readonly jobId:Id;readonly current:boolean;readonly file:PreviousImportFile}; +/** completed/partial/cancelled是本任务终态;paused/preparing不代表源已经不再需要。 */ export const terminalImport=(status:ImportJob['status'])=>['completed','partial','cancelled'].includes(status); +/** 检查退休协议与安全时间值;只是结构合法,不证明引用扫描已完整。 */ export function validFileRetirement(value:unknown):value is FileRetirement { const marker=value as FileRetirement; return !!marker&&marker.version===1&&Number.isSafeInteger(marker.unreferencedAt)&&marker.unreferencedAt>=0&&(marker.purgedAt===undefined||Number.isSafeInteger(marker.purgedAt)&&marker.purgedAt>=marker.unreferencedAt); } +/** 汇总全部当前/旧源,验证job键与ID一致且同fileId不跨job重复,供原子回收检查使用。 */ export function importFiles(jobs:ReadonlyMap|undefined):ReadonlyMap { const files=new Map(); for(const [jobId,job]of jobs??[]){ if(jobId!==job.id)throw fault('import_file_identity_conflict',[jobId]); if(!job.source)continue; + /** 同一个fileId只能出现一次,不能靠current/previous分类掩盖冲突归属。 */ const add=(file:PreviousImportFile,current:boolean)=>{if(files.has(file.fileId))throw fault('import_file_identity_conflict',[file.fileId]);files.set(file.fileId,{jobId:job.id,current,file});}; add(job.source,true);for(const file of job.source.previousFiles??[])add(file,false); } return files; } +/** 当前源仅终态可退休;旧源可独立退休,且同job文件ID互异、元数据和摘要合法。 */ export function validateImportFileRetention(source:NativeImportSource,status:ImportJob['status']) { if(source.retirement!==undefined&&(!validFileRetirement(source.retirement)||!terminalImport(status)))throw fault('invalid_import_file_retirement'); if(source.previousFiles!==undefined&&(!Array.isArray(source.previousFiles)||source.previousFiles.some(file=>!file||typeof file.fileId!=='string'||!file.fileId||!Number.isSafeInteger(file.byteLength)||file.byteLength<0||file.sha256!==undefined&&!/^[a-f0-9]{64}$/.test(file.sha256)||file.retirement!==undefined&&!validFileRetirement(file.retirement))))throw fault('invalid_import_file_retirement'); const ids=[source.fileId,...(source.previousFiles??[]).map(file=>file.fileId)];if(new Set(ids).size!==ids.length)throw fault('import_file_identity_conflict'); } +/** 只改退休标记并保留文件身份;相同标记复用原对象,清标记不删除文件记录。 */ function mark(file:T,retirement:FileRetirement|undefined):T { if(equal(file.retirement,retirement))return file; const {retirement:_,...rest}=file;return freeze(retirement?{...rest,retirement}:rest) as T; } +/** 按fileId批量更新当前/旧源标记,必要时copy-on-write并推进job.generation;不直接删字节。 */ export function markImportFiles(runtime:RuntimeState,markers:ReadonlyMap):RuntimeState { let imports=runtime.imports; for(const [id,job]of runtime.imports??[]){ @@ -42,9 +56,11 @@ export function markImportFiles(runtime:RuntimeState,markers:ReadonlyMap{complete:boolean;pins:ReadonlyMap},reserved:ReadonlySet,referencesChanged=true):RuntimeState { const hasSoft=[...(next.imports?.values()??[])].some(job=>job.source&&[job.source,...(job.source.previousFiles??[])].some(file=>file.retirement&&file.retirement.purgedAt===undefined)); if(previous.imports===next.imports&&previous.resources===next.resources&&previous.resourcePins===next.resourcePins&&!reserved.size&&!hasSoft&&!referencesChanged)return next; + // 旧源可由current降为previous,但不可反向复活;已有摘要/长度/所属job不能在转换时改变。 const before=importFiles(previous.imports),after=importFiles(next.imports); for(const job of next.imports?.values()??[])if(job.source)validateImportFileRetention(job.source,job.status); for(const id of after.keys())if(next.resources.has(id))throw fault('import_file_identity_conflict',[id]); @@ -54,6 +70,7 @@ export function prepareImportFileRetention(previous:RuntimeState,next:RuntimeSta if(old.file.retirement?.purgedAt!==undefined&&!equal(old.file.retirement,entry.file.retirement))throw fault('import_source_retired',[id]); } if(!hasSoft&&!reserved.size&&![...after.values()].some(entry=>entry.file.retirement?.purgedAt!==undefined))return next; + // 覆盖未知或重新引用都清软标记,下一次无引用须重新计保留期,不沿用旧时钟直接删除。 const coverage=pins(),clear=new Map(); for(const [id,entry]of after){ if(entry.file.retirement?.purgedAt!==undefined){if(coverage.pins.has(id))throw fault('import_source_retired',[id]);continue;} diff --git a/src/lib/canvas-runtime/imports.ts b/src/lib/canvas-runtime/imports.ts index 1443357..da67af9 100644 --- a/src/lib/canvas-runtime/imports.ts +++ b/src/lib/canvas-runtime/imports.ts @@ -1,3 +1,8 @@ +/** + * 渐进导入的持久job合同、NDJSON解释器和合作式runner。解析得到稳定ID/locator,依赖闭合后原子加入文档, + * job进度与文档同Root发布;暂停/取消保留已提交内容,失败逐记录或按不可分依赖组报告partial。 + * runner不拥有第二份持久job store:ports.getJob是事实源,所有await后的业务采纳受generation/身份检查。 + */ import {dataArray,dataRecord} from '../canvas-core/data-containers.ts'; import {createImportBatches} from './import-batches.ts'; import {createImportIdentities} from './import-identities.ts'; @@ -10,27 +15,41 @@ import type { Change, Id, Json, Result, Target, Point, Availability } from '../c import { selectImportGroups } from './import-groups.ts'; import { clone, equal, fault, freeze, validJson } from '../canvas-core/values.ts'; +/** 源的一步读取结果:cursor/nextCursor单位由parserVersion约定,value/error/skip由runner分别处理。 */ export type ImportLine = { readonly cursor: number; readonly nextCursor: number; readonly key: string; readonly value?: Json; readonly error?: string; readonly skip?: boolean }; +/** 同fingerprint/parserVersion下可重读稳定游标,null表示EOF;不等于可从任意字节重新解释。 */ export type ImportSource = { readonly fingerprint: string; readonly parserVersion: number; read(cursor: number, signal: AbortSignal): Promise }; +/** 解码后的纯变化单元,dependencies用source key表达,guard保留提交时复核的外部身份。 */ export type ImportUnit = { readonly key: string; readonly dependencies: readonly string[]; readonly changes: readonly Change[]; readonly guard?: Json }; +/** 可持久pending只保存定位/依赖/摘要/成本,不保存无限增长的解码payload副本。 */ 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[] }; +/** partial加载准入时固定源token、逻辑时点与必要canvas集合,和源字节身份分别校验。 */ export type LoadingImportAdmission={readonly version:1;readonly sourceToken:string;readonly operationSequence:number;readonly requiredCanvasIds:readonly Id[]}; +/** 源文件身份、格式、目标入口及完整性/退休信息;copy计划与普通原生记录使用显式kind区分。 */ 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} }; +/** 已加入新ID的生命周期绑定;canvas还绑定入口,防用户删除/恢复后被后续尾批误认。 */ export type ImportBinding = {readonly kind:'element';readonly incarnation:number}|{readonly kind:'canvas';readonly entryElementId?:Id}; +/** 唯一持久任务记录;preparation是准备阶段游标,cursor/eof是解析进度,committed才是已加入前缀。 */ export type ImportJob = { readonly version: 1 | 2 | 3; readonly id: Id; readonly projectId: Id; readonly sourceFingerprint: string; readonly parserVersion: number; readonly source?: NativeImportSource; + /** 复制规划版本独立于job/source/parser版本,allocationIds必须与planner稳定映射一致。 */ readonly preparation?: {readonly version:1|2|3|4;readonly planner?:DuplicateCheckpoint;readonly registrySignature:string;readonly selection:readonly Target[];readonly worldOffset:Point;readonly allocationIds:readonly Id[];readonly steps:number}; readonly generation: number; readonly status: 'preparing' | 'paused' | 'running' | 'completed' | 'partial' | 'cancelled'; readonly cursor: number; readonly eof: boolean; readonly bindings?: Readonly>; + /** 稳定分配跨重开复用;pending可重读,committed不会因后续失败/取消被重新创建。 */ readonly stableIds: Readonly>; readonly pending: readonly ImportLocator[]; readonly committedSourceKeys: readonly string[]; readonly failures: readonly ImportFailure[]; readonly error?: string; }; +/** integrity-v2除零前缀取消外,缺完整SHA或有完整性错误都继续阻止发送;解析EOF不等于源已封存。 */ export const importNeedsSeal=(job:ImportJob)=>job.source?.integrityVersion===2&&!(job.status==='cancelled'&&!job.committedSourceKeys.length)&&(!job.source.sha256||!!job.source.integrityError); +/** 验证基本身份并创建v3 paused空任务;source旧调用缺省integrityVersion=1,不自动伪造摘要证明。 */ export function createImportJob(input: Pick): 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: [] }); } +/** 校验job及独立解释器、准备/加入状态、源证明、映射/绑定/pending/failure间的一致性。 */ export function validateImportJob(job: ImportJob): void { if(!job||typeof job!=='object'||!validJson(job)||!job.stableIds||typeof job.stableIds!=='object'||Array.isArray(job.stableIds)||![job.pending,job.committedSourceKeys,job.failures].every(Array.isArray)||job.error!==undefined&&typeof job.error!=='string')throw fault('invalid_import_job'); if (job.version !== 1 && job.version !== 2 && job.version !== 3) throw fault('import_interpreter_unavailable'); @@ -40,6 +59,7 @@ export function validateImportJob(job: ImportJob): void { if(job.source?.kind==='native-document'&&![4,5].includes(job.parserVersion))throw fault('import_interpreter_unavailable'); if(job.source?.kind==='native-json-array'&&job.parserVersion!==3)throw fault('import_interpreter_unavailable'); if(job.source?.loadingAdmission){const proof=job.source.loadingAdmission;if(proof.version!==1||typeof proof.sourceToken!=='string'||!proof.sourceToken||proof.sourceToken.length>256||!Number.isSafeInteger(proof.operationSequence)||proof.operationSequence<0||!Array.isArray(proof.requiredCanvasIds)||!proof.requiredCanvasIds.length||proof.requiredCanvasIds.some(id=>typeof id!=='string'||!id)||new Set(proof.requiredCanvasIds).size!==proof.requiredCanvasIds.length)throw fault('loading_import_interpreter_unavailable');} + // source验证先处理文件退休/格式,再分别核验integrity1完整摘要与integrity2连续块证明。 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'); @@ -52,6 +72,7 @@ export function validateImportJob(job: ImportJob): void { 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'); } + // 准备任务尚无导入cursor/pending/committed;已预留ID、steps与planner检查点必须严格对应。 if(job.source?.kind==='duplicate-snapshot'){ const p=job.preparation; if(!p||![1,2,3,4].includes(p.version)||typeof p.registrySignature!=='string'||!p.registrySignature||!Array.isArray(p.selection)||!p.selection.length||p.selection.some(t=>!t||typeof t.id!=='string'||!t.id||!Number.isSafeInteger(t.incarnation)||t.incarnation<1)||!p.worldOffset||![p.worldOffset.x,p.worldOffset.y].every(Number.isFinite)||!Array.isArray(p.allocationIds)||p.allocationIds.some(id=>typeof id!=='string'||!id)||new Set(p.allocationIds).size!==p.allocationIds.length||!Number.isSafeInteger(p.steps)||p.steps<0||!['preparing','paused','cancelled'].includes(job.status)||job.cursor!==0||job.eof||job.pending.length||job.committedSourceKeys.length||Object.keys(job.stableIds).length||!job.source.copy||!Number.isSafeInteger(job.source.copy.sourceRevision)||job.source.copy.sourceRevision<0)throw fault('invalid_copy_preparation'); @@ -59,9 +80,11 @@ export function validateImportJob(job: ImportJob): void { if(p.planner){validateDuplicateCheckpoint(p.planner);if(![2,3,4].includes(p.version)||p.planner.sourceToken!==job.source.sha256||p.planner.schemaToken!==p.registrySignature||p.planner.projectId!==job.projectId||!equal(p.planner.selection,p.selection)||!equal(p.planner.offset,p.worldOffset)||p.planner.steps!==p.steps||!equal([...p.planner.elementIds,...p.planner.canvasIds].map(pair=>pair[1]),p.allocationIds))throw fault('invalid_copy_preparation');} else if(p.version!==1&&(p.steps!==0||p.allocationIds.length))throw fault('invalid_copy_preparation'); }else if(job.status==='preparing'||job.preparation!==undefined)throw fault('invalid_copy_preparation'); + // source key在committed和pending中唯一,所有stable目标ID互异;已加入源项必须有生命周期binding。 const keys=new Set(),ids=Object.values(job.stableIds),stableIdSet=new Set(ids); if (stableIdSet.size!==ids.length || ids.some(id=>typeof id!=='string'||!id) || Object.keys(job.stableIds).some(key=>!key)) throw fault('invalid_import_job'); if(job.source&&['duplicate-ndjson','duplicate-plan'].includes(job.source.kind)&&(!job.source.copy||new Set(job.source.copy.roots).size!==job.source.copy.roots.length||job.source.copy.roots.some(id=>!stableIdSet.has(id))))throw fault('invalid_import_source'); + // lazy复制源必须停在emit起点,全部映射已固定但未生成输出;ordinal不可越过order+edges总数。 if(job.source?.kind==='duplicate-plan'){ const plan=job.source.plan;if(!plan)throw fault('invalid_copy_stream_checkpoint');validateDuplicateCheckpoint(plan); const ids=Object.fromEntries([...plan.elementIds.map(([id,mapped])=>[`element:${id}`,mapped]),...plan.canvasIds.map(([id,mapped])=>[`canvas:${id}`,mapped])]); @@ -71,22 +94,27 @@ export function validateImportJob(job: ImportJob): void { for(const key of job.committedSourceKeys){if(!key||keys.has(key))throw fault('invalid_import_job');keys.add(key);} if(job.source)for(const key of job.committedSourceKeys){const id=job.stableIds[key],binding=job.bindings?.[id];if(!Object.hasOwn(job.stableIds,key)||!binding||!key.startsWith(`${binding.kind}:`)||binding.kind==='canvas'&&(!binding.entryElementId||job.bindings?.[binding.entryElementId]?.kind!=='element'))throw fault('invalid_import_job');} for(const unit of job.pending){if(!unit||typeof unit.key!=='string'||!unit.key||!Array.isArray(unit.dependencies)||keys.has(unit.key)||!Number.isSafeInteger(unit.cursor)||unit.cursor<0||!Number.isSafeInteger(unit.nextCursor)||unit.nextCursor<=unit.cursor||unit.nextCursor>job.cursor||!(/^[a-f0-9]{64}$/.test(unit.digest))||!Number.isSafeInteger(unit.byteLength)||unit.byteLength<1||!Number.isSafeInteger(unit.changeCount)||unit.changeCount<1||unit.changeCount>10000||new Set(unit.dependencies).size!==unit.dependencies.length||unit.dependencies.some(key=>typeof key!=='string'||!key))throw fault('invalid_import_job');keys.add(unit.key);} + // partial必须读到EOF,completed还要求无pending/失败;paused保留游标与失败供显式处理。 if(job.status==='partial'&&!job.eof)throw fault('invalid_import_job'); if(job.status==='completed'&&(!job.eof||job.pending.length||job.failures.length))throw fault('invalid_import_job'); for(const failure of job.failures)if(!failure||typeof failure.key!=='string'||!failure.key||typeof failure.code!=='string'||!failure.code||!Number.isSafeInteger(failure.cursor)||failure.cursor<0||failure.cursor>job.cursor)throw fault('invalid_import_job'); } +/** 默认真实任务让出,不是只用Promise.resolve制造微任务而继续占住UI线程。 */ const yieldTask=()=>new Promise(resolve=>setTimeout(resolve,0)); /** Seekable bounded NDJSON parser. Fingerprint must identify the host's immutable, durable byte source. */ +/** parser1按换行边界返回一条记录,最大记录与读取chunk分开约束,空白行skip但仍推进cursor。 */ export function createNdjsonImportSource(options:{bytes:Blob;fingerprint:string;readBytes?:ImportByteReader;maxRecordBytes?:number;chunkBytes?:number}&ImportScheduling):ImportSource { const max=options.maxRecordBytes??256*1024,chunk=options.chunkBytes??16*1024; if(!(options.bytes instanceof Blob)||!options.fingerprint||!Number.isSafeInteger(max)||max<1||!Number.isSafeInteger(chunk)||chunk<1||chunk>max)throw fault('invalid_import_source'); const sourceBytes=options.bytes; let cache={offset:-1,bytes:new Uint8Array()}; return Object.freeze({fingerprint:options.fingerprint,parserVersion:1, + /** 从已保存byte游标扫描到LF/EOF,超大行丢缓存但继续定位下一行;坏UTF8/JSON返回单项error。 */ async read(cursor:number,signal:AbortSignal):Promise{ if(!Number.isSafeInteger(cursor)||cursor<0||cursor>sourceBytes.size)throw fault('invalid_import_cursor'); if(signal.aborted)throw fault('import_aborted');if(cursor===sourceBytes.size)return null; + // 保存有界byte切片;cached块跨行复用,不能每行重新读同一16KiB源块。 const parts:Uint8Array[]=[],start=cursor,budget=createImportBudget(signal,options);let total=0,oversized=false; while(cursorImportJob; source:ImportSource;getJob():ImportJob;createId():Id; /** Synchronous compare-and-publish; never write a second private job store. */ @@ -127,28 +157,38 @@ export type ImportRunnerPorts={ }; /** Cooperative producer/consumer. Pending work stores seekable locators, not unbounded decoded payloads. */ +/** 创建单job执行器;先校验源解释器与完整性能力,duplicate-snapshot准备任务走另一专用路径。 */ 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.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'); + // 稳定ID分配与自适应batch反馈为runner临时索引,持久事实仍保存在getJob返回的记录中。 const identities=createImportIdentities(),allocateIdentity=()=>ports.createId(); const batches=createImportBatches(budget,maxBatch); + // active防并发双pump,holds叠加临时冻结,restart记录活跃执行结束后需继续的新请求。 let active:Promise|undefined,controller:AbortController|undefined,closed=false,restart=false;const holds=new Set(); + /** 错误上报异常隔离,不让日志失败改变已加入内容或已持久游标。 */ const report=(error:unknown)=>{try{ports.onError(error);}catch{/* Reporting cannot change imported content. */}}; + /** 每次读事实都检查job/project/fingerprint/parser身份,旧runner不能采纳新任务的同名状态。 */ 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;} + /** 只允许checkpointSource补readProof/SHA/integrityError;其它source身份和整个job主体都不可偷偷改变。 */ 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'); + /** 排除允许推进的完整性字段后比对源身份,防校验钩子换file/target/复制计划。 */ 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; } + /** 合成下一代候选job并捕获源证明;不在这里写私有任务store。 */ function next(job:ImportJob,patch:Partial):ImportJob{const value={...job,...patch,version:3 as const,generation:job.generation+1};if(value.error===undefined)delete value.error;const checked=checkpointEvidence(value);return checked===value?dataRecord(freeze(value)):freeze(checked);} + /** 以expectedGeneration compare-and-publish,再核对getJob确实返回该候选,防宿主半接受/丢字段。 */ function publish(job:ImportJob,patch:Partial){const value=next(job,patch);ports.publish(value,job.generation);if(!equal(current(),value))throw fault('invalid_import_publish');return value;} + /** 用stableIds解码单元,校验纯JSON/非空changes/依赖/成本,新创建ID必须来自reserve分配集合。 */ function prepare(line:ImportLine,job:ImportJob){ const ids=identities(job.stableIds,allocateIdentity); const unit=ports.prepare(line,job,ids.reserve); @@ -157,16 +197,22 @@ export function createImportRunner(ports:ImportRunnerPorts){ if(new TextEncoder().encode(JSON.stringify(unit)).byteLength>maxBytes)throw fault('import_unit_capacity'); return{unit:freeze(clone(unit)),ids:ids.finish()}; } + /** optimisticLocal可按宿主背压延后磁盘,但网络仍须过durable门;默认每次直接等待persist。 */ const checkpoint=async(stage:'parsed'|'committed'='committed')=>{if(ports.optimisticLocal&&ports.awaitDurabilityBudget)await ports.awaitDurabilityBudget(stage);else await ports.persist();}; + /** 对含guard的完整ImportUnit取SHA,恢复重读时校验转换结果而不只比较源key。 */ const digest=async(unit:ImportUnit)=>[...new Uint8Array(await crypto.subtle.digest('SHA-256',new TextEncoder().encode(JSON.stringify(unit))))].map(value=>value.toString(16).padStart(2,'0')).join(''); + /** 为成功创建记录保存代次/入口绑定,并把source key追加committed,同时移除对应pending。 */ function completed(job:ImportJob,units:readonly ImportUnit[],pending:readonly ImportLocator[]){ const bindings:Record=Object.assign(Object.create(null),job.bindings??{}); 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:dataArray(freeze(pending)),committedSourceKeys:dataArray(freeze([...job.committedSourceKeys,...units.map(unit=>unit.key)])),bindings:dataRecord(freeze({...bindings}))}); } + /** 同步提交前再查generation,提交后核对job与Result一致;只有applied样本用于batch成本反馈。 */ 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),elapsed=Math.max(0,now()-start);if(!['applied','rejected'].includes(result?.status)||!equal(current(),result.status==='applied'?value:base))throw fault('invalid_import_commit');if(result.status==='applied')batches.record(units.length,elapsed);return result;} + /** 合作式生产/消费循环:先消化闭合pending,再按有界slice解析新记录,EOF最后确认partial/completed。 */ async function pump(signal:AbortSignal){ let frame=now(),didWork=false; + /** 异步只读beforeCommit后重查closed/abort/hold/generation/status,迟到准备不能提交过期候选。 */ async function attempt(units:readonly ImportUnit[],job:ImportJob,proposed:ImportJob):Promise{ let availability:Availability={enabled:true}; if(ports.beforeCommit)try{availability=await ports.beforeCommit(units,job,signal);}catch(error){availability={enabled:false,code:(error as {code?:string}).code??'import_preparation_failed',targets:[]};} @@ -174,11 +220,16 @@ export function createImportRunner(ports:ImportRunnerPorts){ if(!availability.enabled)return{status:'rejected',code:availability.code??'import_preparation_failed',targets:availability.targets??[]}; return commit(units,job,proposed); } + /** 预读将异常转成结果以便接住迟到拒绝,不让并行准备出现未处理Promise。 */ type ReadResult={kind:'line';line:ImportLine|null}|{kind:'error';error:unknown}; let reading:{cursor:number;result:Promise}|undefined; + // 仅内存有界热缓存:按maxBatch和maxBytes淘汰,pending持久状态始终是locators而非这些payload。 const decoded=new Map();let decodedBytes=0; + /** 更新热单元/字节水位,超任一预算淘汰最旧项;淘汰后可按locator重读,不丢业务事实。 */ function cache(key:string,value:{unit:ImportUnit;digest:string;bytes:number}){const previous=decoded.get(key);if(previous)decodedBytes-=previous.bytes;decoded.delete(key);decoded.set(key,value);decodedBytes+=value.bytes;while(decoded.size>maxBatch||decodedBytes>maxBytes){const[key,value]=decoded.entries().next().value!;decoded.delete(key);decodedBytes-=value.bytes;}} + /** 提交/失败处理后及时丢热payload,避免已加入内容继续占runner缓存。 */ function forget(key:string){const value=decoded.get(key);if(value){decodedBytes-=value.bytes;decoded.delete(key);}} + /** 同一游标共享一次预读;已有普通ready项时与0ms timer竞速,可先flush显示而不等慢尾部。 */ async function read(cursor:number,flushReady:boolean):Promise{ if(reading&&reading.cursor!==cursor)throw fault('invalid_import_cursor'); reading??={cursor,result:Promise.resolve().then(()=>ports.source.read(cursor,signal)).then(line=>({kind:'line' as const,line}),error=>({kind:'error' as const,error}))}; @@ -194,11 +245,13 @@ export function createImportRunner(ports:ImportRunnerPorts){ didWork=true; const committed=new Set(job.committedSourceKeys); const batchLimit=batches.limit; + // 组选择把循环引用SCC视为不可拆单元;chosen源顺序与依赖可用性共同约束原子提交。 const groups=selectImportGroups(job.pending,committed,{units:batchLimit,bytes:maxBytes,changes:10000}),chosen=groups.flat(); if(chosen.length){ const units:ImportUnit[]=[]; for(const locator of chosen){ const cached=decoded.get(locator.key); + // 热缓存命中须与持久digest相同;否则seek重读并核对nextCursor/依赖/映射/成本及完整unit摘要。 if(cached&&cached.digest===locator.digest){units.push(cached.unit);continue;} // A source need not support concurrent reads. Finish any speculative read before seeking. if(reading)await reading.result; @@ -212,6 +265,7 @@ export function createImportRunner(ports:ImportRunnerPorts){ } let proposed=completed(job,units,job.pending.filter(unit=>!chosen.includes(unit))); let result=await attempt(units,job,proposed);if(!result)return; + // 大批拒绝仅隔离不同SCC:同一循环组要么整组加入,要么整组失败,不能留下半份引用。 if(result.status==='rejected'){ // Reject a reference cycle as a whole; isolate only independent SCCs. const byKey=new Map(units.map(unit=>[unit.key,unit])); @@ -228,21 +282,25 @@ export function createImportRunner(ports:ImportRunnerPorts){ for(const unit of units)forget(unit.key); continue; } + // 终结前先等最后文档/源进度落盘;剩余缺依赖显式报告,不能因为parser结束就直接completed。 if(job.eof){ // Completion must never overtake the final document/source durability barrier. await ports.persist();if(closed||signal.aborted||current().generation!==job.generation)return; const unresolved=job.pending.filter(unit=>!job.failures.some(failure=>failure.key===unit.key&&failure.code==='unresolved_dependency')).map(unit=>({key:unit.key,cursor:unit.cursor,code:'unresolved_dependency',dependencies:unit.dependencies.filter(key=>!committed.has(key))})); publish(job,{status:job.failures.length||unresolved.length?'partial':'completed',failures:[...job.failures,...unresolved]});await ports.persist();return; } + // 等待依赖超容量是可恢复paused,不继续无限解析并在内存里堆积整个文件。 if(job.pending.length>=maxPending||new TextEncoder().encode(JSON.stringify(job.pending)).byteLength>=maxBytes){publish(job,{status:'paused',error:'dependency_capacity'});await ports.persist();return;} // Stage a bounded parser slice, then publish its cursor/IDs/locators once. This is // unpublished transaction preparation, never another durable task state store. + // staged是一次未发布的解析slice,不是第二份任务事实;最终cursor/ID/locator一起发布。 const base=job;let staged=job,records=0; while(!staged.eof&&recordsunit.dependencies.every(key=>committed.has(key))); const result=await read(staged.cursor,ready); if(closed||signal.aborted||current().generation!==base.generation)return; if(result.kind==='flush')break; + // 已有ready前缀先flush,再报告源错误;不能为追尾部而延迟已可展示内容。 if(result.kind==='error'){ if(ready){reading={cursor:staged.cursor,result:Promise.resolve(result)};break;} if(staged!==base){const{generation,...patch}=staged;publish(base,patch);}throw result.error; @@ -253,9 +311,11 @@ export function createImportRunner(ports:ImportRunnerPorts){ let patch:Partial; if(line.skip)patch={cursor:line.nextCursor}; else if(line.error)patch={cursor:line.nextCursor,failures:[...staged.failures,{key:line.key,cursor:line.cursor,code:line.error}]}; + // 单记录准备失败记录failure并推进cursor;源级读取错误走上面的抛出路径,不伪装单项坏记录。 else try{ const prepared=prepare(line,staged),key=prepared.unit.key; if(staged.committedSourceKeys.includes(key)||staged.pending.some(unit=>unit.key===key)||staged.failures.some(failure=>failure.key===key))throw fault('duplicate_import_source_key'); + // 持久pending只记定位/摘要/依赖与成本,stableIds同次保存;旧源变化不能在恢复重读时蒙混过关。 const locator=dataRecord(freeze({key,cursor:line.cursor,nextCursor:line.nextCursor,dependencies:dataArray(prepared.unit.dependencies),digest:await digest(prepared.unit),byteLength:new TextEncoder().encode(JSON.stringify(prepared.unit)).byteLength,changeCount:prepared.unit.changes.length})); cache(key,{unit:prepared.unit,digest:locator.digest,bytes:locator.byteLength}); patch={cursor:line.nextCursor,stableIds:prepared.ids,pending:dataArray([...staged.pending,locator])}; @@ -269,6 +329,7 @@ export function createImportRunner(ports:ImportRunnerPorts){ } } + /** 单执行句柄:异常把仍属于自己的运行任务暂停并尽量保存;finally清索引并按restart条件继续。 */ function launch(){ if(closed||holds.size)return Promise.resolve();if(active){restart=true;return active;} restart=false;controller=new AbortController(); @@ -280,11 +341,17 @@ export function createImportRunner(ports:ImportRunnerPorts){ return active; } return Object.freeze({ + /** paused可显式启动,completed/partial/cancelled不会自动续跑;新一轮清job.error但不清历史内容。 */ start(){if(closed)throw fault('import_runner_closed');const job=current();if(['completed','partial','cancelled'].includes(job.status))return Promise.resolve();if(job.status!=='running')publish(job,{status:'running',error:undefined});return launch();}, + /** 临时叠加冻结并abort旧pump,释放最后hold后仅在job仍running时续跑,不制造新的任务身份。 */ hold(){const token=Symbol();holds.add(token);controller?.abort();return()=>{if(!holds.delete(token)||holds.size||closed)return;if(current().status==='running')void launch();};}, + /** running→paused并持久化,保留cursor/映射/pending/已加入前缀;迟到结果由代次检查挡住。 */ async pause(){const job=current();if(job.status==='running'){controller?.abort();publish(job,{status:'paused'});await ports.persist();}}, + /** running/paused→cancelled,保留成功前缀与失败报告,不执行默认整批回滚。 */ async cancel(){const job=current();if(['running','paused'].includes(job.status)){controller?.abort();publish(job,{status:'cancelled',error:undefined});await ports.persist();}}, + /** 等当前pump(及其已触发重启)收尾,不代表未来任务/网络同步都已完成。 */ async settle(){while(active)await active;}, + /** 关闭本runner并等活跃pump退出,不把持久job改成cancelled,重开仍可按其事实恢复。 */ async close(){closed=true;controller?.abort();while(active)await active;}, }); }