docs(canvas): explain import execution and lazy copy source recovery
This commit is contained in:
@@ -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<void>|undefined{
|
||||
if(signal.aborted)throw fault('copy_aborted');
|
||||
try{return budget.step(units)?.catch(fail);}catch(error){return fail(error);}
|
||||
|
||||
@@ -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<Blob>{
|
||||
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<ElementRecord>};
|
||||
/** 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<IndexedCopySnapshot>{
|
||||
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<Id,number>(),locators=new Map<Id,{offset:number;byteLength:number}>();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<string,Json>;
|
||||
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<number,Uint8Array>(),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);
|
||||
}});
|
||||
|
||||
@@ -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<Blob>{
|
||||
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<Document>{
|
||||
const budget=createCopyBudget(signal,scheduling),source=createNdjsonImportSource({bytes,fingerprint:'captured-copy-snapshot'}),canvases:CanvasRecord[]=[],elements:ElementRecord[]=[],tombstones=new Map<string,number>();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};}));
|
||||
|
||||
@@ -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<ElementRecord>):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<Id,string>(),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<T extends ElementRecord|CanvasRecord>(base:ReadonlyMap<Id,T>,reverse:ReadonlyMap<Id,Id>,kind:'element'|'canvas'):ReadonlyMap<Id,T>{
|
||||
// 未映射ID沿用原表,新ID转换源记录;缓存淘汰只影响计算成本,不改变稳定映射/源数据。
|
||||
const cache=new Map<Id,T>();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<Id,T>={get,has:id=>base.has(id)||reverse.has(id),size:base.size+reverse.size,entries,keys:function*():Generator<Id,undefined>{for(const[id]of entries())yield id;return undefined;},values:function*():Generator<T,undefined>{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=cursor<state.order.length?duplicateTreeChange(document,registry,task.kind,original,elements,canvases,copiedCanvases,state.offset):duplicateEdgeChange(document,registry,original as ElementRecord,elements,canvases,projected),line=duplicateImportLine(document,registry,change,keys);
|
||||
if(new TextEncoder().encode(JSON.stringify(line)).byteLength>256*1024)throw fault('record_too_large');
|
||||
|
||||
@@ -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<void>;
|
||||
@@ -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<DuplicatePlan> {
|
||||
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<void>(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<DuplicateScheduling,'checkpoint'>&{checkpoint?:(state:DuplicateCheckpoint)=>Promise<void>}={}):Promise<DuplicatePlan>{
|
||||
const budget=scheduling.frameBudgetMs??6,limit=scheduling.maxStepsPerSlice??512,now=scheduling.now??(()=>performance.now()),yieldTask=scheduling.yieldTask??(()=>new Promise<void>(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<DuplicateScheduling,'checkpoint'>&{checkpoint?:(state:DuplicateCheckpoint)=>Promise<void>;checkpointIntervalMs?:number;maxStepsPerCheckpoint?:number}={}):Promise<DuplicateCheckpoint>{
|
||||
const budget=scheduling.frameBudgetMs??6,limit=scheduling.maxStepsPerSlice??512,now=scheduling.now??(()=>performance.now()),yieldTask=scheduling.yieldTask??(()=>new Promise<void>(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(count<limit&&now()-start<budget);
|
||||
if(scheduling.checkpoint){
|
||||
// 避免每次UI让出都必须编码/落盘一个大检查点,同时为长准备保留明确的恢复边界。
|
||||
const progress=planner.progress(),at=now(),elapsed=at-savedAt;
|
||||
if(!interval||progress.phase!==savedPhase||!savedAllocated&&progress.allocated>0||!Number.isFinite(elapsed)||elapsed<0||elapsed>=interval||progress.steps-savedSteps>=maxCheckpointSteps){
|
||||
await scheduling.checkpoint(planner.checkpoint());
|
||||
|
||||
@@ -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<Change,{kind:'field'}>; 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<Blob> {
|
||||
const keys = new Map<Id,string>(),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<Change,{kind:'field'}>,keys:ReadonlyMap<Id,string>):CopyLine {
|
||||
if(!change.after||change.before!==null)throw fault('invalid_duplicate_plan');
|
||||
const dependencies=new Set<string>(),elements=new Map<Id,ExternalElement>(),canvases=new Map<Id,ExternalCanvas>();
|
||||
/** 内部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]);}
|
||||
|
||||
@@ -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<Id,ImportJob>|undefined):ReadonlyMap<Id,ImportFile> {
|
||||
const files=new Map<Id,ImportFile>();
|
||||
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<T extends PreviousImportFile>(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<Id,FileRetirement|undefined>):RuntimeState {
|
||||
let imports=runtime.imports;
|
||||
for(const [id,job]of runtime.imports??[]){
|
||||
@@ -42,9 +56,11 @@ export function markImportFiles(runtime:RuntimeState,markers:ReadonlyMap<Id,File
|
||||
return imports===runtime.imports?runtime:Object.freeze({...runtime,imports:readMap(imports!)});
|
||||
}
|
||||
/** A source file cannot disappear from its owning job, become an asset or regain execution after purge. */
|
||||
/** 每次Root发布前校验源身份单调性并取消不安全软退休;已purge仍被pin明确拒绝。 */
|
||||
export function prepareImportFileRetention(previous:RuntimeState,next:RuntimeState,pins:()=>{complete:boolean;pins:ReadonlyMap<Id,{has(owner:string):boolean}>},reserved:ReadonlySet<Id>,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<Id,undefined>();
|
||||
for(const [id,entry]of after){
|
||||
if(entry.file.retirement?.purgedAt!==undefined){if(coverage.pins.has(id))throw fault('import_source_retired',[id]);continue;}
|
||||
|
||||
@@ -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<ImportLine | null> };
|
||||
/** 解码后的纯变化单元,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<Record<Id,ImportBinding>>;
|
||||
/** 稳定分配跨重开复用;pending可重读,committed不会因后续失败/取消被重新创建。 */
|
||||
readonly stableIds: Readonly<Record<string, Id>>; 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, '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: [] });
|
||||
}
|
||||
/** 校验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<string>(),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<void>(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<ImportLine|null>{
|
||||
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(cursor<sourceBytes.size){
|
||||
if(signal.aborted)throw fault('import_aborted');
|
||||
@@ -107,7 +135,9 @@ export function createNdjsonImportSource(options:{bytes:Blob;fingerprint:string;
|
||||
});
|
||||
}
|
||||
|
||||
/** 宿主协调合同:prepare纯解码,beforeCommit可异步只读准备,commit同步发布文档+job或均不发布。 */
|
||||
export type ImportRunnerPorts={
|
||||
/** 引擎拥有的源块证明/SHA采纳钩子,不授予任意调用者改写job执行进度的权限。 */
|
||||
checkpointSource?:(job:ImportJob)=>ImportJob;
|
||||
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<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. */}};
|
||||
/** 每次读事实都检查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>):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<ImportJob>){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<Id,ImportBinding>=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<Result|undefined>{
|
||||
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<ReadResult>}|undefined;
|
||||
// 仅内存有界热缓存:按maxBatch和maxBytes淘汰,pending持久状态始终是locators而非这些payload。
|
||||
const decoded=new Map<string,{unit:ImportUnit;digest:string;bytes:number}>();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<ReadResult|{kind:'flush'}>{
|
||||
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&&records<maxBatch&&staged.pending.length<maxPending&&new TextEncoder().encode(JSON.stringify(staged.pending)).byteLength<maxBytes){
|
||||
const ready=staged.pending.some(unit=>unit.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<ImportJob>;
|
||||
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;},
|
||||
});
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user