From 6ff34b67e7d37197382161bc1c4ed265d4ca461a Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Sun, 8 Nov 2026 18:27:59 +0800 Subject: [PATCH] Store recovery journal deltas between verified checkpoints --- src/lib/canvas-runtime/journal-delta.ts | 43 +++++++++ src/lib/canvas-runtime/storage.ts | 111 +++++++++++++---------- tests/headless/browser-storage-cases.mjs | 22 +++++ tests/headless/journal-delta.test.mjs | 21 +++++ 4 files changed, 148 insertions(+), 49 deletions(-) create mode 100644 src/lib/canvas-runtime/journal-delta.ts create mode 100644 tests/headless/journal-delta.test.mjs diff --git a/src/lib/canvas-runtime/journal-delta.ts b/src/lib/canvas-runtime/journal-delta.ts new file mode 100644 index 0000000..30082a9 --- /dev/null +++ b/src/lib/canvas-runtime/journal-delta.ts @@ -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;ikey===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)[key],depth+1),enumerable:true,writable:true,configurable:true});return copy; + } + result=replace(result,0); + } + return result; +} diff --git a/src/lib/canvas-runtime/storage.ts b/src/lib/canvas-runtime/storage.ts index 87636de..2f57586 100644 --- a/src/lib/canvas-runtime/storage.ts +++ b/src/lib/canvas-runtime/storage.ts @@ -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;patches:JournalPatch[];checksum:string}; +type StoredCut = StoredImage | StoredDelta; +function cutHeader(cut:StoredCut):Omit{ + 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(); + const handles = new Set(), lastCuts=new WeakMap(), writeTails=new WeakMap>(); const opening = new Promise((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{ + 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(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(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 { - if (metadata.throughRevision === undefined) return []; - if (metadata.checkpointRevision === undefined) throw fault('recovery_checkpoint_missing'); - const checkpoint = await requestValue(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(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(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(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(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=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(transaction.objectStore('files').get([key, id]))); }, close() { closed = true; for (const handle of handles) void handle.release(); void opening.then(database => database.close()).catch(() => {}); }, diff --git a/tests/headless/browser-storage-cases.mjs b/tests/headless/browser-storage-cases.mjs index 4883d6f..caa2765 100644 --- a/tests/headless/browser-storage-cases.mjs +++ b/tests/headless/browser-storage-cases.mjs @@ -319,6 +319,26 @@ async function resourceRecoveryCase(name, assert) { } +async function deltaJournalCase(name,assert){ + const databaseName=`${name}-delta-journal`,scope={accountId:'delta-account',workspaceId:'workspace',projectId:'project'},key=JSON.stringify(Object.values(scope)),rows=Array.from({length:100},(_,i)=>({id:String(i),text:'x'.repeat(1000)})),image=revision=>({formatVersion:1,rootRevision:revision,localRevision:revision,payload:{rows:rows.map((row,i)=>i===0?{...row,text:`version-${revision}`}:row),ledger:{frozenBody:'exact original request bytes'}}}); + const done=tx=>new Promise((resolve,reject)=>{tx.oncomplete=resolve;tx.onabort=()=>reject(tx.error);}),read=request=>new Promise((resolve,reject)=>{request.onsuccess=()=>resolve(request.result);request.onerror=()=>reject(request.error);}); + let raw=await new Promise((resolve,reject)=>{const request=indexedDB.open(databaseName,1);request.onupgradeneeded=()=>{for(const store of ['metadata','checkpoints','journal','files'])request.result.createObjectStore(store);};request.onsuccess=()=>resolve(request.result);request.onerror=()=>reject(request.error);}); + const initial=image(0),checksum=[...new Uint8Array(await crypto.subtle.digest('SHA-256',new TextEncoder().encode(JSON.stringify(initial))))].map(byte=>byte.toString(16).padStart(2,'0')).join('');let tx=raw.transaction(['metadata','checkpoints'],'readwrite'),terminal=done(tx);tx.objectStore('metadata').put({fence:0,throughRevision:0,checkpointRevision:0},key);tx.objectStore('checkpoints').put({image:initial,checksum},[key,0]);await terminal;raw.close(); + let storage=createBrowserStorage({name:databaseName,checkpointEvery:3}),leader; + try{ + leader=await storage.claim(scope);assert(JSON.stringify(await storage.recover(leader))===JSON.stringify(initial),'legacy checkpoint did not migrate');await storage.commit(leader,{image:image(1),files:[{id:'asset',bytes:new Blob(['immutable'])}]}); + raw=await new Promise((resolve,reject)=>{const request=indexedDB.open(databaseName);request.onsuccess=()=>resolve(request.result);request.onerror=()=>reject(request.error);}); + const row=await read(raw.transaction('journal').objectStore('journal').get([key,1]));assert(row.journalVersion===1&&!row.image&&JSON.stringify(row).length{const tx=raw.transaction('journal','readwrite'),wait=done(tx);tx.objectStore('journal').put(value,[key,2]);await wait;}; + for(const[kind,expected]of [['payload','recovery_checksum_mismatch'],['base','recovery_delta_base_mismatch'],['future','recovery_delta_interpreter_unavailable']]){const damaged=structuredClone(intact);if(kind==='payload')damaged.patches[0].value='tampered';else if(kind==='base')damaged.baseChecksum='0'.repeat(64);else damaged.journalVersion=99;await write(damaged);code=undefined;try{await storage.recover(leader);}catch(error){code=error.code;}assert(code===expected,`delta ${kind} corruption was not detected: ${code}`);await write(intact);} + await storage.commit(leader,{image:image(3)});assert((await storage.recover(leader)).rootRevision===3,'checkpoint after deltas failed');assert(await read(raw.transaction('journal').objectStore('journal').count())===0,'checkpoint did not compact prior deltas'); + const detached=await storage.recover(leader);detached.payload.rows[0].text='caller mutation';const fourth=image(4),fifth=image(5),one=storage.commit(leader,{image:fourth}),two=storage.commit(leader,{image:fifth});fifth.payload.rows[0].text='late mutation';await Promise.all([one,two]);assert(JSON.stringify(await storage.recover(leader))===JSON.stringify(image(5)),'queued commits or recovered callers mutated the cached base');await storage.commit(leader,{image:image(5)}); + const oldError=await new Promise(resolve=>{const request=indexedDB.open(databaseName,1);request.onerror=event=>{event.preventDefault();resolve(request.error.name);};request.onsuccess=()=>{request.result.close();resolve('accepted');};});assert(oldError==='VersionError','old database reader accepted the delta format'); + }finally{raw?.close();await leader?.release();storage.close();indexedDB.deleteDatabase(databaseName);} +} + async function importBackpressureCase(name,assert){ const root=crypto.randomUUID(),workspace=crypto.randomUUID(),registry=createRegistry(builtinComponents()),database=`${name}-import-pressure`,storage=createBrowserStorage({name:database}),original=createDocument(root,[{id:root,parent:null}],[]);let document=original,versions=new Map([[root,1]]),engine,calls=0,release;const gate=new Promise(resolve=>{release=resolve;}); const transport=async request=>{calls++;const body=JSON.parse(request.body);for(const[id,version]of Object.entries(body.expected_versions))assert(version===(versions.get(id)??0),'pressure lost a version fence');for(const operation of body.operations)document=applyChanges(document,operation.changes);for(const id of Object.keys(body.expected_versions))versions.set(id,(versions.get(id)??0)+1);const participants=Object.keys(body.expected_versions);return{status:200,body:{protocol_version:2,request_id:request.requestId,status:'applied',operation_ids:body.operations.map(row=>row.id),model:{documentFormatVersion:1,projectId:root,complete:false,removedCanvasIds:[],canvases:[...document.canvases.values()],elements:[...document.elements.values()],tombstones:Object.fromEntries(document.tombstones),versions:Object.fromEntries(participants.map(id=>[id,versions.get(id)]))}}};}; @@ -553,6 +573,8 @@ export async function run() { results.push('owned immutable recovery images survive IndexedDB checksums; mutable images and file descriptors are captured before asynchronous work'); await cooperativeCopyCase(assert); results.push('cooperative duplicate planning yields to actual browser tasks and cancellation without mutating the source'); + await deltaJournalCase(name,assert); + results.push('delta journals migrate legacy checkpoints, retain atomic files, verify hashes/base chains, reject old readers and compact through serialized immutable commits'); await importBackpressureCase(name,assert); results.push('blocked source durability permits a bounded local prefix, fences all network writes and recovers all records after real IndexedDB resumes'); await referenceCycleCase(name,assert); diff --git a/tests/headless/journal-delta.test.mjs b/tests/headless/journal-delta.test.mjs new file mode 100644 index 0000000..c74bf46 --- /dev/null +++ b/tests/headless/journal-delta.test.mjs @@ -0,0 +1,21 @@ +import {test} from 'node:test'; +import assert from 'node:assert/strict'; +import {diffJournal,applyJournal} from '../../src/lib/canvas-runtime/journal-delta.ts'; +const replay=(before,after)=>{const captured=JSON.stringify(before),patches=diffJournal(before,after);assert.ok(patches);const result=applyJournal(before,patches);assert.equal(JSON.stringify(result),JSON.stringify(after));assert.equal(JSON.stringify(before),captured);return patches;}; +test('journal patches preserve serialized property order, arrays, null and own dangerous-name payloads',()=>{ + for(const[before,after]of [[null,{a:1}],[[1,2,3],[1,9]],[[1],[2,3,4]],[{a:1,b:2},{b:2,a:1}],[JSON.parse('{"__proto__":{"x":1}}'),JSON.parse('{"__proto__":{"x":2}}')],[{nested:[{id:'A',data:{text:'before'}}]},{nested:[{id:'A',data:{text:'after'}},{id:'B'}]}]])replay(before,after); + assert.equal({}.x,undefined); +}); +test('a small node update and appended table row do not repeat the unchanged large payload',()=>{ + const rows=Array.from({length:1000},(_,i)=>({id:String(i),text:'x'.repeat(1000)})),before={rows,ledger:{body:'exact frozen bytes'}},after={rows:[...rows.slice(0,999),{...rows[999],text:'changed'},{id:'new',text:'new'}],ledger:before.ledger}; + const patches=replay(before,after);assert.ok(JSON.stringify(patches).length<1000);assert.equal(patches.length,2); +}); +test('malformed deltas reject invalid paths, array holes, unsupported operations and invalid ranges',()=>{ + for(const patch of [{op:'set',path:['missing','x'],value:1},{op:'set',path:['rows',9],value:1},{op:'set',path:['rows','0'],value:1},{op:'splice',path:['rows'],start:0,deleteCount:3,items:[]},{op:'splice',path:['rows'],start:-1,deleteCount:0,items:[]},{op:'remove',path:[]},{op:'set',path:[],value:undefined}])assert.throws(()=>applyJournal({rows:[1]},[patch]),/corrupt_recovery_delta/); + assert.throws(()=>applyJournal({},Array(10001).fill({op:'set',path:[],value:1})),/corrupt_recovery_delta/); + assert.equal(diffJournal(Array(10001).fill(0),Array(10001).fill(1)),null); +}); +test('mixed deterministic updates round-trip exactly across a chain without modifying earlier roots',()=>{ + let before={rows:[],nested:{a:1,b:2},history:[]};let seed=71;const random=()=>{seed=(seed*1664525+1013904223)>>>0;return seed;}; + for(let i=0;i<150;i++){const after=structuredClone(before);if(i%3===0)after.rows.push({id:i,text:String(random())});else if(i%3===1&&after.rows.length)after.rows[random()%after.rows.length].text=String(random());else if(after.rows.length)after.rows.splice(random()%after.rows.length,1);after.history.push(i);if(i%7===0)after.nested={b:i,a:random()};replay(before,after);before=after;} +});