Persist shared asset bytes and project pins with Root commits
This commit is contained in:
@@ -0,0 +1,53 @@
|
||||
import {equal,fault,freeze} from '../canvas-core/values.ts';
|
||||
import {sharedAssetRef,validateSharedAsset,type AssetDepotPort,type DepotCut,type DepotEntry,type SharedAsset} from './asset-depot.ts';
|
||||
import type {StorageLeader,StorageScope} from './storage-port.ts';
|
||||
|
||||
export const depotStores=['asset-depot','asset-bytes','asset-pins'];
|
||||
type Fact={ref:SharedAsset;unreferencedAt?:number;purgedAt?:number};
|
||||
type Pins={version:1;rootRevision:number;complete:boolean;pins:readonly (readonly [string,readonly string[]])[]};
|
||||
const value=<T>(request:IDBRequest<T>)=>new Promise<T>((resolve,reject)=>{request.onsuccess=()=>resolve(request.result);request.onerror=()=>reject(request.error);});
|
||||
const prefix=(scope:StorageScope,authority:string)=>[scope.accountId,scope.workspaceId,authority];
|
||||
const range=(scope:StorageScope,authority:string)=>IDBKeyRange.bound(prefix(scope,authority),[...prefix(scope,authority),[]],false,true);
|
||||
function authority(value:string){if(typeof value!=='string'||!value||value.length>8192)throw fault('invalid_depot_cut');}
|
||||
function validateFact(row:Fact,scopeAuthority:string,id:string){validateSharedAsset(row?.ref);if(row.ref.authority!==scopeAuthority||row.ref.asset.id!==id)throw fault('corrupt_asset_depot');for(const n of [row.unreferencedAt,row.purgedAt])if(n!==undefined&&(!Number.isSafeInteger(n)||n<0))throw fault('corrupt_asset_depot');if(row.purgedAt!==undefined&&(row.unreferencedAt===undefined||row.purgedAt<row.unreferencedAt))throw fault('corrupt_asset_depot');}
|
||||
/** Detach descriptors before awaiting crypto. Blob data is immutable and never JSON cloned. */
|
||||
export function detachDepotCut(cut:DepotCut):DepotCut{
|
||||
if(!cut||cut.version!==1||typeof cut.complete!=='boolean'||!Array.isArray(cut.pins)||!Array.isArray(cut.writes))throw fault('invalid_depot_cut');authority(cut.authority);
|
||||
const seen=new Set<string>(),pins=cut.pins.map(row=>{if(!Array.isArray(row)||row.length!==2||typeof row[0]!=='string'||!row[0]||seen.has(row[0])||!Array.isArray(row[1])||!row[1].length||row[1].some(owner=>typeof owner!=='string'||!owner)||new Set(row[1]).size!==row[1].length)throw fault('invalid_depot_cut');seen.add(row[0]);return Object.freeze([row[0],Object.freeze([...row[1]])] as const);});
|
||||
const written=new Set<string>(),writes=cut.writes.map(row=>{validateSharedAsset(row.ref);if(row.ref.authority!==cut.authority||written.has(row.ref.asset.id)||row.bytes!==undefined&&!(row.bytes instanceof Blob))throw fault('invalid_depot_cut');written.add(row.ref.asset.id);return Object.freeze({ref:sharedAssetRef(row.ref),...(row.bytes?{bytes:row.bytes}:{})});});
|
||||
return Object.freeze({version:1,authority:cut.authority,complete:cut.complete,pins:Object.freeze(pins),writes:Object.freeze(writes)});
|
||||
}
|
||||
export async function verifyDepotBytes(cut:DepotCut){
|
||||
for(const row of cut.writes)if(row.bytes){if(row.bytes.size!==row.ref.asset.bytes)throw fault('shared_asset_bytes_changed',[row.ref.asset.id]);const digest=[...new Uint8Array(await crypto.subtle.digest('SHA-256',await row.bytes.arrayBuffer()))].map(n=>n.toString(16).padStart(2,'0')).join('');if(digest!==row.ref.digest)throw fault('shared_asset_bytes_changed',[row.ref.asset.id]);}
|
||||
}
|
||||
/** Must run in the same transaction as the project Root, including idempotent reindexing. */
|
||||
export async function writeDepotCut(tx:IDBTransaction,scope:StorageScope,rootRevision:number,cut:DepotCut){
|
||||
const base=prefix(scope,cut.authority),facts=tx.objectStore('asset-depot'),bytes=tx.objectStore('asset-bytes');
|
||||
for(const row of cut.writes){const key=[...base,row.ref.asset.id],previous=await value<Fact|undefined>(facts.get(key));if(previous){validateFact(previous,cut.authority,row.ref.asset.id);if(!equal(previous.ref,row.ref))throw fault('shared_asset_identity_conflict',[row.ref.asset.id]);}
|
||||
// Digest-addressed bytes are shared across resource IDs and projects in this authority.
|
||||
if(row.bytes)bytes.put(row.bytes,[...base,row.ref.digest]);
|
||||
facts.put({...previous,ref:row.ref,...(row.bytes?{unreferencedAt:undefined,purgedAt:undefined}:{})},key);
|
||||
}
|
||||
tx.objectStore('asset-pins').put({version:1,rootRevision,complete:cut.complete,pins:cut.pins} satisfies Pins,[...base,scope.projectId]);
|
||||
}
|
||||
type Execute=<T>(leader:StorageLeader,stores:string[],mode:IDBTransactionMode,run:(tx:IDBTransaction)=>Promise<T>)=>Promise<T>;
|
||||
export function createDepotStorage(execute:Execute):AssetDepotPort{return Object.freeze({
|
||||
async read(leader,scopeAuthority,id){authority(scopeAuthority);if(typeof id!=='string'||!id)throw fault('invalid_resource_id');return execute(leader,depotStores,'readonly',async tx=>{const base=prefix(leader.scope,scopeAuthority),row=await value<Fact|undefined>(tx.objectStore('asset-depot').get([...base,id]));if(!row)return;validateFact(row,scopeAuthority,id);const bytes=row.purgedAt===undefined?await value<Blob|undefined>(tx.objectStore('asset-bytes').get([...base,row.ref.digest])):undefined;if(bytes!==undefined&&(!(bytes instanceof Blob)||bytes.size!==row.ref.asset.bytes))throw fault('corrupt_asset_depot');return freeze({...row,...(bytes?{bytes}:{})}) as DepotEntry;});},
|
||||
async collect(leader,input){authority(input.authority);const {now,retentionMs,limit,after}=input;if(!Number.isSafeInteger(now)||now<0||!Number.isSafeInteger(retentionMs)||retentionMs<0||!Number.isSafeInteger(limit)||limit<1||limit>128||after!==undefined&&(typeof after!=='string'||!after))throw fault('invalid_depot_collection');
|
||||
return execute(leader,depotStores,'readwrite',async tx=>{
|
||||
const scope=leader.scope,metadata=tx.objectStore('metadata'),keys=await value(metadata.getAllKeys()),heads=await value<{throughRevision?:number}[]>(metadata.getAll()),base=prefix(scope,input.authority),manifests=tx.objectStore('asset-pins'),blockedProjects:string[]=[],pinned=new Set<string>();
|
||||
for(let i=0;i<keys.length;i++){let parts:unknown;try{parts=JSON.parse(String(keys[i]));}catch{throw fault('corrupt_storage_scope');}if(!Array.isArray(parts)||parts.length!==3||parts.some(p=>typeof p!=='string'||!p))throw fault('corrupt_storage_scope');if(parts[0]!==scope.accountId||parts[1]!==scope.workspaceId||heads[i].throughRevision===undefined)continue;
|
||||
const pins=await value<Pins|undefined>(manifests.get([...base,parts[2]]));if(!pins||pins.version!==1||pins.rootRevision!==heads[i].throughRevision||pins.complete!==true){blockedProjects.push(parts[2]);continue;}
|
||||
if(!Array.isArray(pins.pins))throw fault('corrupt_depot_pins');for(const row of pins.pins){if(!Array.isArray(row)||row.length!==2||typeof row[0]!=='string'||!row[0]||!Array.isArray(row[1])||!row[1].length||row[1].some(owner=>typeof owner!=='string'||!owner))throw fault('corrupt_depot_pins');pinned.add(row[0]);}
|
||||
}
|
||||
if(blockedProjects.length)return freeze({complete:false,blockedProjects,markedIds:[],purgedIds:[],pinnedIds:[]});
|
||||
const store=tx.objectStore('asset-depot'),factKeys=await value(store.getAllKeys(range(scope,input.authority))),rows=await value<Fact[]>(store.getAll(range(scope,input.authority))),markedIds:string[]=[],purgedIds:string[]=[],pinnedIds:string[]=[];
|
||||
for(let i=0;i<rows.length;i++){const key=factKeys[i];if(!Array.isArray(key)||key.length!==4||typeof key[3]!=='string')throw fault('corrupt_asset_depot');validateFact(rows[i],input.authority,key[3]);}
|
||||
const candidates=rows.map((row,i)=>({row,key:factKeys[i] as string[]})).filter(({key})=>after===undefined||key[3]>after).slice(0,limit+1),page=candidates.slice(0,limit);
|
||||
for(const {row,key}of page){const id=key[3];if(pinned.has(id)){pinnedIds.push(id);if(row.purgedAt===undefined&&row.unreferencedAt!==undefined){delete row.unreferencedAt;store.put(row,key);}continue;}if(row.purgedAt!==undefined)continue;if(row.unreferencedAt===undefined){row.unreferencedAt=now;markedIds.push(id);store.put(row,key);}else if(now>=row.unreferencedAt&&now-row.unreferencedAt>=retentionMs){row.purgedAt=now;purgedIds.push(id);store.put(row,key);}}
|
||||
// Preserve a digest while any other live resource identity still owns its bytes.
|
||||
for(const digest of new Set(page.filter(({row})=>row.purgedAt===now).map(({row})=>row.ref.digest)))if(!rows.some(row=>row.ref.digest===digest&&row.purgedAt===undefined))tx.objectStore('asset-bytes').delete([...base,digest]);
|
||||
return freeze({complete:true,blockedProjects:[],markedIds,purgedIds,pinnedIds,...(candidates.length>limit?{nextAfter:page.at(-1)!.key[3]}:{})});
|
||||
});
|
||||
},
|
||||
});}
|
||||
@@ -0,0 +1,35 @@
|
||||
import type {Id} from '../canvas-core/types.ts';
|
||||
import {equal,fault,freeze,readMap} from '../canvas-core/values.ts';
|
||||
import type {StoredAsset,LoadingAssetAdmission} from './resources.ts';
|
||||
import type {StorageLeader,StorageScope} from './storage-port.ts';
|
||||
|
||||
export type SharedAsset={readonly version:1;readonly authority:string;readonly digest:string;readonly asset:StoredAsset};
|
||||
export type SharedAssetBinding=SharedAsset&{readonly loadingAdmission?:(LoadingAssetAdmission&{readonly operationId:Id})};
|
||||
export type AssetAuthority={readonly scope:string;resolve(id:Id,signal:AbortSignal):Promise<SharedAsset>;read(id:Id,signal:AbortSignal):Promise<Blob>};
|
||||
export type DepotWrite={readonly ref:SharedAsset;readonly bytes?:Blob};
|
||||
export type DepotCut={readonly version:1;readonly authority:string;readonly complete:boolean;readonly pins:readonly (readonly [Id,readonly string[]])[];readonly writes:readonly DepotWrite[]};
|
||||
export type DepotEntry={readonly ref:SharedAsset;readonly bytes?:Blob;readonly unreferencedAt?:number;readonly purgedAt?:number};
|
||||
export type DepotCollection={readonly complete:boolean;readonly blockedProjects:readonly Id[];readonly markedIds:readonly Id[];readonly purgedIds:readonly Id[];readonly pinnedIds:readonly Id[];readonly nextAfter?:Id};
|
||||
export type AssetDepotPort={
|
||||
read(leader:StorageLeader,authority:string,id:Id):Promise<DepotEntry|undefined>;
|
||||
collect(leader:StorageLeader,input:{readonly authority:string;readonly now:number;readonly retentionMs:number;readonly limit:number;readonly after?:Id}):Promise<DepotCollection>;
|
||||
};
|
||||
export function validateSharedAsset(value:SharedAsset):void{
|
||||
if(!value||value.version!==1||typeof value.authority!=='string'||!value.authority||value.authority.length>8192||typeof value.digest!=='string'||!/^([a-f0-9]{64})$/.test(value.digest))throw fault('invalid_shared_asset');
|
||||
const asset=value.asset;if(!asset||typeof asset.id!=='string'||!asset.id||typeof asset.url!=='string'||!asset.url||asset.url.length>8192||![asset.width,asset.height,asset.bytes].every(n=>Number.isSafeInteger(n)&&n>0)||asset.width*asset.height>40_000_000||asset.bytes>20*1024*1024||!['image/png','image/jpeg','image/gif','image/webp'].includes(asset.mime_type))throw fault('invalid_shared_asset');
|
||||
}
|
||||
export function sharedAssetRef(value:SharedAsset):SharedAsset{validateSharedAsset(value);return freeze({version:1 as const,authority:value.authority,digest:value.digest,asset:{...value.asset}});}
|
||||
const same=(a:SharedAsset,b:SharedAsset)=>equal(sharedAssetRef(a),sharedAssetRef(b));
|
||||
/** Immutable metadata cache only. Every new adoption resolves current permission first. */
|
||||
export function createAssetDepot(options:{scope:Pick<StorageScope,'accountId'|'workspaceId'>;port:AssetDepotPort;current():boolean;digest(bytes:Blob,signal?:AbortSignal):Promise<string>}){
|
||||
const scope=freeze({...options.scope});let closed=false;const records=new Map<string,SharedAsset>(),controllers=new Set<AbortController>();
|
||||
const check=(leader:StorageLeader)=>{if(closed||!options.current())throw fault('identity_changed');if(!leader.active||leader.scope.accountId!==scope.accountId||leader.scope.workspaceId!==scope.workspaceId)throw fault('asset_depot_scope_mismatch');};
|
||||
function remember(ref:SharedAsset){validateSharedAsset(ref);const key=JSON.stringify([ref.authority,ref.asset.id]),previous=records.get(key);if(previous&&!same(previous,ref))throw fault('shared_asset_identity_conflict',[ref.asset.id]);records.set(key,sharedAssetRef(ref));}
|
||||
return Object.freeze({scope,getResources:()=>readMap(records),
|
||||
async authorize(leader:StorageLeader,authority:AssetAuthority,id:Id,signal:AbortSignal){check(leader);const controller=new AbortController(),combined=AbortSignal.any([signal,controller.signal]);controllers.add(controller);try{const ref=await authority.resolve(id,combined);check(leader);combined.throwIfAborted();validateSharedAsset(ref);if(ref.authority!==authority.scope||ref.asset.id!==id)throw fault('asset_authorization_mismatch');const cached=await options.port.read(leader,authority.scope,id);check(leader);combined.throwIfAborted();if(cached&&!same(cached.ref,ref))throw fault('shared_asset_identity_conflict',[id]);let bytes=cached?.bytes;if(!bytes)bytes=await authority.read(id,combined);check(leader);combined.throwIfAborted();if(bytes.size!==ref.asset.bytes||await options.digest(bytes,combined)!==ref.digest)throw fault('shared_asset_bytes_changed',[id]);check(leader);combined.throwIfAborted();remember(ref);return Object.freeze({ref,bytes});}finally{controllers.delete(controller);}},
|
||||
async read(leader:StorageLeader,ref:SharedAsset,signal?:AbortSignal){check(leader);signal?.throwIfAborted();const row=await options.port.read(leader,ref.authority,ref.asset.id);check(leader);signal?.throwIfAborted();if(!row)return;if(!same(row.ref,ref))throw fault('shared_asset_identity_conflict',[ref.asset.id]);if(row.bytes&&(row.bytes.size!==ref.asset.bytes||await options.digest(row.bytes,signal)!==ref.digest))throw fault('shared_asset_bytes_changed',[ref.asset.id]);check(leader);signal?.throwIfAborted();remember(ref);return row.bytes;},
|
||||
async collect(leader:StorageLeader,input:Parameters<AssetDepotPort['collect']>[1]){check(leader);const result=await options.port.collect(leader,input);check(leader);return result;},
|
||||
close(){closed=true;for(const controller of controllers)controller.abort();controllers.clear();records.clear();},
|
||||
});
|
||||
}
|
||||
export type AssetDepot=ReturnType<typeof createAssetDepot>;
|
||||
@@ -0,0 +1,28 @@
|
||||
import type {Registry} from '../canvas-core/types.ts';
|
||||
import {fault} from '../canvas-core/values.ts';
|
||||
import type {EngineSnapshot} from './contracts.ts';
|
||||
import type {DepotCut,DepotWrite} from './asset-depot.ts';
|
||||
import {sharedAssetRef} from './asset-depot.ts';
|
||||
import type {StorageLeader,StoragePort} from './storage-port.ts';
|
||||
import {resourceRetentionPins} from './resource-retention.ts';
|
||||
|
||||
/** Derive one project's shared-byte manifest from the exact Root being persisted. */
|
||||
export async function prepareDepotCut(input:{snapshot:EngineSnapshot;registry:Registry;storage:StoragePort;leader:StorageLeader;authority:string;files:ReadonlyMap<string,Blob>;displays:ReadonlyMap<string,number>;temporary:ReadonlyMap<string,readonly string[]>}):Promise<DepotCut>{
|
||||
const {snapshot,storage,leader,authority}=input,pins=resourceRetentionPins(snapshot,input.registry,input.displays,input.temporary),writes:DepotWrite[]=[];
|
||||
if(!storage.assetDepot)throw fault('asset_depot_unavailable');
|
||||
for(const resource of snapshot.runtime.resources.values())if(resource.stored&&resource.digest){
|
||||
const ref=sharedAssetRef({version:1,authority,digest:resource.digest,asset:resource.stored});
|
||||
if(ref.asset.id!==resource.id||ref.asset.bytes!==resource.byteLength)throw fault('shared_asset_identity_conflict',[resource.id]);
|
||||
let bytes:Blob|undefined;
|
||||
if(resource.retirement?.purgedAt===undefined&&pins.pins.has(resource.id)){
|
||||
const cached=await storage.assetDepot.read(leader,authority,resource.id);
|
||||
// The store enforces immutable identity even when bytes are already present.
|
||||
if(!cached?.bytes)bytes=input.files.get(resource.id)??await storage.readFile(leader,resource.id);
|
||||
}
|
||||
writes.push({ref,...(bytes?{bytes}:{})});
|
||||
}
|
||||
// A source/run envelope may refer to resources outside this project's upload table.
|
||||
// Its complete resource interpreter must be installed before global deletion is allowed.
|
||||
const complete=pins.complete&&!snapshot.runtime.loading&&![...(snapshot.runtime.imports?.values()??[])].some(job=>!['completed','partial','cancelled'].includes(job.status))&&snapshot.runtime.runs.size===0;
|
||||
return{version:1,authority,complete,pins:[...pins.pins].map(([id,owners])=>[id,[...owners]] as const),writes};
|
||||
}
|
||||
@@ -1,3 +1,4 @@
|
||||
import {prepareDepotCut} from './depot-cut.ts';
|
||||
import {loadingAssetURLs} from './loading-assets.ts';
|
||||
import {loadingImportAdmission,validateLoadingCopySource} from './loading-import.ts';
|
||||
import {loadingCanSend,mergeLoadingRead,loadingQueryCoverage,assertLoadingChanges,loadingCommand,validateLoadingRuntime,validatePrefixExtension,validateLoadingCompletion,type LoadingState} from './loading.ts';
|
||||
@@ -169,7 +170,8 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
const sourceFiles=importFiles(snapshot.runtime.imports);
|
||||
const files = [...fileWrites].filter(([id]) => snapshot.runtime.resources.has(id)||sourceFiles.has(id)).map(([id, bytes]) => ({ id, bytes }));
|
||||
const deleteFiles=[...[...snapshot.runtime.resources.values()].filter(resource=>resource.retirement?.purgedAt!==undefined&&!durablyPurged.has(resource.id)).map(resource=>resource.id),...[...sourceFiles].filter(([id,entry])=>entry.file.retirement?.purgedAt!==undefined&&!durablyPurged.has(id)).map(([id])=>id),...[...(snapshot.runtime.orphanFiles??[])].filter(([id,file])=>file.retirement?.purgedAt!==undefined&&!durablyPurged.has(id)).map(([id])=>id)];
|
||||
const receipt = await options.storage.commit(leader, { image: encodeRecovery(snapshot), files,deleteFiles });
|
||||
const depot=options.assets?.authority&&options.storage.assetDepot?await prepareDepotCut({snapshot,registry:options.registry,storage:options.storage,leader,authority:options.assets.authority.scope,files:fileWrites,displays:displayPins,temporary:admissionPins(snapshot)}):undefined;
|
||||
const receipt = await options.storage.commit(leader, { image: encodeRecovery(snapshot), files,deleteFiles,...(depot?{depot}:{}) });
|
||||
if (receipt.durableRootRevision !== revision) throw fault('invalid_durable_receipt');
|
||||
durableRootRevision = revision; pending.delete(revision);
|
||||
for(const id of deleteFiles){durablyPurged.add(id);collectionReservations.delete(id);}
|
||||
|
||||
@@ -12,6 +12,7 @@ export type LoadingAssetAdmission = {readonly version:1;readonly sourceToken:str
|
||||
export type ResourceRecord = { readonly loadingAdmission?:LoadingAssetAdmission; readonly id: Id; readonly operationId: Id; readonly name: string; readonly mimeType: string; readonly byteLength: number; readonly sizing?: DefaultImageSizing; readonly image?:ImageDecodeState; readonly digest?: string; readonly retirement?: {readonly version:1;readonly unreferencedAt:number;readonly purgedAt?:number}; readonly stored?: StoredAsset };
|
||||
export type DisplayHandle = { readonly url: string; release(): void };
|
||||
export type AssetPort = {
|
||||
readonly authority?:import('./asset-depot.ts').AssetAuthority;
|
||||
sourceURL(resourceId: Id): string;
|
||||
displayURL(resourceId: Id): string;
|
||||
upload(request: FrozenRequest, bytes: Blob, signal: AbortSignal): Promise<OperationStatus>;
|
||||
|
||||
@@ -3,11 +3,12 @@ import type { Json } from '../canvas-core/types.ts';
|
||||
/** Portable durable-store contract. Browser database types belong to the adapter. */
|
||||
export type StorageScope = { readonly accountId: string; readonly workspaceId: string; readonly projectId: string };
|
||||
export type RecoveryImage = { readonly formatVersion: 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8 | 9 | 10 | 11 | 12; readonly rootRevision: number; readonly localRevision: number; readonly payload: Json };
|
||||
export type DurableBatch = { readonly image: RecoveryImage; readonly deleteFiles?:readonly string[]; readonly files?: readonly { readonly id: string; readonly bytes: Blob }[] };
|
||||
export type DurableBatch = { readonly image: RecoveryImage; readonly deleteFiles?:readonly string[]; readonly depot?:import('./asset-depot.ts').DepotCut; readonly files?: readonly { readonly id: string; readonly bytes: Blob }[] };
|
||||
export type StorageLeader = { readonly scope: StorageScope; readonly fence: number; readonly active: boolean; release(): Promise<void> };
|
||||
export type StoredFileInfo={readonly id:string;readonly byteLength:number;readonly mimeType:string};
|
||||
export type FileInventoryPage={readonly files:readonly StoredFileInfo[];readonly rootRevision:number;readonly nextAfter?:string};
|
||||
export type StoragePort = {
|
||||
readonly assetDepot?:import('./asset-depot.ts').AssetDepotPort;
|
||||
scanFiles?(leader:StorageLeader,input:{readonly expectedRootRevision:number;readonly after?:string;readonly limit:number}):Promise<FileInventoryPage>;
|
||||
readonly fileDeletion?: 'atomic-v1';
|
||||
claim(scope: StorageScope): Promise<StorageLeader>;
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import {depotStores,createDepotStorage,detachDepotCut,verifyDepotBytes,writeDepotCut} from './asset-depot-storage.ts';
|
||||
import { applyJournal, diffJournal, type JournalPatch } from './journal-delta.ts';
|
||||
import { isSealedRecoveryImage } from './recovery-encoding.ts';
|
||||
import type { Json } from '../canvas-core/types.ts';
|
||||
@@ -45,9 +46,9 @@ export function createBrowserStorage(options: { indexedDB?: IDBFactory; locks?:
|
||||
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);
|
||||
const request = factory.open(options.name ?? 'endoleap-headless', 3);
|
||||
request.onupgradeneeded = () => {
|
||||
for (const name of ['metadata', 'checkpoints', 'journal', 'files']) if (!request.result.objectStoreNames.contains(name)) request.result.createObjectStore(name);
|
||||
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'));
|
||||
@@ -100,6 +101,7 @@ export function createBrowserStorage(options: { indexedDB?: IDBFactory; locks?:
|
||||
}
|
||||
return Object.freeze({
|
||||
fileDeletion:'atomic-v1' as const,
|
||||
assetDepot:createDepotStorage((leader,stores,mode,run)=>execute(leader,stores,mode,tx=>run(tx))),
|
||||
async claim(scope: StorageScope) {
|
||||
const key = keyFor(scope);
|
||||
if (closed) throw fault('storage_closed');
|
||||
@@ -128,6 +130,7 @@ export function createBrowserStorage(options: { indexedDB?: IDBFactory; locks?:
|
||||
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 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);
|
||||
if(deleteFiles.length){
|
||||
type FileProof={id?:string;version?:number;byteLength?:number;mimeType?:string;digest?:string;fileId?:string;retirement?:{version:number;unreferencedAt:number;purgedAt?:number}};
|
||||
@@ -141,6 +144,7 @@ export function createBrowserStorage(options: { indexedDB?: IDBFactory; locks?:
|
||||
}
|
||||
check(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');}
|
||||
@@ -149,15 +153,16 @@ export function createBrowserStorage(options: { indexedDB?: IDBFactory; locks?:
|
||||
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 receipt=await execute(leader,['checkpoints','journal','files',...depotStores],'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(!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');if(depot)await writeDepotCut(transaction,leader.scope,image.rootRevision,depot);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');
|
||||
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]);
|
||||
// A format transition must not depend on a newer partial checkpoint in an older reader.
|
||||
|
||||
Reference in New Issue
Block a user