107 lines
11 KiB
TypeScript
107 lines
11 KiB
TypeScript
/**
|
||
* 解析器实际读取过的源块证明，不是完整源摘要，更不是发送修改的许可。
|
||
* 以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<number,Block>(),pending=new Map<number,Promise<Block>>(),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<T>(task:Promise<T>,signal:AbortSignal):Promise<T>{
|
||
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<Block>{
|
||
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||end<start||end>bytes.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<Block,'index'|'sha256'>[]=[],budget=createImportBudget(combined,options);
|
||
for(let index=Math.floor(start/IMPORT_PROOF_BLOCK_BYTES);index*IMPORT_PROOF_BLOCK_BYTES<end;index++){
|
||
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}))});
|
||
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}),
|
||
});
|
||
}
|