194 lines
18 KiB
TypeScript
194 lines
18 KiB
TypeScript
import { applyJournal, diffJournal, type JournalPatch } from './journal-delta.ts';
|
|
import { isSealedRecoveryImage } from './recovery-encoding.ts';
|
|
import type { Json } from '../canvas-core/types.ts';
|
|
import { clone, equal, fault, freeze, validJson } from '../canvas-core/values.ts';
|
|
|
|
import type {StorageScope,RecoveryImage,DurableBatch,StorageLeader,StoredFileInfo,FileInventoryPage,StoragePort} from './storage-port.ts';
|
|
export type {StorageScope,RecoveryImage,DurableBatch,StorageLeader,StoredFileInfo,FileInventoryPage,StoragePort} from './storage-port.ts';
|
|
type Metadata = { fence: number; throughRevision?: number; checkpointRevision?: number };
|
|
type StoredImage = { readonly image: RecoveryImage; readonly checksum: string };
|
|
type StoredDelta = {journalVersion:1;baseRevision:number;baseChecksum:string;header:Omit<RecoveryImage,'payload'>;patches:JournalPatch[];checksum:string};
|
|
type StoredCut = StoredImage | StoredDelta;
|
|
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(cut.journalVersion!==1)throw fault('recovery_delta_interpreter_unavailable');
|
|
if(!cut.header)throw fault('corrupt_recovery_delta');validateImage({...cut.header,payload:null});return cut.header;
|
|
}
|
|
async function imageDigest(image: RecoveryImage) {
|
|
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};
|
|
}
|
|
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]);
|
|
};
|
|
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;
|
|
}
|
|
function validateImage(image: RecoveryImage) {
|
|
if (![1,2,3,4,5,6,7,8].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');
|
|
}
|
|
const requestValue = <T>(request: IDBRequest<T>) => new Promise<T>((resolve, reject) => { request.onsuccess = () => resolve(request.result); request.onerror = () => reject(request.error); });
|
|
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. */
|
|
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;
|
|
const handles = new Set<StorageLeader>(), lastCuts=new WeakMap<StorageLeader,StoredImage>(), writeTails=new WeakMap<StorageLeader,Promise<unknown>>();
|
|
const opening = new Promise<IDBDatabase>((resolve, reject) => {
|
|
const request = factory.open(options.name ?? 'endoleap-headless', 2);
|
|
request.onupgradeneeded = () => {
|
|
for (const name of ['metadata', 'checkpoints', 'journal', 'files']) if (!request.result.objectStoreNames.contains(name)) request.result.createObjectStore(name);
|
|
};
|
|
request.onerror = () => reject(request.error);
|
|
request.onblocked = () => reject(fault('storage_upgrade_blocked'));
|
|
request.onsuccess = () => {
|
|
if (closed) { request.result.close(); reject(fault('storage_closed')); 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(() => {});
|
|
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);
|
|
}
|
|
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);
|
|
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; }
|
|
}
|
|
async function recoverCut(leader:StorageLeader):Promise<StoredImage|null>{
|
|
const cuts=await execute(leader,['checkpoints','journal'],'readonly',async(transaction,metadata,key)=>{
|
|
if(metadata.throughRevision===undefined)return [];
|
|
if(metadata.checkpointRevision===undefined)throw fault('recovery_checkpoint_missing');
|
|
const checkpoint=await requestValue<StoredImage|undefined>(transaction.objectStore('checkpoints').get([key,metadata.checkpointRevision]));
|
|
if(!checkpoint?.image)throw fault('recovery_checkpoint_missing');
|
|
let header=cutHeader(checkpoint);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])));
|
|
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 [checkpoint,...journal];
|
|
});
|
|
let previous:StoredImage|null=null;
|
|
// Hash and apply after the readonly transaction; awaiting crypto inside IDB would close it.
|
|
for(const cut of cuts){
|
|
let image:RecoveryImage;
|
|
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)).checksum)throw fault('recovery_checksum_mismatch');previous=freeze({image,checksum:cut.checksum});
|
|
}
|
|
check(leader);if(previous&&(!lastCuts.has(leader)||lastCuts.get(leader)!.image.rootRevision<=previous.image.rootRevision))lastCuts.set(leader,previous);return previous;
|
|
}
|
|
return Object.freeze({
|
|
fileDeletion:'atomic-v1' as const,
|
|
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; });
|
|
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;
|
|
},
|
|
async recover(leader: StorageLeader) {const cut=await recoverCut(leader);return cut?clone(cut.image):null;},
|
|
async commit(leader: StorageLeader, batch: DurableBatch) {
|
|
// Detach mutable callers and descriptors before the first await, including queued commits.
|
|
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);
|
|
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>();
|
|
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);
|
|
const task=(writeTails.get(leader)??Promise.resolve()).catch(()=>{}).then(async()=>{
|
|
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');}
|
|
const digest=await imageDigest(image),stored:StoredImage={image,checksum:digest.checksum};let candidate:StoredCut=stored;
|
|
if(base&&image.rootRevision===base.image.rootRevision+1){
|
|
const patches=diffJournal(base.image.payload,image.payload);
|
|
if(patches){const delta:StoredDelta={journalVersion:1,baseRevision:base.image.rootRevision,baseChecksum:base.checksum,header:{formatVersion:image.formatVersion,rootRevision:image.rootRevision,localRevision:image.localRevision},patches,checksum:digest.checksum};if(new TextEncoder().encode(JSON.stringify(delta)).byteLength<digest.byteLength)candidate=delta;}
|
|
}
|
|
const receipt=await execute(leader,['checkpoints','journal','files'],'readwrite',async(transaction,metadata,key)=>{
|
|
const previous=metadata.throughRevision;
|
|
const saved=previous===undefined?undefined:await requestValue<StoredCut>(transaction.objectStore(metadata.checkpointRevision===previous?'checkpoints':'journal').get([key,previous]));
|
|
if(previous!==undefined&&image.rootRevision!==previous+1){
|
|
if(image.rootRevision!==previous)throw fault('storage_revision_gap');
|
|
if(!saved||saved.checksum!==stored.checksum||!base||!equal(base.image,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');return{durableRootRevision:previous};
|
|
}
|
|
if(previous!==undefined&&(!saved||image.localRevision<cutHeader(saved).localRevision))throw fault('storage_revision_conflict');
|
|
if(!('image' in candidate)&&(candidate.baseRevision!==previous||candidate.baseChecksum!==saved?.checksum))throw fault('recovery_delta_base_mismatch');
|
|
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]);
|
|
// A format transition must not depend on a newer partial checkpoint in an older reader.
|
|
if(previous===undefined||((base?.image.formatVersion??0)>=7&&base?.image.formatVersion!==image.formatVersion)||image.rootRevision-(metadata.checkpointRevision??0)>=checkpointEvery){
|
|
transaction.objectStore('checkpoints').put(stored,[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(candidate,[key,image.rootRevision]);
|
|
transaction.objectStore('metadata').put({...metadata,throughRevision:image.rootRevision},key);return{durableRootRevision:image.rootRevision};
|
|
});
|
|
lastCuts.set(leader,freeze(stored));return receipt;
|
|
});
|
|
writeTails.set(leader,task);return task;
|
|
},
|
|
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'],'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=()=>{const cursor=request.result;if(!cursor){resolve(files);return;}const parts=cursor.key,bytes=cursor.value;
|
|
if(!Array.isArray(parts)||parts.length!==2||parts[0]!==key||typeof parts[1]!=='string'||!parts[1]||!(bytes instanceof Blob)){reject(fault('corrupt_file_store'));return;}
|
|
files.push(Object.freeze({id:parts[1],byteLength:bytes.size,mimeType:bytes.type}));if(files.length>limit)resolve(files);else cursor.continue();
|
|
};
|
|
});
|
|
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}:{})});
|
|
});
|
|
},
|
|
readFile(leader: StorageLeader, id: string) { return execute(leader, ['files'], 'readonly', (transaction, _metadata, key) => requestValue<Blob | undefined>(transaction.objectStore('files').get([key, id]))); },
|
|
close() { closed = true; for (const handle of handles) void handle.release(); void opening.then(database => database.close()).catch(() => {}); },
|
|
});
|
|
}
|