diff --git a/src/lib/canvas-runtime/import-batches.ts b/src/lib/canvas-runtime/import-batches.ts index 8808597..8dd32da 100644 --- a/src/lib/canvas-runtime/import-batches.ts +++ b/src/lib/canvas-runtime/import-batches.ts @@ -1,10 +1,19 @@ +/** + * 导入runner内的自适应批大小反馈，不持久化进job，也不改变记录/ID或依赖规则。 + * 从一条开始试探，超时缩小、满批且便宜才增长；每次原子发布仍可能有不可抢占固定成本。 + */ /** Runner-owned feedback, not persisted job state. Probe bounded batches to * distinguish fixed publication cost from the marginal cost of another unit. */ +/** budget/max由runner给出，用实际已提交units和耗时更新下一批软上限。 */ export function createImportBatches(budget:number,max:number){ + // 初始未观测时保守为1；observed只表示收到反馈，不是源校验或导入已完成。 let limit=1,observed=false; return Object.freeze({ + /** 下次组选择的单位软预算，依赖SCC仍可能需要一次超过它。 */ get limit(){return limit;}, + /** 是否已有至少一次发布成本样本，用于runner的调度决策。 */ get observed(){return observed;}, + /** 超时按比例减小并至少降一档；不足当前limit的尾批不能证明更大批也便宜。 */ record(units:number,elapsed:number){ observed=true; if(elapsed>budget){limit=Math.max(1,Math.min(limit-1,Math.floor(units*budget/elapsed)));return;} diff --git a/src/lib/canvas-runtime/import-byte-reader.ts b/src/lib/canvas-runtime/import-byte-reader.ts index 4102fc6..9cef1a1 100644 --- a/src/lib/canvas-runtime/import-byte-reader.ts +++ b/src/lib/canvas-runtime/import-byte-reader.ts @@ -1,17 +1,27 @@ +/** + * 解析器的统一精确字节读取边界。默认读不可变 Blob，也可注入带块证明的 reader。 + * 这里检查范围/长度与取消，不负责整源SHA、JSON语法或业务记录；源级读取失败不能降格为可跳过的坏记录。 + */ import {fault} from '../canvas-core/values.ts'; +/** 会破坏整个源读取可信性的错误集合，供上层区分单记录失败与源级暂停/失败。 */ const failures=new Set(['import_source_changed','invalid_import_byte_range','import_read_capacity','import_source_closed','import_source_read_failed','invalid_import_read_proof','import_read_proof_interpreter_unavailable']); +/** 仅按稳定领域错误码分类，不依赖展示 message 或错误文本包含某个词。 */ export const isImportByteFailure=(error:unknown)=>failures.has((error as {code?:string}|undefined)?.code??''); /** Optional immutable-source port. Implementations return exactly the requested bytes. */ +/** 可选只读端口声明完整长度，必须返回所请求区间的精确字节，不能截断或换源。 */ export type ImportByteReader={readonly byteLength:number;read(start:number,end:number,signal:AbortSignal):Promise}; +/** 校验安全整数范围，end 可截至EOF；读取前后查取消，外来 reader 结果额外复制以隔离缓冲复用。 */ export async function readImportRange(bytes:Blob,reader:ImportByteReader|undefined,start:number,end:number,signal:AbortSignal):Promise>{ if(signal.aborted)throw fault('import_aborted'); if(!Number.isSafeInteger(start)||!Number.isSafeInteger(end)||start<0||endbytes.size||reader&&reader.byteLength!==bytes.size)throw fault('invalid_import_byte_range'); const limit=Math.min(end,bytes.size);let value:Uint8Array; + // 保留已识别源错误，其它I/O异常统一为source_read_failed；取消优先返回import_aborted。 try{value=reader?await reader.read(start,limit,signal):new Uint8Array(await bytes.slice(start,end).arrayBuffer());} catch(error){if(signal.aborted)throw fault('import_aborted');if(isImportByteFailure(error))throw error;throw fault('import_source_read_failed');} if(signal.aborted)throw fault('import_aborted'); + // 长度或响应类型不符视为源改变，不能让解析器在缺字节内容上继续产出业务修改。 if(!(value instanceof Uint8Array)||value.byteLength!==limit-start)throw fault('import_source_changed'); return reader?new Uint8Array(value):value as Uint8Array; } diff --git a/src/lib/canvas-runtime/import-document-stream.ts b/src/lib/canvas-runtime/import-document-stream.ts index 81a3ef1..7c77b0c 100644 --- a/src/lib/canvas-runtime/import-document-stream.ts +++ b/src/lib/canvas-runtime/import-document-stream.ts @@ -1,29 +1,47 @@ +/** + * 导出文档parser5：标准头先建立源树，随后每请求一项才延伸elements范围，实现先转换/加入/显示。 + * 头乱序时明确回退索引解释器，再保留同一ordinal接口；不能因此声称所有文档都零预扫描。 + * 范围扫描串行推进，已定位记录可独立重读；失败尾部成为明确错误记录，成功前缀不被回滚。 + */ import {readImportRange,isImportByteFailure,type ImportByteReader} from './import-byte-reader.ts'; import type {ImportLine,ImportSource} from './imports.ts'; import {createImportBudget,type ImportScheduling} from './import-budget.ts'; import {openDocumentImportSource} from './import-document.ts'; import {fault,validJson} from '../canvas-core/values.ts'; +/** 已定位元素的原字节范围；只随请求积累，不在标准头准备时扫完整elements尾部。 */ type Range={start:number;end:number}; +/** 单源共享byte缓存，仅减少重复读取；各scanner仍有自己的逻辑cursor。 */ type ScanCache={start:number;bytes:Uint8Array}; +/** envelope内的轻量拥有树记录，不包含入口payload或renderer状态。 */ type Canvas={id:string;parent:null|{canvasId:string;entryElementId:string}}; +/** 原源/指纹、可选已验证byte端口、取消与预算；指纹由上层完整性流程管理。 */ type Options=ImportScheduling&{bytes:Blob;fingerprint:string;readBytes?:ImportByteReader;signal:AbortSignal;maxRecordBytes?:number}; +/** 明确支持的format1顶层字段，重复/未知字段拒绝，不隐式猜未来格式。 */ const allowed=new Set(['documentFormatVersion','projectId','canvases','elements','tombstones']); +/** JSON语法空白字节，不用于处理字符串内容。 */ const whitespace=(byte:number)=>byte===9||byte===10||byte===13||byte===32; /** Each scan owns its cursor. An aborted scan cannot advance the resumable source. */ +/** 每次扫描从指定位置建立私有cursor，完成边界后外层才更新可恢复源进度。 */ function scanner(options:Options,signal:AbortSignal,start=0,cache:ScanCache={start:-1,bytes:new Uint8Array()}){ const {bytes}=options,budget=createImportBudget(signal,options); let cursor=start; + /** 在读取/采纳边界检查本次scanner的signal，缓存命中也不能跳过取消。 */ const check=()=>{if(signal.aborted)throw fault('import_aborted');}; + /** 共享缓存不足时精确读16KiB，不推进私有cursor，EOF返回-1。 */ async function peek(){ check();if(cursor>=bytes.size)return -1; if(cursor=cache.start+cache.bytes.length){const loaded=await readImportRange(bytes,options.readBytes,cursor,cursor+16384,signal);cache.start=cursor;cache.bytes=loaded;check();} return cache.bytes[cursor-cache.start]; } + /** 推进一个字节并返回预算可能产生的让出Promise。 */ function step(){cursor++;return budget.step();} + /** 连续消费非字符串区的合法JSON空白。 */ async function spaces(){while(whitespace(await peek()))await step();} + /** 严格匹配一个结构字节，不修补缺逗号/冒号等源损坏。 */ async function take(byte:number){await spaces();if(await peek()!==byte)throw fault('invalid_document_import');await step();} + /** 引号/转义/栈配对扫描一个完整值range，限制嵌套深度，避免把字符串里的分隔符当边界。 */ async function value():Promise{ await spaces();const start=cursor,stack:number[]=[];let quoted=false,escaped=false; while(true){ @@ -39,6 +57,7 @@ function scanner(options:Options,signal:AbortSignal,start=0,cache:ScanCache={sta } if(cursor===start)throw fault('invalid_document_import');return{start,end:cursor}; } + /** 对单range校验大小、fatal UTF8、JSON语法和值域，返回原值，不修改业务ID。 */ async function decode(range:Range,limit=options.maxRecordBytes??256*1024):Promise{ check();if(range.end-range.start>limit)throw fault('record_too_large'); const data=await readImportRange(bytes,options.readBytes,range.start,range.end,signal);let text:string; @@ -48,6 +67,7 @@ function scanner(options:Options,signal:AbortSignal,start=0,cache:ScanCache={sta return{peek,step,spaces,take,value,decode,position:()=>cursor}; } +/** 仅验证envelope墓碑元数据为正安全整数表，目标导入仍使用新ID/新生命周期。 */ function tombstones(value:unknown){ if(!value||typeof value!=='object'||Array.isArray(value)||Object.values(value).some(n=>!Number.isSafeInteger(n)||(n as number)<1))throw fault('invalid_document_import'); } @@ -55,6 +75,7 @@ function tombstones(value:unknown){ /** Parser 5: canonical headers open before scanning the elements tail. * Out-of-order headers require the indexed interpreter to establish ownership first. * Ordinals remain stable; ranges are retained only as records are requested. */ +/** 打开standard header后即可返回源；真正elements/tail结构检查跟随read进行，未扫描尾部不冒称合法。 */ export async function openStreamingDocumentImportSource(input:Options):Promise{ const options:Options=Object.freeze({...input}); const {bytes,signal}=options,max=options.maxRecordBytes??256*1024; @@ -68,6 +89,7 @@ export async function openStreamingDocumentImportSource(input:Options):Promisecanvas.parent===null);if(roots.length!==1)throw fault('invalid_document_import'); const rootId=roots[0].id,rooted=new Set([rootId]); for(const canvas of canvases){ @@ -97,7 +120,9 @@ export async function openStreamingDocumentImportSource(input:Options):Promisecanvas.id!==rootId),ranges:Range[]=[]; + // 源级进度只记录完整range边界/phase，serial只串行化延伸；failure最多输出一条终结错误。 let cursor=header.position(),phase:'elements'|'tail'|'done'='elements',first=true,failure:string|undefined,serial=Promise.resolve(); + /** 延伸一个element range，或验证剩余envelope与EOF；只在结构边界确认后更新cursor/phase。 */ async function scanNext(readSignal:AbortSignal){ const scan=scanAt(readSignal,cursor); if(phase==='elements'){ @@ -109,6 +134,7 @@ export async function openStreamingDocumentImportSource(input:Options):Promise{ if(readSignal.aborted)throw fault('import_aborted'); if(!Number.isSafeInteger(position)||position<0)throw fault('invalid_import_cursor'); const elementIndex=position-children.length; + // 未定位位置才推进源扫描；取消/byte证据错误抛出，其它结构失败转成尾部error记录。 while(elementIndex>=ranges.length&&phase!=='done'){ try{await scanNext(readSignal);}catch(error){if(readSignal.aborted)throw fault('import_aborted');if(isImportByteFailure(error))throw error;failure=(error as {code?:string}).code??'invalid_document_import';phase='done';} } const count=children.length+ranges.length; + // 失败点恰好返回一条可报告记录，下一ordinal为EOF，不能吞掉错误而宣称正常结束。 if(position>=count){ if(failure&&position===count)return{cursor:position,nextCursor:position+1,key:`record:${position}`,error:failure}; if(position===count+(failure?1:0))return null; @@ -140,6 +169,7 @@ export async function openStreamingDocumentImportSource(input:Options):Promise{ const {bytes,signal}=options,max=options.maxRecordBytes??256*1024; if(!(bytes instanceof Blob)||bytes.size>64*1024*1024||!options.fingerprint||!Number.isSafeInteger(max)||max<1)throw fault('invalid_import_source'); let cursor=0,cache={start:-1,bytes:new Uint8Array()};const budget=createImportBudget(signal,options); + /** JSON合法语法空白判定；不在这里做文本编码猜测或容错清洗。 */ const space=(b:number)=>[9,10,13,32].includes(b); + /** 16KiB分块缓存读取，异步边界检查signal，EOF用-1表示。 */ async function peek(){if(signal.aborted)throw fault('import_aborted');if(cursor>=bytes.size)return -1;if(cursor=cache.start+cache.bytes.length){cache={start:cursor,bytes:await readImportRange(bytes,options.readBytes,cursor,cursor+16384,signal)};if(signal.aborted)throw fault('import_aborted');}return cache.bytes[cursor-cache.start];} + /** 推进扫描cursor并按预算计工作量；单原生JSON.parse仍由大小上限约束。 */ function step(){cursor++;return budget.step();} + /** 跳过值/分隔符之间空白，保持range不含前导空白。 */ async function spaces(){while(space(await peek()))await step();} + /** 在跳空白后严格消费预期结构字节，错位立即拒绝，不猜字段/格式。 */ async function take(byte:number){await spaces();if(await peek()!==byte)throw fault('invalid_document_import');await step();} + /** 只定位一个值的范围，跟踪引号/转义/容器配对，扫描嵌套上限256；真正JSON解析稍后做。 */ async function value():Promise{await spaces();const start=cursor,stack:number[]=[];let quoted=false,escaped=false; while(true){const b=await peek();if(b<0){if(quoted||stack.length)throw fault('invalid_document_import');break;}if(!quoted&&!stack.length&&(b===44||b===93||b===125||space(b)))break; if(quoted){if(escaped)escaped=false;else if(b===92)escaped=true;else if(b===34)quoted=false;}else if(b===34)quoted=true;else if(b===91||b===123){stack.push(b);if(stack.length>256)throw fault('json_nesting_limit');}else if(b===93||b===125){if(stack.pop()!==(b===93?91:123))throw fault('invalid_document_import');} const ended=!quoted&&stack.length===0&&(b===34||b===93||b===125);await step();if(ended)break; }if(cursor===start)throw fault('invalid_document_import');return{start,end:cursor}; } + /** 对限定range执行fatal UTF8和JSON值域校验；key/元数据/记录可使用不同字节上限。 */ async function decode(range:Range,limit=max,readSignal=signal){if(range.end-range.start>limit)throw fault('record_too_large');const data=await readImportRange(bytes,options.readBytes,range.start,range.end,readSignal);let text:string;try{text=new TextDecoder('utf-8',{fatal:true}).decode(data);}catch{throw fault('invalid_utf8');}try{const result=JSON.parse(text);if(!validJson(result))throw Error();return result;}catch{throw fault('invalid_json');}} + // envelope字段名去重；elements仅保留ranges，其它字段按有限metadata大小解码。 const fields=new Map(),elements:Range[]=[]; await take(123);await spaces();if(await peek()===125)throw fault('invalid_document_import'); while(true){const key=await decode(await value(),1024);if(typeof key!=='string'||fields.has(key))throw fault('invalid_document_import');await take(58); @@ -28,17 +43,21 @@ export async function openDocumentImportSource(options:{bytes:Blob;fingerprint:s else fields.set(key,await decode(await value(),4*1024*1024)); await spaces();if(await peek()===125){await step();break;}await take(44); } + // 只支持显式format1与已知顶层字段，未知envelope不能偷偷当一组普通记录导入。 await spaces();if(cursor!==bytes.size||fields.get('documentFormatVersion')!==1||typeof fields.get('projectId')!=='string'||!fields.get('projectId')||!Array.isArray(fields.get('canvases'))||!fields.has('elements')||[...fields.keys()].some(key=>!['documentFormatVersion','projectId','canvases','elements','tombstones'].includes(key)))throw fault('invalid_document_import'); const canvases=fields.get('canvases') as {id:string;parent:null|{canvasId:string;entryElementId:string}}[],ids=new Set(); for(const canvas of canvases){if(!canvas||typeof canvas.id!=='string'||!canvas.id||ids.has(canvas.id)||canvas.parent!==null&&(!canvas.parent||typeof canvas.parent.canvasId!=='string'||typeof canvas.parent.entryElementId!=='string'))throw fault('invalid_document_import');ids.add(canvas.id);} const roots=canvases.filter(canvas=>canvas.parent===null);if(roots.length!==1)throw fault('invalid_document_import');const rootId=roots[0].id; + // parent链必须存在且无环；入口实际节点/引用闭合仍由后续原生转换与文档校验完成。 const byId=new Map(canvases.map(canvas=>[canvas.id,canvas])); for(const canvas of canvases){const seen=new Set();let current=canvas;while(current.parent){if(seen.has(current.id))throw fault('invalid_document_import');seen.add(current.id);await budget.step();const parent=byId.get(current.parent.canvasId);if(!parent)throw fault('invalid_document_import');current=parent;}} const tombstones=fields.get('tombstones');if(tombstones!==undefined&&(!tombstones||typeof tombstones!=='object'||Array.isArray(tombstones)||Object.values(tombstones).some(n=>!Number.isSafeInteger(n)||(n as number)<1)))throw fault('invalid_document_import'); + // 排除源根canvas，按固定ordinal拼接子canvas与元素；恢复cursor不是原Blob字节偏移。 const children=canvases.filter(canvas=>canvas.id!==rootId),count=children.length+elements.length; return Object.freeze({fingerprint:options.fingerprint,parserVersion:4,async read(position:number,readSignal:AbortSignal):Promise{ if(readSignal.aborted)throw fault('import_aborted');if(!Number.isSafeInteger(position)||position<0||position>count)throw fault('invalid_import_cursor');if(position===count)return null; const base={cursor:position,nextCursor:position+1,key:`record:${position}`};let record:any,kind:string; + // 单记录payload失败返回error/next ordinal，保留后续合法记录；源级字节失败和取消仍抛出。 try{if(position, limits: {units: number; bytes: number; changes: number}): readonly (readonly ImportLocator[])[] { + // 正向边仅含当前pending内依赖，反向表用于第二次DFS；原index保留稳定源顺序。 const byKey = new Map(pending.map((row, index) => [row.key, {row, index}])), edges = new Map(), reverse = new Map(); for (const row of pending) { const dependencies = row.dependencies.filter(key => byKey.has(key)); edges.set(row.key, dependencies); for (const key of dependencies) { const list = reverse.get(key) ?? []; list.push(row.key); reverse.set(key, list); } } + // 第一遍显式DFS保存完成序，frame.index支持逐邻居推进，避免深链递归调用栈溢出。 const seen = new Set(), finish: string[] = []; for (const row of pending) { if (seen.has(row.key)) continue; @@ -18,6 +25,7 @@ export function selectImportGroups(pending: readonly ImportLocator[], committed: const key = neighbors[frame.index++]; if (!seen.has(key)) { seen.add(key); stack.push({key, index: 0}); } } } + // 第二遍按反向完成序走反向边收集SCC，组内/组间再按最早源index排序。 seen.clear(); const components: ImportLocator[][] = []; for (let i = finish.length - 1; i >= 0; i--) { const first = finish[i]; if (seen.has(first)) continue; @@ -26,13 +34,16 @@ export function selectImportGroups(pending: readonly ImportLocator[], committed: group.sort((a, b) => byKey.get(a.key)!.index - byKey.get(b.key)!.index); components.push(group); } components.sort((a, b) => byKey.get(a[0].key)!.index - byKey.get(b[0].key)!.index); + // committed与本轮已选择组构成可用依赖；一组不能满足自己之外尚未到齐的引用。 const available = new Set(committed), chosen: ImportLocator[][] = []; let units = 0, bytes = 0, changes = 0; for (const group of components) { const internal = new Set(group.map(row => row.key)); if (!group.every(row => row.dependencies.every(key => available.has(key) || internal.has(key)))) continue; + // SCC不可拆：单组字节/changes超安全上限必须报错，不能分批提交半个循环引用。 const size = group.reduce((sum, row) => sum + row.byteLength, 0), count = group.reduce((sum, row) => sum + row.changeCount, 0); if (size > limits.bytes || count > limits.changes) throw fault('import_dependency_group_capacity'); // The frame's adaptive unit budget is soft for one SCC, byte/change safety limits are hard. + // 第一组可超过units软预算以闭合环；后续组和字节/changes总预算仍保守限制。 if (units && units + group.length > limits.units || bytes + size > limits.bytes || changes + count > limits.changes) continue; chosen.push(group); units += group.length; bytes += size; changes += count; for (const row of group) available.add(row.key); } diff --git a/src/lib/canvas-runtime/import-identities.ts b/src/lib/canvas-runtime/import-identities.ts index 897c63b..74ad770 100644 --- a/src/lib/canvas-runtime/import-identities.ts +++ b/src/lib/canvas-runtime/import-identities.ts @@ -1,10 +1,17 @@ +/** + * 一个导入runner的稳定source key→新ID分配器，复用持久job.stableIds而非重开重新随机分配。 + * 使用copy-on-write避免每条无新依赖的记录都复制完整映射；本模块只检查本映射内重复，项目全局身份由上层校验。 + */ import {dataRecord} from '../canvas-core/data-containers.ts'; import type {Id} from '../canvas-core/types.ts'; import {fault} from '../canvas-core/values.ts'; +/** 源记录/依赖键到目标ID的只读字典，element/canvas前缀由调用者约定。 */ type Ids=Readonly>; /** One runner's last immutable ID index. Copies are deferred until allocation. */ +/** 返回带clear方法的捕获工厂，每次为一个源记录的准备提供reserve/has/finish私有上下文。 */ export function createImportIdentities(){ let cached:{ids:Ids;used:Set}|undefined; + /** 相同已冻结纯字符串属性表可复用used索引；其它输入先复制为无原型字典。 */ const capture=(input:Ids,createId:()=>Id)=>{ let identity=cached?.ids===input?cached:undefined; if(!identity){ @@ -13,6 +20,7 @@ export function createImportIdentities(){ } let ids=identity.ids,used=identity.used,owned=ids!==input; return Object.freeze({ + /** 已有键返回原ID；首次遇到依赖才分配并检查本任务冲突，第一次写入时才复制映射/Set。 */ reserve(key:string){ if(typeof key!=='string'||!key)throw fault('invalid_import_key'); if(Object.hasOwn(ids,key))return ids[key]; @@ -20,9 +28,12 @@ export function createImportIdentities(){ if(!owned){ids=Object.assign(Object.create(null),ids);used=new Set(used);owned=true;} (ids as Record)[key]=id;used.add(id);return id; }, + /** 判断目标ID是否已被本映射使用，不声称该ID已加入文档或已同步。 */ has:(id:Id)=>used.has(id), + /** 返回本次分配后的只读映射；没有新分配时保留输入身份便于下一记录复用索引。 */ finish:():Ids=>owned?dataRecord(ids):ids, }); }; + /** clear只丢弃临时复用索引，不清除已持久化job的stableIds。 */ return Object.freeze(Object.assign(capture,{clear(){cached=undefined;}})); } diff --git a/src/lib/canvas-runtime/import-json-array.ts b/src/lib/canvas-runtime/import-json-array.ts index 26828ed..62677dd 100644 --- a/src/lib/canvas-runtime/import-json-array.ts +++ b/src/lib/canvas-runtime/import-json-array.ts @@ -1,25 +1,38 @@ +/** + * 原生记录JSON数组的逐项byte解析器（parserVersion=3），不先JSON.parse整文件。 + * 游标是[之后/逗号之后的字节位置，必须由runner保存合法边界；这里只校验数值范围，不重新证明任意seek位置。 + * 记录级JSON/UTF8/大小错误可报告后继续；数组结构无法恢复时结束源，已发布业务前缀由runner保留。 + */ import {readImportRange,type ImportByteReader} from './import-byte-reader.ts'; import {createImportBudget,type ImportScheduling} from './import-budget.ts'; import type {ImportSource,ImportLine} from './imports.ts'; import {fault,validJson} from '../canvas-core/values.ts'; /** Seekable JSON array reader. Cursors are byte offsets after '[' or a record comma. * Structure errors end the source; previously published records remain explicit. */ +/** 捕获Blob/指纹与读取预算，缓存一块源字节；本解析器不分配业务ID、不发布history或发送请求。 */ export function createJsonArrayImportSource(options:{bytes:Blob;fingerprint:string;readBytes?:ImportByteReader;maxRecordBytes?:number;chunkBytes?:number}&ImportScheduling):ImportSource{ const bytes=options.bytes,max=options.maxRecordBytes??256*1024,chunk=options.chunkBytes??16384; if(!(bytes instanceof Blob)||!options.fingerprint||!Number.isSafeInteger(max)||max<1||!Number.isSafeInteger(chunk)||chunk<1||chunk>max)throw fault('invalid_import_source'); + // byte缓存可跨record复用；空白按JSON四种ASCII空白处理，不误把UTF8字符串内容当分隔符。 let cache={offset:-1,value:new Uint8Array()};const whitespace=(b:number)=>[9,10,13,32].includes(b); + /** read每次返回一个ImportLine及nextCursor；EOF返回null，源级I/O失败仍抛出而不伪装坏记录。 */ return Object.freeze({fingerprint:options.fingerprint,parserVersion:3,async read(start:number,signal:AbortSignal):Promise{ if(!Number.isSafeInteger(start)||start<0||start>bytes.size)throw fault('invalid_import_cursor'); if(start>0&&start===bytes.size){if(signal.aborted)throw fault('import_aborted');return null;} if(bytes.size===0)throw fault('invalid_json_array'); let cursor=start;const budget=createImportBudget(signal,options); + /** 按需填充定长byte缓存，不推进cursor；每次读取前后检查取消。 */ async function peek(){if(signal.aborted)throw fault('import_aborted');if(cursor>=bytes.size)return -1;if(cursor=cache.offset+cache.value.length){cache={offset:cursor,value:await readImportRange(bytes,options.readBytes,cursor,cursor+chunk,signal)};if(signal.aborted)throw fault('import_aborted');}return cache.value[cursor-cache.offset];} + /** 单字节推进并记入合作预算，调用者await可在预算要求时实际让出。 */ function advance(){cursor++;return budget.step();} + /** 只跳过JSON语法空白，字符串中的空白由状态机原样收集。 */ async function spaces(){while(whitespace(await peek()))await advance();} + /** 结构失败转成一条终结错误记录，nextCursor直接EOF，不能继续猜后面的记录边界。 */ const failure=(error:string):ImportLine=>({cursor:start,nextCursor:bytes.size,key:`byte:${start}`,error}); await spaces();if(start===0){if(await peek()!==91)return failure('invalid_json_array');await advance();await spaces();} if(await peek()===93){await advance();await spaces();if(cursor!==bytes.size)return failure('invalid_json_array');if(start!==0){/* Empty tail is valid only after a record, not a trailing comma. */return failure('invalid_json_array');}return null;} if(await peek()===-1)return failure('incomplete_json_array'); + // quoted/escaped隔离字符串分隔符，stack配对{}与[]；超大小后仍扫描至合法边界但不再保留字节。 const recordStart=cursor,parts:number[]=[],stack:number[]=[];let quoted=false,escaped=false,oversized=false; while(true){const byte=await peek();if(byte===-1)return failure('incomplete_json_array'); if(!quoted&&stack.length===0&&(byte===44||byte===93))break; @@ -28,11 +41,13 @@ export function createJsonArrayImportSource(options:{bytes:Blob;fingerprint:stri else if(byte===34)quoted=true;else if(byte===123||byte===91){stack.push(byte);if(stack.length>256)return failure('json_nesting_limit');}else if(byte===125||byte===93){if(stack.pop()!==(byte===125?123:91))return failure('invalid_json_array');} await advance(); } + // 消费外层逗号或]并检查尾部/尾逗号；内嵌数组的]已由stack处理，不会截断记录。 const delimiter=await peek();await advance(); if(delimiter===93){await spaces();if(cursor!==bytes.size)return failure('invalid_json_array');} else {await spaces();if(await peek()===93||await peek()===-1)return failure('invalid_json_array');} const base={cursor:start,nextCursor:cursor,key:`byte:${start}`}; if(oversized)return{...base,error:'record_too_large'}; + // 只解码这一记录：fatal UTF8防替换字符掩盖损坏，再JSON.parse/validJson校验值域。 let text:string;try{text=new TextDecoder('utf-8',{fatal:true}).decode(new Uint8Array(parts));}catch{return{...base,error:'invalid_utf8'};} try{const value=JSON.parse(text);return validJson(value)?{...base,value}:{...base,error:'invalid_json_value'};}catch{return{...base,error:'invalid_json'};} }}); diff --git a/src/lib/canvas-runtime/import-native.ts b/src/lib/canvas-runtime/import-native.ts index a221e40..6235459 100644 --- a/src/lib/canvas-runtime/import-native.ts +++ b/src/lib/canvas-runtime/import-native.ts @@ -1,15 +1,22 @@ +/** + * 原生导入记录到纯ImportUnit的转换：源ID经reserve映射为新身份，依赖保留为source key等待闭合。 + * $target或文档源根映射到显式选定的现有canvas；副本incarnation从1开始，不继承旧墓碑/任务执行。 + * 本模块只规划changes，不写文档、不上传/授权；资源地址和授权事实由宿主/engine另行提供与校验。 + */ import type { Document, ElementRecord, Id, Json, Registry } from '../canvas-core/types.ts'; import { copyReferenceIds } from '../canvas-core/copy-paths.ts'; import { remapCopyData } from '../canvas-core/duplicate.ts'; import { clone, fault } from '../canvas-core/values.ts'; import type { ImportJob, ImportLine, ImportUnit } from './imports.ts'; +/** 校验目标canvas仍存在且入口ID/代次或根身份未变，防旧任务把尾批导入后来替换的分区。 */ export function assertNativeImportTarget(job:ImportJob,document:Document){ const source=job.source,canvas=source?document.canvases.get(source.targetCanvasId):undefined; if(!source||!canvas)throw fault('import_target_missing'); if(source.targetEntry?(canvas.parent?.entryElementId!==source.targetEntry.id||document.elements.get(source.targetEntry.id)?.incarnation!==source.targetEntry.incarnation):canvas.parent!==null)throw fault('import_target_changed'); return source; } +/** 已committed依赖必须仍是原绑定生命周期；用户删后重建同ID不能被后续导入误认领。 */ export function assertNativeImportDependencies(job:ImportJob,document:Document,dependencies:readonly string[]){ for(const key of dependencies)if(job.committedSourceKeys.includes(key)){ const id=job.stableIds[key],binding=job.bindings?.[id]; @@ -20,27 +27,35 @@ export function assertNativeImportDependencies(job:ImportJob,document:Document,d } /** Native NDJSON records use source IDs; $target names the explicitly chosen existing canvas. */ +/** 校验记录封装/源ID，预留自身与引用ID，生成一条创建变化及去除自引用后的依赖列表。 */ export function prepareNativeImport(line:ImportLine,job:ImportJob,reserve:(key:string)=>Id,document:Document,registry:Registry,resourceURL?:(id:Id)=>string):ImportUnit { const source=assertNativeImportTarget(job,document); const raw=line.value as Record|undefined; if(!raw||Array.isArray(raw)||!['element','canvas'].includes(raw.kind as string)||typeof raw.key!=='string'||!raw.key||raw.key==='$target'||!raw.record||typeof raw.record!=='object'||Array.isArray(raw.record))throw fault('invalid_native_import_record'); const record=clone(raw.record) as Record;if(record.id!==raw.key)throw fault('import_source_identity_mismatch'); const dependencies=new Set(),elementIds=new Map(),canvasIds=new Map(); + /** 元素引用先登记依赖再reserve，可先遇到引用后遇到定义，恢复仍复用同一个新ID。 */ const element=(id:Json):Id=>{if(typeof id!=='string'||!id)throw fault('invalid_import_reference');const key=`element:${id}`;dependencies.add(key);const mapped=reserve(key);elementIds.set(id,mapped);return mapped;}; + /** 根/$target直接重定向目标，子canvas则分配稳定新ID并登记canvas依赖。 */ const owner=(id:Json):Id=>{if(id==='$target'||source.kind==='native-document'&&id===raw.sourceRootId){canvasIds.set(id as string,source.targetCanvasId);return source.targetCanvasId;}if(typeof id!=='string'||!id)throw fault('invalid_import_reference');const key=`canvas:${id}`;dependencies.add(key);const mapped=reserve(key);canvasIds.set(id,mapped);return mapped;}; const key=`${raw.kind}:${raw.key}`,id=reserve(key); + // 不把源根作为第二个根插入；这里只处理有parent的子canvas，parent与入口都使用新映射。 if(raw.kind==='canvas'){ const parent=record.parent as Record|null; if(!parent||typeof parent!=='object'||Array.isArray(parent))throw fault('invalid_import_canvas'); const after={...record,id,parent:{...parent,canvasId:owner(parent.canvasId),entryElementId:element(parent.entryElementId)}}; dependencies.delete(key);return{key,dependencies:[...dependencies],changes:[{kind:'canvas',id,before:null,after:after as never}]}; } + // record是脱离源的副本，original是其类型视图；未知组件/缺复制声明不猜字段含义。 const original=record as unknown as ElementRecord,definition=registry.components.get(original.componentId); if(registry.resolve(original)!=='known'||!definition?.copy||!original.data||typeof original.data!=='object'||Array.isArray(original.data))throw fault('import_component_unavailable'); + // 文档图片需宿主提供规范资源URL；仅重建引用不代表完成授权，授权由运行时后续门禁证明。 if(source.kind==='native-document')for(const key of definition.resourceFields??[]){const value=original.data[key];if(value===undefined||value===null)continue;if(!resourceURL)throw fault('import_resource_authorization_required');if(typeof value!=='string'||!value)throw fault('invalid_import_reference');if(key==='assetId')record.data={...original.data,src:resourceURL(value)};} record.data=original.data; + // 只遍历schema声明引用，避免重写普通文字里的ID；omit字段由共享remapCopyData处理。 for(const ref of definition.copy.references)for(const item of copyReferenceIds(original.data,ref)){if(ref.target==='element')element(item);else owner(item);} const after={...record,id,incarnation:1,ownerCanvasId:owner(original.ownerCanvasId),data:remapCopyData(original,registry,elementIds,canvasIds,document)}; + // 连线端点同样进入依赖/映射；不能以源owner或字符串形似就猜端点关系。 if(original.structure==='connection'){ if(!original.source||!original.target)throw fault('invalid_import_endpoint'); Object.assign(after,{source:{...original.source,elementId:element(original.source.elementId)},target:{...original.target,elementId:element(original.target.elementId)}}); diff --git a/src/lib/canvas-runtime/import-read-proof.ts b/src/lib/canvas-runtime/import-read-proof.ts index dcf0d1c..e6c571a 100644 --- a/src/lib/canvas-runtime/import-read-proof.ts +++ b/src/lib/canvas-runtime/import-read-proof.ts @@ -1,12 +1,21 @@ +/** + * 解析器实际读取过的源块证明，不是完整源摘要，更不是发送修改的许可。 + * 以16KiB块哈希绑定fileId/长度，检查点只存摘要，不存缓冲；重开需verify旧块，最终还需整源SHA封存。 + * LRU限制驻留块，并分别限制并发range读/底层块读，防seek与恢复耗尽内存。 + */ import {fault,freeze,validJson} from '../canvas-core/values.ts'; import {createSha256} from './sha256.ts'; import {createImportBudget,type ImportScheduling} from './import-budget.ts'; import type {ImportByteReader} from './import-byte-reader.ts'; +/** 固定证明块大小，协议字段也必须等于16384，不能让恢复解释器随运行参数改边界。 */ export const IMPORT_PROOF_BLOCK_BYTES=16384; +/** 本证明源最大64MiB，与原始图片/单记录/一次range读的上限分别约束。 */ const MAX_SOURCE_BYTES=64*1024*1024; +/** version=1是块证明DTO版本，与source.integrityVersion=2的业务封存协议不是同一版本域。 */ export type ImportReadProof={readonly version:1;readonly fileId:string;readonly byteLength:number;readonly blockBytes:16384;readonly blocks:readonly {readonly index:number;readonly sha256:string}[]}; /** This binds only read blocks. It is not a whole-source digest or permission to send mutations. */ +/** 校验fileId/长度/固定块大小、严格递增的合法块索引及摘要；可表示只读过部分非连续块。 */ export function validateImportReadProof(value:unknown,fileId:string,byteLength:number):ImportReadProof{ if(typeof fileId!=='string'||!fileId||!Number.isSafeInteger(byteLength)||byteLength<0||byteLength>MAX_SOURCE_BYTES)throw fault('invalid_import_read_proof'); const proof=value as ImportReadProof; @@ -20,22 +29,28 @@ export function validateImportReadProof(value:unknown,fileId:string,byteLength:n /** Bounded, seekable verified reads for parsers; checkpoints contain hashes, never byte buffers. * Restored evidence must be checked with verify() before a runner resumes business changes. */ +/** 创建可seek的验证reader；初始证明与源绑定，port.read成功后才把新块追加到本地证明。 */ export function createImportReadProof(input:{bytes:Blob;fileId:string;proof?:ImportReadProof;maxResidentBlocks?:number;maxReadBytes?:number}&ImportScheduling){ const options=Object.freeze({...input}); const {bytes,fileId}=options,residentLimit=options.maxResidentBlocks??4,readLimit=options.maxReadBytes??4*1024*1024; if(!(bytes instanceof Blob)||!fileId||typeof fileId!=='string'||bytes.size>MAX_SOURCE_BYTES||!Number.isSafeInteger(residentLimit)||residentLimit<1||residentLimit>64||!Number.isSafeInteger(readLimit)||readLimit<1||readLimit>4*1024*1024)throw fault('invalid_import_read_source'); const initial:ImportReadProof=options.proof?validateImportReadProof(options.proof,fileId,bytes.size):freeze({version:1 as const,fileId,byteLength:bytes.size,blockBytes:IMPORT_PROOF_BLOCK_BYTES,blocks:[]}); const hashes=new Map(initial.blocks.map(block=>[block.index,block.sha256])); + /** 驻留缓存包含字节；持久proof只取index/sha256，不能把短生命周期buffer塞进job。 */ type Block={index:number;sha256:string;data:Uint8Array}; const resident=new Map(),pending=new Map>(),lifecycle=new AbortController(); let closed=false,activeReads=0,proof:ImportReadProof=initial; + /** 每次等待前后检查生命周期/取消；close是源关闭，调用者abort是停止此订阅采纳。 */ const check=(signal:AbortSignal)=>{if(closed)throw fault('import_source_closed');if(signal.aborted)throw fault('import_aborted');}; + /** 单调用可取消等待共享块Promise，不强制取消可能正被其它range复用的只读底层读取。 */ function wait(task:Promise,signal:AbortSignal):Promise{ check(signal);return new Promise((resolve,reject)=>{const abort=()=>{try{check(signal);}catch(error){reject(error);}};signal.addEventListener('abort',abort,{once:true});task.then(value=>{signal.removeEventListener('abort',abort);try{check(signal);resolve(value);}catch(error){reject(error);}},error=>{signal.removeEventListener('abort',abort);reject(error);});}); } + /** 先命中LRU或共享pending，再读块并对旧摘要；新原图与已证明块不符即源改变。 */ function block(index:number):Promise{ const cached=resident.get(index);if(cached){resident.delete(index);resident.set(index,cached);return Promise.resolve(cached);} const active=pending.get(index);if(active)return active; + // 最多4个不同块真实读取，不能靠大量seek创建无限pending；命中已有块仍可共享。 if(pending.size>=4)throw fault('import_read_capacity'); const task=(async()=>{ const start=index*IMPORT_PROOF_BLOCK_BYTES,end=Math.min(start+IMPORT_PROOF_BLOCK_BYTES,bytes.size),data=new Uint8Array(await bytes.slice(start,end).arrayBuffer()); @@ -46,6 +61,7 @@ export function createImportReadProof(input:{bytes:Blob;fileId:string;proof?:Imp })(); pending.set(index,task);void task.finally(()=>{if(pending.get(index)===task)pending.delete(index);}).catch(()=>{});return task; } + /** 精确range端口：只分配所需输出，逐块拼接并按软预算让出，最后成功才提交used块证明。 */ const port:ImportByteReader=Object.freeze({byteLength:bytes.size,async read(start:number,end:number,signal:AbortSignal){ const combined=AbortSignal.any([signal,lifecycle.signal]);check(combined); if(!Number.isSafeInteger(start)||!Number.isSafeInteger(end)||start<0||endbytes.size||end-start>readLimit)throw fault('invalid_import_byte_range'); @@ -57,6 +73,7 @@ export function createImportReadProof(input:{bytes:Blob;fileId:string;proof?:Imp const value=await wait(block(index),combined),offset=index*IMPORT_PROOF_BLOCK_BYTES,from=Math.max(start,offset),to=Math.min(end,offset+value.data.byteLength); output.set(value.data.subarray(from-offset,to-offset),from-start);used.push({index:value.index,sha256:value.sha256});await budget.step(128); } + // 同一读失败不把未完成的业务进度当证明成功；采纳前再次检查缓存/并发累积的摘要一致性。 check(combined);for(const value of used)if(hashes.has(value.index)&&hashes.get(value.index)!==value.sha256)throw fault('import_source_changed'); let changed=false;for(const value of used)if(!hashes.has(value.index)){hashes.set(value.index,value.sha256);changed=true;} if(changed)proof=freeze({version:1 as const,fileId,byteLength:bytes.size,blockBytes:IMPORT_PROOF_BLOCK_BYTES,blocks:[...hashes].sort(([a],[b])=>a-b).map(([index,sha256])=>({index,sha256}))}); @@ -64,8 +81,11 @@ export function createImportReadProof(input:{bytes:Blob;fileId:string;proof?:Imp }finally{activeReads--;} }}); return Object.freeze({port,getProof:()=>proof, + /** 重开后验证所有已保存块；不扫描尚未读到的尾部，故不能代替完整SHA或已durable封存。 */ async verify(signal:AbortSignal){const combined=AbortSignal.any([signal,lifecycle.signal]),budget=createImportBudget(combined,options);check(combined);for(const index of [...hashes.keys()]){await wait(block(index),combined);await budget.step(128);}check(combined);}, + /** 幂等关闭，清驻留字节并取消等待者；已开始的只读Blob读取可能迟到，由closed检查拒绝采纳。 */ close(){if(closed)return;closed=true;resident.clear();lifecycle.abort();}, + /** 只读诊断：驻留/活range/pending块/证明块分别计数，不当作导入完成百分比。 */ stats:()=>Object.freeze({residentBlocks:resident.size,activeReads,pendingBlocks:pending.size,proofBlocks:hashes.size,closed}), }); } diff --git a/src/lib/canvas-runtime/import-verification.ts b/src/lib/canvas-runtime/import-verification.ts index 09acd61..88b65f0 100644 --- a/src/lib/canvas-runtime/import-verification.ts +++ b/src/lib/canvas-runtime/import-verification.ts @@ -1,25 +1,35 @@ +/** + * 导入源完整性证据的拥有权与发送屏障。块读取证明允许先显示/编辑合法前缀， + * 但source SHA必须在可靠Root中封存后才能发送受影响修改；普通内存布尔值不能冒充durable证据。 + */ import type {Id} from '../canvas-core/types.ts'; import {equal,fault} from '../canvas-core/values.ts'; import {importNeedsSeal,validateImportJob,type ImportJob} from './imports.ts'; import {validateImportReadProof} from './import-read-proof.ts'; /** A derived durable-cut index, never an in-memory permission boolean. */ +/** 从给定job表派生fileId→SHA索引；调用者须传已可靠保存的cut，本函数不执行落盘。 */ export function importSeals(jobs:ReadonlyMap|undefined){ const seals=new Map();for(const job of jobs?.values()??[])if(job.source?.integrityVersion===2&&job.source.sha256&&!job.source.integrityError)seals.set(job.source.fileId,job.source.sha256);return seals; } +/** integrity-v2任务需自身已完整验证且durable seals匹配；取消且零已提交前缀的任务可不阻塞。 */ export function importsMaySend(jobs:ReadonlyMap|undefined,seals:ReadonlyMap){ for(const job of jobs?.values()??[])if(job.source?.integrityVersion===2&&!(job.status==='cancelled'&&!job.committedSourceKeys.length)&&(importNeedsSeal(job)||seals.get(job.source.fileId)!==job.source.sha256))return false;return true; } /** Only engine-owned publication paths may advance read evidence or source verification. */ +/** 发布前保护引擎拥有的readProof/SHA/执行进度，未持owner任务ID的路径不能伪造或删除它们。 */ export function assertImportEvidence(previous:ReadonlyMap|undefined,next:ReadonlyMap|undefined,owner:Id|undefined){ if(previous===next)return; for(const[id,job]of previous??[])if(job.source?.integrityVersion===2&&!next?.has(id))throw fault('import_integrity_owned',[id]); + /** 对比执行事实时剔除generation和源retirement生命周期字段；其它进度/身份仍要求合法owner。 */ const execution=(job:ImportJob|undefined)=>{if(!job)return job;const{generation:_,source,...body}=job;if(!source)return body;const{retirement:__,...identity}=source;return {...body,source:identity};}; for(const[id,job]of next??[]){const source=job.source,old=previous?.get(id)?.source;if(source?.integrityVersion!==2){if(old?.integrityVersion===2)throw fault('import_integrity_owned',[id]);continue;} validateImportJob(job);const proof=validateImportReadProof(source.readProof,source.fileId,source.byteLength); + // engine的解析进度要求证明为从0开始的连续块前缀，比通用seek reader的证明形状更严格。 if(proof.blocks.some((block,index)=>block.index!==index))throw fault('invalid_import_read_proof'); const changed=!equal(old?.readProof,proof)||old?.sha256!==source.sha256||old?.integrityError!==source.integrityError; if(owner!==id&&(changed||!equal(execution(previous?.get(id)),execution(job))))throw fault('import_integrity_owned',[id]); + // 相同fileId/长度和已存在SHA不可改变，旧块摘要也必须原样保留，不能用“重算成功”覆盖冲突源。 if(old?.integrityVersion===2){if(old.fileId!==source.fileId||old.byteLength!==source.byteLength||old.sha256&&old.sha256!==source.sha256)throw fault('import_source_changed',[id]);const hashes=new Map(proof.blocks.map(block=>[block.index,block.sha256]));for(const block of old.readProof!.blocks)if(hashes.get(block.index)!==block.sha256)throw fault('import_source_changed',[id]);} } }