Persist versioned Blob recovery cuts with cooperative JSON serialization
This commit is contained in:
@@ -62,3 +62,6 @@ export function isSealedRecoveryImage(image:RecoveryImage):boolean{return sealed
|
||||
|
||||
/** Exact UTF-8 JSON size, available only for codec-owned immutable images. */
|
||||
export function sealedRecoveryByteLength(image:RecoveryImage){return imageSizes.get(image);}
|
||||
|
||||
/** Internal codec ownership/size fact, never inferred from external Object.freeze. */
|
||||
export function encodedRecoveryByteLength(value:object){return encodedSizes.get(value);}
|
||||
|
||||
@@ -0,0 +1,32 @@
|
||||
import type {RecoveryImage} from './storage-port.ts';
|
||||
import {isSealedRecoveryImage,sealedRecoveryByteLength,encodedRecoveryByteLength} from './recovery-encoding.ts';
|
||||
import {createImportBudget,type ImportScheduling} from './import-budget.ts';
|
||||
import {fault} from '../canvas-core/values.ts';
|
||||
|
||||
function* stringTokens(value:string):Generator<string>{
|
||||
if(value.length<=4096){yield JSON.stringify(value);return;}
|
||||
yield '"';
|
||||
for(let start=0;start<value.length;){let end=Math.min(start+4096,value.length);const a=value.charCodeAt(end-1),b=value.charCodeAt(end);if(a>=0xd800&&a<=0xdbff&&b>=0xdc00&&b<=0xdfff)end--;yield JSON.stringify(value.slice(start,end)).slice(1,-1);start=end;}
|
||||
yield '"';
|
||||
}
|
||||
/** Only traverse codec-owned normalized JSON. Small owned subtrees use native stringify. */
|
||||
function* tokens(value:unknown):Generator<string>{
|
||||
if(value&&typeof value==='object'){
|
||||
const size=encodedRecoveryByteLength(value);if(size!==undefined&&size<=16384){yield JSON.stringify(value);return;}
|
||||
if(Array.isArray(value)){yield '[';for(let i=0;i<value.length;i++){if(i)yield ',';yield* tokens(value[i]);}yield ']';return;}
|
||||
yield '{';let first=true;for(const key of Object.keys(value)){if(!first)yield ',';first=false;yield* stringTokens(key);yield ':';yield* tokens((value as Record<string,unknown>)[key]);}yield '}';return;
|
||||
}
|
||||
if(typeof value==='string')yield* stringTokens(value);else yield JSON.stringify(value);
|
||||
}
|
||||
/** Preserve exact JSON.stringify bytes; scheduling never exposes a partial recovery image. */
|
||||
export async function serializeRecoveryImage(image:RecoveryImage,options:ImportScheduling&{check?:()=>void}={}):Promise<Blob>{
|
||||
options.check?.();
|
||||
if(!isSealedRecoveryImage(image)){const bytes=new Blob([JSON.stringify(image)],{type:'application/json'});options.check?.();return bytes;}
|
||||
const budget=createImportBudget(new AbortController().signal,options),parts:BlobPart[]=[];let pending:string[]=[],length=0;
|
||||
for(const token of tokens(image)){
|
||||
pending.push(token);length+=token.length;
|
||||
if(length>=8192){parts.push(pending.join(''));pending=[];length=0;options.check?.();await budget.step(128);options.check?.();}
|
||||
}
|
||||
if(pending.length)parts.push(pending.join(''));options.check?.();
|
||||
const bytes=new Blob(parts,{type:'application/json'});if(bytes.size!==sealedRecoveryByteLength(image))throw fault('recovery_encoding_size_mismatch');return bytes;
|
||||
}
|
||||
@@ -1,3 +1,4 @@
|
||||
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';
|
||||
@@ -9,10 +10,13 @@ export type {StorageScope,RecoveryImage,DurableBatch,StorageLeader,StoredFileInf
|
||||
type Metadata = { fence: number; throughRevision?: number; checkpointRevision?: number };
|
||||
type StoredImage = { readonly image: RecoveryImage; readonly checksum: string };
|
||||
type StoredDelta = {journalVersion:1|2;baseRevision:number;baseChecksum:string;header:Omit<RecoveryImage,'payload'>;patches:JournalPatch[];checksum:string};
|
||||
type StoredCut = StoredImage | StoredDelta;
|
||||
type StoredBlobImage={readonly checkpointVersion:1;readonly header:Omit<RecoveryImage,'payload'>;readonly bytes:Blob;readonly checksum:string};
|
||||
type StoredCut = StoredImage | StoredDelta | StoredBlobImage;
|
||||
const isDelta=(cut:StoredCut):cut is StoredDelta=>!('image' in cut)&&!('checkpointVersion' in cut);
|
||||
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;
|
||||
}
|
||||
@@ -20,6 +24,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};
|
||||
}
|
||||
async function bytesDigest(bytes:ArrayBuffer){return [...new Uint8Array(await crypto.subtle.digest('SHA-256',bytes))].map(value=>value.toString(16).padStart(2,'0')).join('');}
|
||||
async function blobDigest(bytes:Blob){return bytesDigest(await bytes.arrayBuffer());}
|
||||
function deltaBody(cut:StoredDelta){const {journalVersion,baseRevision,baseChecksum,header,patches}=cut;return{journalVersion,baseRevision,baseChecksum,header,patches};}
|
||||
const keyFor = (scope: StorageScope) => {
|
||||
if (![scope.accountId, scope.workspaceId, scope.projectId].every(value => typeof value === 'string' && value.length > 0)) throw fault('invalid_storage_scope');
|
||||
@@ -44,17 +50,18 @@ export function createBrowserStorage(options: { indexedDB?: IDBFactory; 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;
|
||||
let closed = false, openingFailure:Error|undefined;
|
||||
const handles = new Set<StorageLeader>(), lastCuts=new WeakMap<StorageLeader,StoredImage>(), writeTails=new WeakMap<StorageLeader,Promise<unknown>>(),lastCheckpoints=new WeakMap<StorageLeader,number>();
|
||||
const opening = new Promise<IDBDatabase>((resolve, reject) => {
|
||||
const request = factory.open(options.name ?? 'endoleap-headless', 4);
|
||||
const request = factory.open(options.name ?? 'endoleap-headless', 5);
|
||||
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(request.error);
|
||||
request.onblocked = () => reject(fault('storage_upgrade_blocked'));
|
||||
request.onerror = () => reject(closed?fault('storage_closed'):openingFailure??request.error);
|
||||
request.onblocked = () => {openingFailure=fault('storage_upgrade_blocked');reject(openingFailure);};
|
||||
request.onsuccess = () => {
|
||||
if (closed) { request.result.close(); reject(fault('storage_closed')); return; }
|
||||
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);
|
||||
};
|
||||
@@ -83,9 +90,9 @@ export function createBrowserStorage(options: { indexedDB?: IDBFactory; locks?:
|
||||
const captured=await execute(leader,['checkpoints','journal'],'readonly',async(transaction,metadata,key)=>{
|
||||
if(metadata.throughRevision===undefined)return{cuts:[] as StoredCut[],checkpointRevision:undefined};
|
||||
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 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])));
|
||||
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};
|
||||
@@ -94,9 +101,17 @@ export function createBrowserStorage(options: { indexedDB?: IDBFactory; locks?:
|
||||
// Hash and apply after the readonly transaction; awaiting crypto inside IDB would close it.
|
||||
for(const cut of captured.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' in cut||cut.journalVersion===1?image:deltaBody(cut))).checksum)throw fault('recovery_checksum_mismatch');previous=freeze({image,checksum:cut.checksum});
|
||||
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{
|
||||
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});
|
||||
}
|
||||
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;
|
||||
}
|
||||
@@ -162,7 +177,7 @@ export function createBrowserStorage(options: { indexedDB?: IDBFactory; locks?:
|
||||
if(digest.byteLength+78<fullBytes){stored={image,checksum:delta.checksum};candidate=delta;}
|
||||
}
|
||||
}
|
||||
if(!stored||!candidate){stored={image,checksum:(await imageDigest(image)).checksum};candidate=stored;}
|
||||
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;
|
||||
const receipt=await execute(leader,['checkpoints','journal','files',...depotStores],'readwrite',async(transaction,metadata,key)=>{
|
||||
const previous=metadata.throughRevision;
|
||||
@@ -174,14 +189,14 @@ export function createBrowserStorage(options: { indexedDB?: IDBFactory; locks?:
|
||||
}
|
||||
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(!('image' in cut)&&(cut.baseRevision!==previous||cut.baseChecksum!==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);
|
||||
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.
|
||||
if(checkpoint){
|
||||
transaction.objectStore('checkpoints').put(accepted,[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;
|
||||
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!};
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user