/** * 解析器实际读取过的源块证明,不是完整源摘要,更不是发送修改的许可。 * 以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. */ /** 校验源身份/长度、固定块宽、严格递增的合法块索引、摘要格式及允许字段,再捕获冻结 DTO。 * 可包含不连续的部分块,不要求完整前缀;这里不读 Blob 或验证摘要真实性,也不是整源 SHA。 * 缺记录/非法 JSON/未知版本报解释器不可用,其它身份/结构错误报 invalid_import_read_proof。 */ 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; if(!proof||!validJson(proof)||proof.version!==1)throw fault('import_read_proof_interpreter_unavailable'); if(proof.fileId!==fileId||proof.byteLength!==byteLength||proof.blockBytes!==IMPORT_PROOF_BLOCK_BYTES||!Array.isArray(proof.blocks)||Object.keys(proof).some(key=>!['version','fileId','byteLength','blockBytes','blocks'].includes(key)))throw fault('invalid_import_read_proof'); let last=-1;const count=Math.ceil(byteLength/IMPORT_PROOF_BLOCK_BYTES); if(proof.blocks.length>count)throw fault('invalid_import_read_proof'); for(const block of proof.blocks){if(!block||!Number.isSafeInteger(block.index)||block.index<=last||block.index>=count||typeof block.sha256!=='string'||!/^[a-f0-9]{64}$/.test(block.sha256)||Object.keys(block).some(key=>!['index','sha256'].includes(key)))throw fault('invalid_import_read_proof');last=block.index;} return freeze({version:1 as const,fileId,byteLength,blockBytes:IMPORT_PROOF_BLOCK_BYTES,blocks:proof.blocks.map(block=>({...block}))}); } /** 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:hashes 保存期望摘要,resident 是按 Map 顺序淘汰的 LRU,pending 合并同块在途读取。 * 初始 proof 仅结构校验即进入 hashes/getProof;没有 verified 标志,调用方须在恢复业务前主动 verify。 * 最多 4 个活跃非空 range、4 个不同块的原生读取;驻留默认 4 块(最多 64),单 range 默认/最多 4MiB。 * range 输出、在途块和 Blob 本身另占内存,residentLimit 不是整体内存硬上限;本实例不持久化证明。 */ 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; /** 入口/成功采纳时检查 closed 与 signal,关闭优先报 import_source_closed,其次报 import_aborted。 * 单个订阅取消只停止等待,不终止共享原生块读取;已拒绝的底层 Promise 仍可能保留自己的原始错误。 */ 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);});}); } /** 命中 resident 时移动到 Map 尾部作为最近使用,命中 pending 时复用同一 Promise。 * 缺块就读完整对齐 16KiB(最后块截 EOF)并计算 SHA,可能覆盖请求 range 之外的相邻字节。 * 比对 hashes 中旧期望后放 resident,但不在此新增持久候选 proof;range 失败也可留下可复用缓存。 * 缓存命中不重新读 Blob/算摘要,依赖 Blob 不可变合同;淘汰只丢字节,不丢已提交的期望摘要。 */ 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()); if(closed)throw fault('import_source_closed');if(data.byteLength!==end-start)throw fault('import_source_changed'); const hash=createSha256();hash.update(data);const sha256=hash.hex(),expected=hashes.get(index); if(expected!==undefined&&expected!==sha256)throw fault('import_source_changed'); const value={index,sha256,data};resident.set(index,value);while(resident.size>residentLimit)resident.delete(resident.keys().next().value!);return value; })(); pending.set(index,task);void task.finally(()=>{if(pending.get(index)===task)pending.delete(index);}).catch(()=>{});return task; } /** 精确 range 端口:要求 end ≤ EOF,不自行截断;空范围在容量检查前返回,不读块/新增证明。 * 非空范围分配独立输出并逐块拼接,每块向预算记 128 单位,最后检查取消与并发摘要一致性。 * 只有整次范围成功才将 used 中新块加入 hashes/proof;取消/失败不提交本次部分块,也不回退其它成功读取的证明。 * activeReads 随等待退出而归还,真实不可取消 I/O 仍占 pending 容量,不能用大量取消绕过底层上限。 */ 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'); if(start===end)return new Uint8Array(); if(activeReads>=4||pending.size>=4&&!resident.has(Math.floor(start/IMPORT_PROOF_BLOCK_BYTES))&&!pending.has(Math.floor(start/IMPORT_PROOF_BLOCK_BYTES)))throw fault('import_read_capacity'); activeReads++;try{ const output=new Uint8Array(end-start),used:Pick[]=[],budget=createImportBudget(combined,options); for(let index=Math.floor(start/IMPORT_PROOF_BLOCK_BYTES);index*IMPORT_PROOF_BLOCK_BYTESa-b).map(([index,sha256])=>({index,sha256}))}); return output; }finally{activeReads--;} }}); return Object.freeze({port,getProof:()=>proof, /** 对调用开始时 hashes 中的索引逐块核对,不增加新证明块、不产生完整源摘要或自动持久化。 * 重开实例缓存为空时实际重读旧块;同实例调用可命中 resident,不强制绕过缓存重新读物理字节。 * 不核对尚未证明的尾部,也不锁住并发 range;是否已安全恢复、何时封存由引擎决定。 */ 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);}, /** 幂等关闭并清 resident/取消等待者,不清 hashes/proof,getProof 仍可读取最后不可变快照。 * 不等待原生 Blob 读取退出;pending 由各 Promise finally 清理,迟到读取经 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}), }); }