307 lines
32 KiB
TypeScript
307 lines
32 KiB
TypeScript
/**
|
||
* 浏览器持久化 adapter：Web Locks 保证同 origin/数据库/scope 单写者，metadata.fence 拒绝旧代次句柄。
|
||
* DB schema v5 的七个 store 保存 Root checkpoint/journal、文件与共享 depot，历史/ledger 等在恢复图像内。
|
||
* crypto/JSON 编码在事务外完成；Root、文件增删和 depot 清单在同一事务提交，oncomplete 才算 durable。
|
||
* DB schema、checkpointVersion、journalVersion、RecoveryImage.formatVersion 是独立兼容域，不能混用。
|
||
* adapter 实例持有连接与 leader 集合，per-leader WeakMap 保存写尾链/可靠基线；close 不代替 engine 的持久化排空。
|
||
*/
|
||
import {serializeRecoveryImage} from './recovery-json.ts';
|
||
import {depotStores,createDepotStorage,detachDepotCut,verifyDepotBytes,writeDepotCut,promoteDepotFiles,resolveDepotFile} from './asset-depot-storage.ts';
|
||
import { applyJournal, diffJournal, type JournalPatch } from './journal-delta.ts';
|
||
import { isSealedRecoveryImage, sealedRecoveryByteLength } from './recovery-encoding.ts';
|
||
import type { Json } from '../canvas-core/types.ts';
|
||
import { clone, fault, freeze, validJson } from '../canvas-core/values.ts';
|
||
|
||
import type {StorageScope,RecoveryImage,DurableBatch,StorageLeader,StoredFileInfo,FileInventoryPage,StoragePort} from './storage-port.ts';
|
||
// 仅重导出便携端口类型；浏览器专属 IDB/Web Locks 实现留在本模块，不污染纯 runtime 入口。
|
||
export type {StorageScope,RecoveryImage,DurableBatch,StorageLeader,StoredFileInfo,FileInventoryPage,StoragePort} from './storage-port.ts';
|
||
/** scope 头：fence 是 writer 代次，throughRevision 是可靠 Root 末端，checkpointRevision 是回放起点。 */
|
||
type Metadata = { fence: number; throughRevision?: number; checkpointRevision?: number };
|
||
/** 兼容旧完整图像行：image 与对其 JSON 表示计算的摘要。 */
|
||
type StoredImage = { readonly image: RecoveryImage; readonly checksum: string };
|
||
/** journal 差分：必须匹配前一 Root 的 revision/checksum；v1 哈希结果图像，v2 哈希 deltaBody。
|
||
* 因此 StoredImage.checksum 在内存中也可能是当前日志链节点摘要，并非总是 image 的独立摘要。 */
|
||
type StoredDelta = {journalVersion:1|2;baseRevision:number;baseChecksum:string;header:Omit<RecoveryImage,'payload'>;patches:JournalPatch[];checksum:string};
|
||
/** 新完整 checkpoint 以 UTF-8 Blob 保存，header 便于事务内检查连续性，解析/校验在事务外。 */
|
||
type StoredBlobImage={readonly checkpointVersion:1;readonly header:Omit<RecoveryImage,'payload'>;readonly bytes:Blob;readonly checksum:string};
|
||
/** 恢复读取同时支持旧完整行、差分行与 Blob checkpoint，未知解释器版本明确拒绝。 */
|
||
type StoredCut = StoredImage | StoredDelta | StoredBlobImage;
|
||
/** 区分完整行与差分分支；真正格式/版本校验仍由 cutHeader 完成。 */
|
||
const isDelta=(cut:StoredCut):cut is StoredDelta=>!('image' in cut)&&!('checkpointVersion' in cut);
|
||
/** 在事务内抽取并校验版本头，避免为检查日志顺序就在事务中 await Blob/crypto。 */
|
||
function cutHeader(cut:StoredCut):Omit<RecoveryImage,'payload'>{
|
||
if(!cut||typeof cut!=='object')throw fault('corrupt_recovery_image');
|
||
if('image' in cut){validateImage(cut.image);return cut.image;}
|
||
if('checkpointVersion' in cut){if(cut.checkpointVersion!==1)throw fault('recovery_checkpoint_interpreter_unavailable');if(!(cut.bytes instanceof Blob)||!cut.header)throw fault('corrupt_recovery_image');validateImage({...cut.header,payload:null});return cut.header;}
|
||
if(![1,2].includes(cut.journalVersion))throw fault('recovery_delta_interpreter_unavailable');
|
||
if(!cut.header)throw fault('corrupt_recovery_delta');validateImage({...cut.header,payload:null});return cut.header;
|
||
}
|
||
/** 对旧 image 或 v2 unsigned delta 的精确 JSON 字节做 SHA256，同时返回 UTF-8 长度用于成本选择。 */
|
||
async function imageDigest(image: unknown) {
|
||
const bytes = new TextEncoder().encode(JSON.stringify(image));
|
||
return {checksum:[...new Uint8Array(await crypto.subtle.digest('SHA-256', bytes))].map(value => value.toString(16).padStart(2, '0')).join(''),byteLength:bytes.byteLength};
|
||
}
|
||
/** ArrayBuffer 的 SHA256 小写十六进制表示，验证存储字节完整性，不代表服务权限。 */
|
||
async function bytesDigest(bytes:ArrayBuffer){return [...new Uint8Array(await crypto.subtle.digest('SHA-256',bytes))].map(value=>value.toString(16).padStart(2,'0')).join('');}
|
||
/** 读取序列化 Blob 并哈希，调用方必须在 IDB 事务外等待。 */
|
||
async function blobDigest(bytes:Blob){return bytesDigest(await bytes.arrayBuffer());}
|
||
/** 剔除 checksum 自身，仅重建 v2 差分签名覆盖的原字段和顺序。 */
|
||
function deltaBody(cut:StoredDelta){const {journalVersion,baseRevision,baseChecksum,header,patches}=cut;return{journalVersion,baseRevision,baseChecksum,header,patches};}
|
||
/** 校验 account/workspace/project 并编码稳定复合 scope 键，防止相同项目跨账号串用。 */
|
||
const keyFor = (scope: StorageScope) => {
|
||
if (![scope.accountId, scope.workspaceId, scope.projectId].every(value => typeof value === 'string' && value.length > 0)) throw fault('invalid_storage_scope');
|
||
return JSON.stringify([scope.accountId, scope.workspaceId, scope.projectId]);
|
||
};
|
||
/** 从已持久 Root 汇总资源/孤儿/导入源的永久清理 ID，拒绝后来用同 ID 写入新内容。 */
|
||
function purgedFileIds(image:RecoveryImage):Set<string>{
|
||
const result=new Set<string>(),payload=image.payload as {runtime?:{resources?:[string,{retirement?:{purgedAt?:number}}][];orphanFiles?:[string,{retirement?:{purgedAt?:number}}][];imports?:[string,{source?:{fileId:string;retirement?:{purgedAt?:number};previousFiles?:{fileId:string;retirement?:{purgedAt?:number}}[]}}][]}};
|
||
for(const rows of [payload?.runtime?.resources,payload?.runtime?.orphanFiles])if(Array.isArray(rows))for(const row of rows)if(Array.isArray(row)&&typeof row[0]==='string'&&row[1]?.retirement?.purgedAt!==undefined)result.add(row[0]);
|
||
if(Array.isArray(payload?.runtime?.imports))for(const [,job]of payload.runtime.imports){if(!job?.source)continue;for(const file of [job.source,...(job.source.previousFiles??[])])if(file.retirement?.purgedAt!==undefined)result.add(file.fileId);}
|
||
return result;
|
||
}
|
||
/** 存储信封校验：版本受支持、revision 合法且 local≤root、payload 为 JSON；领域恢复另做深校验。 */
|
||
function validateImage(image: RecoveryImage) {
|
||
if (![1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17].includes(image.formatVersion)) throw fault('unsupported_recovery_version');
|
||
if (!Number.isSafeInteger(image.rootRevision) || image.rootRevision < 0 || !Number.isSafeInteger(image.localRevision) || image.localRevision < 0 || image.localRevision > image.rootRevision || !validJson(image.payload)) throw fault('corrupt_recovery_image');
|
||
}
|
||
/** 将单条 IDB 请求转 Promise；put 成功不意味着整个事务成功。 */
|
||
const requestValue = <T>(request: IDBRequest<T>) => new Promise<T>((resolve, reject) => { request.onsuccess = () => resolve(request.result); request.onerror = () => reject(request.error); });
|
||
/** 唯一提交终态屏障：oncomplete 成功、onabort 失败；onerror 可能还会接着 abort，不提前 resolve。 */
|
||
const transactionDone = (transaction: IDBTransaction) => new Promise<void>((resolve, reject) => { transaction.oncomplete = () => resolve(); transaction.onabort = () => reject(transaction.error ?? fault('storage_aborted')); transaction.onerror = () => { /* onabort is the terminal barrier. */ }; });
|
||
|
||
/** Single-origin browser adapter. Web Locks serialize tabs; a DB fence rejects stale handles. */
|
||
/** 需要 IndexedDB 与 Web Locks；缺能力时明确拒绝，不悄悄退成无法可靠恢复的内存存储。
|
||
* 构造即开始打开数据库，尚未持有项目写锁；claim 才按 scope 取锁。工厂返回冻结能力外壳，内部连接状态仍可变。
|
||
* name/options 应由宿主保持稳定，锁名读取 options.name；不能把同一数据库意外装配成不同锁域。 */
|
||
export function createBrowserStorage(options: { indexedDB?: IDBFactory; locks?: LockManager; name?: string; checkpointEvery?: number } = {}): StoragePort {
|
||
const factory = options.indexedDB ?? globalThis.indexedDB, locks = options.locks ?? globalThis.navigator?.locks;
|
||
if (!factory || !locks) throw fault('persistent_storage_unavailable');
|
||
const checkpointEvery = options.checkpointEvery ?? 32;
|
||
if (!Number.isSafeInteger(checkpointEvery) || checkpointEvery < 1) throw fault('invalid_checkpoint_interval');
|
||
let closed = false, openingFailure:Error|undefined;
|
||
// 每 leader 保存上一可靠 cut/检查点和顺序写尾链；未提交成功不能更新这些缓存。
|
||
const handles = new Set<StorageLeader>(), lastCuts=new WeakMap<StorageLeader,StoredImage>(), writeTails=new WeakMap<StorageLeader,Promise<unknown>>(),lastCheckpoints=new WeakMap<StorageLeader,number>();
|
||
// 打开/升级一次数据库；blocked 必须显式上报，不能无限等待并让用户误以为已保存。
|
||
const opening = new Promise<IDBDatabase>((resolve, reject) => {
|
||
const request = factory.open(options.name ?? 'endoleap-headless', 5);
|
||
/** 增量建立缺失 store；不在 schema 升级阶段隐式迁移业务画布或重写冻结请求。 */
|
||
request.onupgradeneeded = () => {
|
||
if(closed||openingFailure){request.transaction?.abort();return;}
|
||
for (const name of ['metadata', 'checkpoints', 'journal', 'files',...depotStores]) if (!request.result.objectStoreNames.contains(name)) request.result.createObjectStore(name);
|
||
};
|
||
request.onerror = () => reject(closed?fault('storage_closed'):openingFailure??request.error);
|
||
request.onblocked = () => {openingFailure=fault('storage_upgrade_blocked');reject(openingFailure);};
|
||
/** 迟到打开先检查关闭/blocked 状态；versionchange 及时关连接并使 leader 失效。 */
|
||
request.onsuccess = () => {
|
||
if (closed||openingFailure) { request.result.close(); reject(closed?fault('storage_closed'):openingFailure); return; }
|
||
request.result.onversionchange = () => { closed = true; request.result.close(); for (const handle of handles) void handle.release(); };
|
||
resolve(request.result);
|
||
};
|
||
});
|
||
// Opening failures are observable through operations without an unhandled background rejection.
|
||
void opening.catch(() => {});
|
||
/** 同步检查 adapter/leader 生命周期并返回 scope key；真正 DB fence 还要在事务内核对。 */
|
||
function check(leader: StorageLeader) {
|
||
if (closed) throw fault('storage_closed');
|
||
if (!handles.has(leader) || !leader.active) throw fault('storage_leader_lost');
|
||
return keyFor(leader.scope);
|
||
}
|
||
/** 统一事务包装：await DB 后再查句柄，事务内比 fence，运行回调后等待 oncomplete。
|
||
* run 只能等待适合事务生命周期的 IDB 工作；任何失败 abort 整批并等待终态再抛原错误。 */
|
||
async function execute<T>(leader: StorageLeader, stores: string[], mode: IDBTransactionMode, run: (transaction: IDBTransaction, metadata: Metadata, key: string) => Promise<T>): Promise<T> {
|
||
const key = check(leader), database = await opening;
|
||
check(leader);
|
||
const transaction = database.transaction([...new Set(['metadata', ...stores])], mode), done = transactionDone(transaction);
|
||
try {
|
||
const metadata = await requestValue<Metadata | undefined>(transaction.objectStore('metadata').get(key));
|
||
if (!metadata || metadata.fence !== leader.fence) throw fault('storage_fence_lost');
|
||
check(leader);
|
||
const result = await run(transaction, metadata, key);
|
||
// readwrite 的原子可见性与可靠回执都在 oncomplete 后成立；callback 返回值只是待采纳结果。
|
||
await done; // Never resolve durability on an individual put callback.
|
||
return result;
|
||
} catch (error) { try { transaction.abort(); } catch { /* Already terminal. */ } await done.catch(() => {}); throw error; }
|
||
}
|
||
/** 先在只读事务捕获完整日志区间，再在事务外重建/哈希，最后才缓存可靠恢复结果。
|
||
* 返回的是该只读快照，不保证 crypto 完成时仍是最新 Root；并发较新恢复已入缓存时不将其倒退。
|
||
* 捕获整个回放区间，逐条 apply/hash 没有独立合作预算；存储信封通过后仍需领域 decodeRecovery 深校验。 */
|
||
async function recoverCut(leader:StorageLeader):Promise<StoredImage|null>{
|
||
const captured=await execute(leader,['checkpoints','journal'],'readonly',async(transaction,metadata,key)=>{
|
||
// 没有可靠头就返回 null；不据此推断 files/depot 中没有尚未归属或共享的字节。
|
||
if(metadata.throughRevision===undefined)return{cuts:[] as StoredCut[],checkpointRevision:undefined};
|
||
if(metadata.checkpointRevision===undefined)throw fault('recovery_checkpoint_missing');
|
||
const checkpoint=await requestValue<StoredCut|undefined>(transaction.objectStore('checkpoints').get([key,metadata.checkpointRevision]));
|
||
if(!checkpoint)throw fault('recovery_checkpoint_missing');
|
||
let header=cutHeader(checkpoint);if(isDelta(checkpoint))throw fault('recovery_checkpoint_missing');if(header.rootRevision!==metadata.checkpointRevision)throw fault('corrupt_recovery_image');
|
||
const journal=metadata.checkpointRevision===metadata.throughRevision?[]:await requestValue<StoredCut[]>(transaction.objectStore('journal').getAll(IDBKeyRange.bound([key,metadata.checkpointRevision+1],[key,metadata.throughRevision])));
|
||
// 日志必须逐个 rootRevision 连续且 localRevision 不倒退，缺一段也不能恢复成较短成功前缀。
|
||
for(const row of journal){const next=cutHeader(row);if(next.rootRevision!==header.rootRevision+1||next.localRevision<header.localRevision)throw fault('recovery_journal_gap');header=next;}
|
||
if(header.rootRevision!==metadata.throughRevision)throw fault('recovery_journal_gap');return{cuts:[checkpoint,...journal],checkpointRevision:metadata.checkpointRevision};
|
||
});
|
||
let previous:StoredImage|null=null;
|
||
// Hash and apply after the readonly transaction; awaiting crypto inside IDB would close it.
|
||
for(const cut of captured.cuts){
|
||
let image:RecoveryImage;
|
||
// Blob 摘要先验证再 fatal UTF-8/JSON 解码；解码信封头必须与外部 header 一致。
|
||
if('checkpointVersion' in cut){
|
||
const serialized=await cut.bytes.arrayBuffer();if(cut.checksum!==await bytesDigest(serialized))throw fault('recovery_checksum_mismatch');
|
||
try{image=JSON.parse(new TextDecoder('utf-8',{fatal:true}).decode(serialized));}catch{throw fault('corrupt_recovery_image');}
|
||
if(!image||typeof image!=='object'||Array.isArray(image))throw fault('corrupt_recovery_image');validateImage(image);
|
||
if(image.formatVersion!==cut.header.formatVersion||image.rootRevision!==cut.header.rootRevision||image.localRevision!==cut.header.localRevision)throw fault('recovery_checksum_mismatch');
|
||
}else{
|
||
// 差分必须串在 exact base revision/checksum 后；v1 哈希结果 image，v2 哈希差分 body。
|
||
if('image' in cut)image=cut.image;
|
||
else{if(!previous||cut.baseRevision!==previous.image.rootRevision||cut.baseChecksum!==previous.checksum)throw fault('recovery_delta_base_mismatch');image={...cut.header,payload:applyJournal(previous.image.payload,cut.patches)};validateImage(image);}
|
||
if(cut.checksum!==(await imageDigest('image' in cut||cut.journalVersion===1?image:deltaBody(cut))).checksum)throw fault('recovery_checksum_mismatch');
|
||
}
|
||
previous=freeze({image,checksum:cut.checksum});
|
||
}
|
||
// crypto 可能较慢，结束后再查 leader；迟到旧恢复不能把 lastCuts 缓存倒退。
|
||
check(leader);if(previous&&(!lastCuts.has(leader)||lastCuts.get(leader)!.image.rootRevision<=previous.image.rootRevision)){lastCuts.set(leader,previous);lastCheckpoints.set(leader,captured.checkpointRevision!);}return previous;
|
||
}
|
||
return Object.freeze({
|
||
/** 显式声明支持与 Root 同事务删除文件，配合退休证明校验使用。 */
|
||
fileDeletion:'atomic-v1' as const,
|
||
/** 共享仓库也走 execute，因此同样受 metadata fence 与事务终态约束。 */
|
||
assetDepot:createDepotStorage((leader,stores,mode,run)=>execute(leader,stores,mode,tx=>run(tx))),
|
||
/** 用 ifAvailable 申请 scope 排他 Web Lock：不排队等其它持有者，但仍异步等数据库与 fence 事务。
|
||
* scope 按入口标量构造锁键、取得后复制进 handle，调用者须在异步 claim 期间保持输入稳定。
|
||
* 同一 adapter 不同 scope 可持不同锁；execute 仍用 metadata fence 拒绝旧句柄。 */
|
||
async claim(scope: StorageScope) {
|
||
const key = keyFor(scope);
|
||
if (closed) throw fault('storage_closed');
|
||
let unlock!: () => void, acquired!: (leader: StorageLeader) => void, failed!: (error: unknown) => void;
|
||
const released = new Promise<void>(resolve => { unlock = resolve; });
|
||
const result = new Promise<StorageLeader>((resolve, reject) => { acquired = resolve; failed = reject; });
|
||
// result 只交付 handle，held 则覆盖整个持锁生命周期；released 是二者之间的释放闸门。
|
||
// 首次 release 先失效/移出集合再解锁并等 held，重复调用会立即返回，不是另一次排空等待。
|
||
const held = locks.request(`endoleap-headless:${options.name ?? 'endoleap-headless'}:${key}`, { mode: 'exclusive', ifAvailable: true }, async lock => {
|
||
if (!lock) throw fault('storage_scope_in_use');
|
||
const database = await opening;
|
||
if (closed) throw fault('storage_closed');
|
||
const transaction = database.transaction('metadata', 'readwrite'), done = transactionDone(transaction);
|
||
try {
|
||
const store = transaction.objectStore('metadata'), previous = await requestValue<Metadata | undefined>(store.get(key));
|
||
const fence = (previous?.fence ?? 0) + 1;
|
||
if (!Number.isSafeInteger(fence)) throw fault('storage_fence_exhausted');
|
||
store.put({ ...previous, fence }, key); await done;
|
||
let active = true;
|
||
const handle: StorageLeader = Object.freeze({ scope: Object.freeze({ ...scope }), fence, get active() { return active; }, async release() { if (!active) return; active = false; handles.delete(handle); unlock(); await held; } });
|
||
handles.add(handle); acquired(handle);
|
||
await released;
|
||
} catch (error) { try { transaction.abort(); } catch { /* Already terminal. */ } await done.catch(() => {}); throw error; }
|
||
});
|
||
void held.catch(failed);
|
||
return result;
|
||
},
|
||
/** 给调用者返回恢复图像副本，避免它改写 lastCuts 持有的可信基线。 */
|
||
async recover(leader: StorageLeader) {const cut=await recoverCut(leader);return cut?clone(cut.image):null;},
|
||
/** 捕获批次→校验删除证明→顺序准备编码→原子写 Root/文件/depot；返回实际 durableRootRevision。 */
|
||
async commit(leader: StorageLeader, batch: DurableBatch) {
|
||
// 入队前捕获图像与描述符，避免排在旧写入后等待时被调用者改写；codec-owned sealed 图像可直接复用。
|
||
// file 外壳复制而 Blob 原样借用；depot 用自己的捕获器，均不代表已经可靠落盘。
|
||
const depot=batch.depot?detachDepotCut(batch.depot):undefined;
|
||
const image=isSealedRecoveryImage(batch.image)?batch.image:freeze(clone(batch.image)),files=(batch.files??[]).map(file=>({id:file.id,bytes:file.bytes})),deleteFiles=[...(batch.deleteFiles??[])];if(deleteFiles.length>128||new Set(deleteFiles).size!==deleteFiles.length||deleteFiles.some(id=>typeof id!=='string'||!id||files.some(file=>file.id===id)))throw fault('invalid_file_deletion');if(!isSealedRecoveryImage(image))validateImage(image);
|
||
// 删除依据当前提交图像的退休声明：最多128个唯一 ID，不能同批又新增。
|
||
// 这里检查证明形状/归属/任务终态，不自行重算 history pin 或保留期；Root 完整性由引擎负责。
|
||
if(deleteFiles.length){
|
||
type FileProof={id?:string;version?:number;byteLength?:number;mimeType?:string;digest?:string;fileId?:string;retirement?:{version:number;unreferencedAt:number;purgedAt?:number}};
|
||
const payload=image.payload as {runtime?:{orphanFiles?:[string,FileProof][];resources?:[string,FileProof][];imports?:[string,{status:string;source?:FileProof&{previousFiles?:FileProof[]}}][]}};
|
||
const proofs=new Map<string,FileProof>(),owned=new Set<string>();
|
||
/** 文件 ID 只能属于一个资源/源/孤儿身份，按任务终态决定其退休证明是否可用于删除。 */
|
||
const own=(id:string,file:FileProof,eligible:boolean)=>{if(typeof id!=='string'||!id||owned.has(id))throw fault('invalid_file_deletion');owned.add(id);if(eligible)proofs.set(id,file);};
|
||
for(const [id,resource]of payload?.runtime?.resources??[])own(id,resource,true);
|
||
if(image.formatVersion>=4)for(const [,job]of payload?.runtime?.imports??[]){if(job.source){own(job.source.fileId!,job.source,['completed','partial','cancelled'].includes(job.status));for(const file of job.source.previousFiles??[])own(file.fileId!,file,true);}}
|
||
if(image.formatVersion>=5)for(const [id,file]of payload?.runtime?.orphanFiles??[]){if(file.version!==1||file.id!==id||!Number.isSafeInteger(file.byteLength)||file.byteLength!<0||typeof file.mimeType!=='string'||typeof file.digest!=='string'||!/^[a-f0-9]{64}$/.test(file.digest))throw fault('invalid_file_deletion');own(id,file,true);}
|
||
if(image.formatVersion<3||deleteFiles.some(id=>{const marker=proofs.get(id)?.retirement;return !marker||marker.version!==1||!Number.isSafeInteger(marker.unreferencedAt)||marker.unreferencedAt<0||!Number.isSafeInteger(marker.purgedAt)||marker.purgedAt!<marker.unreferencedAt;}))throw fault('invalid_file_deletion');
|
||
}
|
||
check(leader);
|
||
// 同 leader 的 crypto/编码/事务也串在一条 Promise 尾链上，避免各自准备不同基线后乱序提交。
|
||
// 吞掉旧尾链失败只为继续调度，新请求错误仍向其调用者传播；不自动重试旧批次或跨 leader 排队。
|
||
const task=(writeTails.get(leader)??Promise.resolve()).catch(()=>{}).then(async()=>{
|
||
if(depot)await verifyDepotBytes(depot);
|
||
let base=lastCuts.get(leader)??await recoverCut(leader);
|
||
if(base?.image.rootRevision===image.rootRevision)base=await recoverCut(leader);
|
||
if(files.length&&base){const retired=purgedFileIds(base.image);if(files.some(file=>retired.has(file.id)))throw fault('resource_identity_conflict');}
|
||
// checkpoint 决定替换可靠回放起点；没有基线、达到间隔，或基线版本 >=7 且格式变化时建立新起点。
|
||
// delta 不划算也会写完整 Blob，但若 checkpoint=false，它仍是 journal 中的一行，不压缩旧链。
|
||
const baseCheckpoint=lastCheckpoints.get(leader),checkpoint=!base||((base.image.formatVersion>=7)&&base.image.formatVersion!==image.formatVersion)||image.rootRevision-(baseCheckpoint??0)>=checkpointEvery;
|
||
let stored:StoredImage|undefined,candidate:StoredCut|undefined;
|
||
// 同 revision 先沿用已恢复的链摘要；事务内仍逐字比较整 image、查 DB 头，不能只凭 revision 接受重试。
|
||
if(base?.image.rootRevision===image.rootRevision){stored={image,checksum:base.checksum};candidate=stored;}
|
||
// 连续且标准信封才尝试 delta；比较真实字节成本，差分过大/不可表达则回退完整 Blob。
|
||
else if(base&&!checkpoint&&image.rootRevision===base.image.rootRevision+1&&JSON.stringify(Object.keys(image))===JSON.stringify(['formatVersion','rootRevision','localRevision','payload'])){
|
||
const patches=diffJournal(base.image.payload,image.payload);
|
||
if(patches){
|
||
const unsigned={journalVersion:2 as const,baseRevision:base.image.rootRevision,baseChecksum:base.checksum,header:{formatVersion:image.formatVersion,rootRevision:image.rootRevision,localRevision:image.localRevision},patches},digest=await imageDigest(unsigned),delta:StoredDelta={...unsigned,checksum:digest.checksum};
|
||
// Adding the ASCII checksum member costs 78 UTF-8 bytes. Owned codec
|
||
// images carry an exact size; mutable callers keep the defensive path.
|
||
const fullBytes=sealedRecoveryByteLength(image)??new TextEncoder().encode(JSON.stringify(image)).byteLength;
|
||
if(digest.byteLength+78<fullBytes){stored={image,checksum:delta.checksum};candidate=delta;}
|
||
}
|
||
}
|
||
// 编码/哈希在事务外，期间通过 check 防退休句柄继续计算或采纳结果。
|
||
if(!stored||!candidate){const bytes=await serializeRecoveryImage(image,{check:()=>check(leader)}),checksum=await blobDigest(bytes);stored={image,checksum};candidate={checkpointVersion:1,header:{formatVersion:image.formatVersion,rootRevision:image.rootRevision,localRevision:image.localRevision},bytes,checksum};}
|
||
const accepted=stored,cut=candidate;
|
||
// Root、文件、共享身份/pins 和 metadata 在同一 readwrite 事务中，不暴露已写字节但没写 Root 的状态。
|
||
const receipt=await execute(leader,['checkpoints','journal','files',...depotStores],'readwrite',async(transaction,metadata,key)=>{
|
||
const previous=metadata.throughRevision;
|
||
if(previous===undefined&&base)throw fault('recovery_delta_base_mismatch');
|
||
const saved=previous===undefined?undefined:await requestValue<StoredCut>(transaction.objectStore(metadata.checkpointRevision===previous?'checkpoints':'journal').get([key,previous]));
|
||
// 同 Root 重试须图像/摘要相同、无新增文件，且请求删除的行已经不存在；不再写 Root。
|
||
// 可在这条分支补写 depot 清单/字节与文件指针，支持幂等重建索引；事务失败仍整体回滚。
|
||
if(previous!==undefined&&image.rootRevision!==previous+1){
|
||
if(image.rootRevision!==previous)throw fault('storage_revision_gap');
|
||
if(!saved||saved.checksum!==accepted.checksum||!base||JSON.stringify(base.image)!==JSON.stringify(image)||files.length)throw fault('storage_revision_conflict');for(const id of deleteFiles)if(await requestValue(transaction.objectStore('files').getKey([key,id])))throw fault('storage_revision_conflict');if(depot){await writeDepotCut(transaction,leader.scope,image.rootRevision,depot);await promoteDepotFiles(transaction,leader.scope,key,image,depot);}return{durableRootRevision:previous,checkpointRevision:metadata.checkpointRevision!};
|
||
}
|
||
// 事务内重新比对真实基线，防止准备编码时取的 lastCuts 已不对应当前 DB 头。
|
||
if(previous!==undefined&&(!saved||image.localRevision<cutHeader(saved).localRevision))throw fault('storage_revision_conflict');
|
||
if(previous!==undefined&&(metadata.checkpointRevision!==baseCheckpoint||base?.image.rootRevision!==previous||base.checksum!==saved?.checksum))throw fault('recovery_delta_base_mismatch');
|
||
if(isDelta(cut)&&(cut.baseRevision!==previous||cut.baseChecksum!==saved?.checksum))throw fault('recovery_delta_base_mismatch');
|
||
if(depot)await writeDepotCut(transaction,leader.scope,image.rootRevision,depot);
|
||
// 新文件不可覆盖既有 ID；后面的 delete/promote 任一失败都回滚本事务全部更改。
|
||
for(const file of files){if(!file.id||!(file.bytes instanceof Blob))throw fault('invalid_staged_file');if(await requestValue(transaction.objectStore('files').getKey([key,file.id])))throw fault('resource_identity_conflict',[file.id]);transaction.objectStore('files').put(file.bytes,[key,file.id]);}
|
||
for(const id of deleteFiles)transaction.objectStore('files').delete([key,id]);
|
||
if(depot)await promoteDepotFiles(transaction,leader.scope,key,image,depot);
|
||
// A format transition must not depend on a newer partial checkpoint in an older reader.
|
||
// 检查点持久化和旧 journal 压缩一起提交，崩溃后不会只剩半份回放链。
|
||
if(checkpoint){
|
||
transaction.objectStore('checkpoints').put(cut,[key,image.rootRevision]);if(metadata.checkpointRevision!==undefined)transaction.objectStore('checkpoints').delete([key,metadata.checkpointRevision]);transaction.objectStore('journal').delete(IDBKeyRange.bound([key,0],[key,image.rootRevision]));metadata.checkpointRevision=image.rootRevision;
|
||
}else transaction.objectStore('journal').put(cut,[key,image.rootRevision]);
|
||
transaction.objectStore('metadata').put({...metadata,throughRevision:image.rootRevision},key);return{durableRootRevision:image.rootRevision,checkpointRevision:metadata.checkpointRevision!};
|
||
});
|
||
// 只有 execute 已等到事务成功后才推进内存可靠水位，quota/abort 不虚报已保存。
|
||
lastCuts.set(leader,freeze(accepted));lastCheckpoints.set(leader,receipt.checkpointRevision);return{durableRootRevision:receipt.durableRootRevision};
|
||
});
|
||
writeTails.set(leader,task);return task;
|
||
},
|
||
/** 按 expectedRootRevision 的稳定 ID 游标分页盘点，最多128项；Root 改变则拒绝拼接旧页。
|
||
* limit 约束返回的可读取文件数，不约束游标经过的已清理指针数；不计算 digest，也不做远端读取。
|
||
* 同 Root 的 depot 回收可能改变可读字节，revision 门禁不等于跨多次调用持有同一 IDB 快照。 */
|
||
scanFiles(leader:StorageLeader,input:{expectedRootRevision:number;after?:string;limit:number}) {
|
||
const {expectedRootRevision,after,limit}=input;
|
||
if(!Number.isSafeInteger(expectedRootRevision)||expectedRootRevision<0||!Number.isSafeInteger(limit)||limit<1||limit>128||after!==undefined&&(typeof after!=='string'||!after))return Promise.reject(fault('invalid_file_inventory'));
|
||
return execute(leader,['files',...depotStores],'readonly',async(transaction,metadata,key)=>{
|
||
if(metadata.throughRevision!==expectedRootRevision)throw fault('stale_file_inventory');
|
||
// 游标多取一个实际有字节的文件判断下一页；共享指针先解算，已显式清理的字节不当活文件计数。
|
||
const rows=await new Promise<StoredFileInfo[]>((resolve,reject)=>{
|
||
const files:StoredFileInfo[]=[],range=IDBKeyRange.bound(after===undefined?[key]:[key,after],[key,[]],after!==undefined,true),request=transaction.objectStore('files').openCursor(range);
|
||
request.onerror=()=>reject(request.error);
|
||
request.onsuccess=()=>{void(async()=>{const cursor=request.result;if(!cursor){resolve(files);return;}const parts=cursor.key;
|
||
if(!Array.isArray(parts)||parts.length!==2||parts[0]!==key||typeof parts[1]!=='string'||!parts[1]){throw fault('corrupt_file_store');}
|
||
const bytes=await resolveDepotFile(transaction,leader.scope,parts[1],cursor.value);
|
||
if(bytes)files.push(Object.freeze({id:parts[1],byteLength:bytes.size,mimeType:bytes.type}));if(files.length>limit)resolve(files);else cursor.continue();
|
||
})().catch(reject);};
|
||
});
|
||
const more=rows.length>limit,files=more?rows.slice(0,limit):rows;
|
||
return Object.freeze({files:Object.freeze(files),rootRevision:expectedRootRevision,...(more?{nextAfter:files.at(-1)!.id}:{})});
|
||
});
|
||
},
|
||
/** 在有效 fence 下读取本地 Blob 或共享字节指针；缺字节返回 undefined，不伪造空文件。 */
|
||
readFile(leader: StorageLeader, id: string) { return execute(leader, ['files',...depotStores], 'readonly', async(transaction, _metadata, key) => resolveDepotFile(transaction,leader.scope,id,await requestValue<unknown>(transaction.objectStore('files').get([key, id])))); },
|
||
/** 同步使 adapter 关闭并触发 leader.release/连接关闭，不返回排空 Promise，也不主动 abort 所有在途事务。
|
||
* 已开始事务由 IDB 自行结束，未开始任务会在 check 被拒绝；宿主须先等 engine 保存完成再释放 adapter。 */
|
||
close() { closed = true; for (const handle of handles) void handle.release(); void opening.then(database => database.close()).catch(() => {}); },
|
||
});
|
||
}
|