92 lines
9.0 KiB
TypeScript
92 lines
9.0 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. */
|
|
/** 校验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;
|
|
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；初始证明与源绑定，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<number,Block>(),pending=new Map<number,Promise<Block>>(),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<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);});});
|
|
}
|
|
/** 先命中LRU或共享pending，再读块并对旧摘要；新原图与已证明块不符即源改变。 */
|
|
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端口：只分配所需输出，逐块拼接并按软预算让出，最后成功才提交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||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,
|
|
/** 重开后验证所有已保存块；不扫描尚未读到的尾部，故不能代替完整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}),
|
|
});
|
|
}
|