Zayden-Jung/Endoleap
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Store recovery journal deltas between verified checkpoints

Browse Source
This commit is contained in:
Codex Bot
2026-11-08 18:27:59 +08:00
parent e00ac017f1
commit 6ff34b67e7
4 changed files with 148 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/canvas-runtime/journal-delta.ts
+43
View File
@@ -0,0 +1,43 @@
import type { Json } from '../canvas-core/types.ts';
import { fault, validJson } from '../canvas-core/values.ts';
type Path = readonly (string | number)[];
export type JournalPatch = {op:'set';path:Path;value:Json}|{op:'splice';path:Path;start:number;deleteCount:number;items:Json[]};
/** Exact JSON property order is retained because recovery checksums cover serialized bytes. */
export function diffJournal(before:Json,after:Json):JournalPatch[]|null {
const patches:JournalPatch[]=[];
function visit(a:Json,b:Json,path:(string|number)[]){
if(a===b||patches.length>10000)return;
if(Array.isArray(a)&&Array.isArray(b)){
const common=Math.min(a.length,b.length);for(let i=0;i<common;i++)visit(a[i],b[i],[...path,i]);
if(a.length!==b.length)patches.push({op:'splice',path,start:common,deleteCount:a.length-common,items:b.slice(common)});
}else if(a&&b&&typeof a==='object'&&typeof b==='object'&&!Array.isArray(a)&&!Array.isArray(b)){
const left=Object.keys(a),right=Object.keys(b);
if(left.length===right.length&&left.every((key,i)=>key===right[i]))for(const key of right)visit(a[key],b[key],[...path,key]);
else patches.push({op:'set',path,value:b});
}else patches.push({op:'set',path,value:b});
}
visit(before,after,[]);return patches.length>10000?null:patches;
}
/** Copy only changed ancestors; never mutate the preceding checkpoint or another patch's input. */
export function applyJournal(before:Json,patches:readonly JournalPatch[]):Json {
if(!Array.isArray(patches)||patches.length>10000)throw fault('corrupt_recovery_delta');
let result=before;
for(const patch of patches){
if(!patch||!['set','splice'].includes(patch.op)||!Array.isArray(patch.path)||patch.path.length>64||patch.path.some((key:unknown)=>typeof key!=='string'&&(typeof key!=='number'||!Number.isSafeInteger(key)||key<0)))throw fault('corrupt_recovery_delta');
if(patch.op==='set'?!validJson(patch.value):!Array.isArray(patch.items)||!validJson(patch.items))throw fault('corrupt_recovery_delta');
function replace(value:Json,depth:number):Json {
if(depth===patch.path.length){
if(patch.op==='set')return patch.value;
if(!Array.isArray(value)||!Number.isSafeInteger(patch.start)||!Number.isSafeInteger(patch.deleteCount)||patch.start<0||patch.deleteCount<0||patch.start>value.length||patch.deleteCount>value.length-patch.start)throw fault('corrupt_recovery_delta');
return [...value.slice(0,patch.start),...patch.items,...value.slice(patch.start+patch.deleteCount)];
}
const key=patch.path[depth];
if(!value||typeof value!=='object'||!Object.hasOwn(value,key)||(Array.isArray(value)?typeof key!=='number'||key>=value.length:typeof key!=='string'))throw fault('corrupt_recovery_delta');
const copy=Array.isArray(value)?[...value]:{...value};
Object.defineProperty(copy,key,{value:replace((value as Record<string|number,Json>)[key],depth+1),enumerable:true,writable:true,configurable:true});return copy;
}
result=replace(result,0);
}
return result;
}
src/lib/canvas-runtime/storage.ts
+62 -49
View File
@@ -1,6 +1,7 @@
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, validJson } from '../canvas-core/values.ts';
import { clone, equal, fault, freeze, validJson } from '../canvas-core/values.ts';
export type StorageScope = { readonly accountId: string; readonly workspaceId: string; readonly projectId: string };
export type RecoveryImage = { readonly formatVersion: 1 | 2; readonly rootRevision: number; readonly localRevision: number; readonly payload: Json };
@@ -15,9 +16,17 @@ export type StoragePort = {
};
type Metadata = { fence: number; throughRevision?: number; checkpointRevision?: number };
type StoredImage = { readonly image: RecoveryImage; readonly checksum: string };
async function checksum(image: RecoveryImage) {
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 [...new Uint8Array(await crypto.subtle.digest('SHA-256', bytes))].map(value => value.toString(16).padStart(2, '0')).join('');
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');
@@ -37,9 +46,9 @@ export function createBrowserStorage(options: { indexedDB?: IDBFactory; locks?:
const checkpointEvery = options.checkpointEvery ?? 32;
if (!Number.isSafeInteger(checkpointEvery) || checkpointEvery < 1) throw fault('invalid_checkpoint_interval');
let closed = false;
const handles = new Set<StorageLeader>();
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', 1);
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);
};
@@ -71,6 +80,27 @@ export function createBrowserStorage(options: { indexedDB?: IDBFactory; locks?:
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({
async claim(scope: StorageScope) {
const key = keyFor(scope);
@@ -97,53 +127,36 @@ export function createBrowserStorage(options: { indexedDB?: IDBFactory; locks?:
void held.catch(failed);
return result;
},
async recover(leader: StorageLeader) {
const images = 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 image = checkpoint.image;
validateImage(image);
if (image.rootRevision !== metadata.checkpointRevision) throw fault('corrupt_recovery_image');
const journal = metadata.checkpointRevision === metadata.throughRevision ? [] : await requestValue<StoredImage[]>(transaction.objectStore('journal').getAll(IDBKeyRange.bound([key, metadata.checkpointRevision + 1], [key, metadata.throughRevision])));
for (const next of journal) { validateImage(next.image); if (next.image.rootRevision !== image.rootRevision + 1 || next.image.localRevision < image.localRevision) throw fault('recovery_journal_gap'); image = next.image; }
if (image.rootRevision !== metadata.throughRevision) throw fault('recovery_journal_gap');
return [checkpoint, ...journal];
});
// Hashing must run after the IDB transaction, which would otherwise auto-close while awaiting it.
for (const stored of images) if (stored.checksum !== await checksum(stored.image)) throw fault('recovery_checksum_mismatch');
check(leader);
return images.at(-1)?.image ?? null;
},
async recover(leader: StorageLeader) {const cut=await recoverCut(leader);return cut?clone(cut.image):null;},
async commit(leader: StorageLeader, batch: DurableBatch) {
// Copy before awaiting: callers cannot change the durable cut while a transaction queues.
const image = isSealedRecoveryImage(batch.image) ? batch.image : clone(batch.image), files = (batch.files ?? []).map(file => ({id:file.id,bytes:file.bytes})); if(!isSealedRecoveryImage(image))validateImage(image);
const stored: StoredImage = { image, checksum: await checksum(image) };
return execute(leader, ['checkpoints', 'journal', 'files'], 'readwrite', async (transaction, metadata, key) => {
const previous = metadata.throughRevision;
if (previous !== undefined && image.rootRevision !== previous + 1) {
if (image.rootRevision !== previous) throw fault('storage_revision_gap');
const saved = await requestValue<StoredImage>(transaction.objectStore(metadata.checkpointRevision === previous ? 'checkpoints' : 'journal').get([key, previous]));
if (!equal(saved, stored) || files.length) throw fault('storage_revision_conflict');
return { durableRootRevision: previous };
// 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}));if(!isSealedRecoveryImage(image))validateImage(image);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);
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;}
}
if (previous !== undefined && image.localRevision < (await requestValue<StoredImage>(transaction.objectStore(metadata.checkpointRevision === previous ? 'checkpoints' : 'journal').get([key, previous]))).image.localRevision) throw fault('storage_revision_conflict');
for (const file of files) {
if (!file.id || !(file.bytes instanceof Blob)) throw fault('invalid_staged_file');
// Immutable resource identity: replacing bytes requires a new resource ID.
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]);
}
if (previous === undefined || 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(stored, [key, image.rootRevision]);
transaction.objectStore('metadata').put({ ...metadata, throughRevision: image.rootRevision }, key);
return { durableRootRevision: image.rootRevision };
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');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]);}
if(previous===undefined||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;
},
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(() => {}); },