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

Deduplicate stored upload files and retain adoption bytes until their Root is durable

Browse Source
This commit is contained in:
Codex Bot
2026-11-09 16:31:31 +08:00
parent a626e6c300
commit 902e30a65b
10 changed files with 167 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/canvas-runtime/asset-depot-storage.ts
+39 -1
View File
@@ -1,6 +1,6 @@
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';
import type {StorageLeader,StorageScope,RecoveryImage} from './storage-port.ts';
export const depotStores=['asset-depot','asset-bytes','asset-pins'];
type Fact={ref:SharedAsset;unreferencedAt?:number;purgedAt?:number};
@@ -30,6 +30,44 @@ export async function writeDepotCut(tx:IDBTransaction,scope:StorageScope,rootRev
}
tx.objectStore('asset-pins').put({version:1,rootRevision,complete:cut.complete,pins:cut.pins} satisfies Pins,[...base,scope.projectId]);
}
/** DB v4 file rows may refer to immutable shared bytes after a proven upload ACK. */
export type DepotFile={readonly fileVersion:1;readonly kind:'asset-depot';readonly authority:string;readonly resourceId:string;readonly digest:string;readonly byteLength:number;readonly mimeType:string};
function depotFile(row:unknown,id:string):DepotFile{
const file=row as DepotFile;
if(!file||file.fileVersion!==1)throw fault('file_interpreter_unavailable');
authority(file.authority);
if(file.kind!=='asset-depot'||file.resourceId!==id||typeof file.digest!=='string'||!/^[a-f0-9]{64}$/.test(file.digest)||!Number.isSafeInteger(file.byteLength)||file.byteLength<1||file.byteLength>20*1024*1024||typeof file.mimeType!=='string')throw fault('corrupt_file_store');
return file;
}
/** Undefined means bytes were explicitly evicted; identity metadata remains. */
export async function resolveDepotFile(tx:IDBTransaction,scope:StorageScope,id:string,row:unknown):Promise<Blob|undefined>{
if(row===undefined||row instanceof Blob)return row;
const file=depotFile(row,id),base=prefix(scope,file.authority),fact=await value<Fact|undefined>(tx.objectStore('asset-depot').get([...base,id]));
if(!fact)throw fault('corrupt_file_store');validateFact(fact,file.authority,id);
if(fact.ref.digest!==file.digest||fact.ref.asset.bytes!==file.byteLength)throw fault('corrupt_file_store');
if(fact.purgedAt!==undefined)return;
const bytes=await value<Blob|undefined>(tx.objectStore('asset-bytes').get([...base,file.digest]));
if(bytes===undefined)return;
if(!(bytes instanceof Blob)||bytes.size!==file.byteLength)throw fault('corrupt_file_store');
// Preserve the original local File MIME even if the server detected another MIME.
return bytes.type===file.mimeType?bytes:bytes.slice(0,bytes.size,file.mimeType);
}
/** Convert only Root-proven stored uploads, after shared bytes and local files are written. */
export async function promoteDepotFiles(tx:IDBTransaction,scope:StorageScope,key:string,image:RecoveryImage,cut:DepotCut){
type Resource={id:string;digest?:string;byteLength:number;mimeType:string;stored?:SharedAsset['asset'];retirement?:{purgedAt?:number}};
const payload=image.payload as {runtime?:{resources?:readonly (readonly [string,Resource])[]}},entries=payload?.runtime?.resources;
if(entries===undefined)return;if(!Array.isArray(entries))throw fault('corrupt_recovery_image');
const resources=new Map<string,Resource>(entries),files=tx.objectStore('files'),base=prefix(scope,cut.authority);
for(const write of cut.writes){const id=write.ref.asset.id,resource=resources.get(id);if(!resource?.stored||resource.retirement?.purgedAt!==undefined)continue;
if(resource.id!==id||resource.digest!==write.ref.digest||resource.byteLength!==write.ref.asset.bytes||typeof resource.mimeType!=='string'||!equal(resource.stored,write.ref.asset))throw fault('shared_asset_identity_conflict',[id]);
const local=await value<unknown>(files.get([key,id]));if(local===undefined)continue;
if(!(local instanceof Blob)){const pointer=depotFile(local,id);if(pointer.authority!==cut.authority||pointer.digest!==write.ref.digest||pointer.byteLength!==resource.byteLength||pointer.mimeType!==resource.mimeType)throw fault('shared_asset_identity_conflict',[id]);continue;}
const bytes=await value<Blob|undefined>(tx.objectStore('asset-bytes').get([...base,write.ref.digest]));
if(!bytes)continue;
if(!(bytes instanceof Blob)||bytes.size!==resource.byteLength||local.size!==resource.byteLength||local.type!==resource.mimeType)throw fault('shared_asset_bytes_changed',[id]);
files.put({fileVersion:1,kind:'asset-depot',authority:cut.authority,resourceId:id,digest:write.ref.digest,byteLength:resource.byteLength,mimeType:resource.mimeType} satisfies DepotFile,[key,id]);
}
}
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;});},
src/lib/canvas-runtime/depot-cut.ts
+2 -2
View File
@@ -7,7 +7,7 @@ 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[]>;adoptedFiles?:ReadonlyMap<string,Blob>}):Promise<DepotCut>{
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[]>;adoptedFiles?:ReadonlyMap<string,{readonly bytes:Blob;readonly rootRevision:number}>}):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){
@@ -23,7 +23,7 @@ export async function prepareDepotCut(input:{snapshot:EngineSnapshot;registry:Re
}
for(const [id,binding]of snapshot.runtime.assetBindings??[]){
if(binding.authority!==authority||binding.asset.id!==id||snapshot.runtime.resources.has(id))throw fault('shared_asset_identity_conflict',[id]);
const bytes=input.adoptedFiles?.get(id);writes.push({ref:sharedAssetRef(binding),...(bytes?{bytes}:{})});
const pending=input.adoptedFiles?.get(id),bytes=pending&&pending.rootRevision<=snapshot.rootRevision?pending.bytes:undefined;writes.push({ref:sharedAssetRef(binding),...(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.
src/lib/canvas-runtime/engine.ts
+5 -5
View File
@@ -79,7 +79,7 @@ export async function openProjectEngine(options: EngineOptions) {
const network = createNetworkGate(2);
const depot=options.assetDepot??(options.storage.assetDepot?createAssetDepot({scope:options.scope,port:options.storage.assetDepot,current:()=>!closing&&!closed&&(!options.isCurrentIdentity||options.isCurrentIdentity()),digest:sourceDigest}):undefined);
if(depot&&(depot.scope.accountId!==options.scope.accountId||depot.scope.workspaceId!==options.scope.workspaceId))throw fault('asset_depot_scope_mismatch');
const adoptionTasks=new Map<AbortController,Promise<Result>>(),adoptedFiles=new Map<Id,Blob>();
const adoptionTasks=new Map<AbortController,Promise<Result>>(),adoptedFiles=new Map<Id,{readonly bytes:Blob;readonly rootRevision:number}>();
let adopting:SharedAsset|undefined;
const runClock: Clock = options.clock ?? { now: () => Date.now(), setTimer: (fn, delay) => setTimeout(fn, delay), clearTimer: handle => clearTimeout(handle as ReturnType<typeof setTimeout>) };
const displays = new Set<() => void>();
@@ -198,7 +198,7 @@ export async function openProjectEngine(options: EngineOptions) {
durableRootRevision = revision; pending.delete(revision);
for(const id of deleteFiles){durablyPurged.add(id);collectionReservations.delete(id);}
for (const file of files) fileWrites.delete(file.id);
for(const id of snapshot.runtime.assetBindings?.keys()??[])adoptedFiles.delete(id);
for(const [id,entry]of adoptedFiles)if(entry.rootRevision<=snapshot.rootRevision&&depot?.writes.some(write=>write.ref.asset.id===id&&write.bytes===entry.bytes))adoptedFiles.delete(id);
if (editor.getSnapshot().runtime.ledger.durableGeneration < snapshot.runtime.ledger.generation) mutate(ledger => ledger.markDurable(snapshot.runtime.ledger.generation));
}
persistenceFailed = false;
@@ -922,9 +922,9 @@ export async function openProjectEngine(options: EngineOptions) {
assertWritable(false);signal.throwIfAborted();if(authorized.ref.asset.url!==options.assets!.sourceURL(id))throw fault('asset_authorization_mismatch');
if(staged.length||adopting)throw fault('reentrant_dispatch');const runtime=editor.getSnapshot().runtime;
if(runtime.resources.has(id)||runtime.orphanFiles?.has(id)||importFiles(runtime.imports).has(id)||fileWrites.has(id)||collectionReservations.has(id))throw fault('resource_identity_conflict',[id]);
const previous=adoptedFiles.get(id);adopting=authorized.ref;adoptedFiles.set(id,authorized.bytes);
const previous=adoptedFiles.get(id);adopting=authorized.ref;adoptedFiles.set(id,{bytes:authorized.bytes,rootRevision:Number.MAX_SAFE_INTEGER});
let committed=false;
try{const result=editor.transaction(tx=>callback(freeze({id,src:authorized.ref.asset.url,asset:authorized.ref.asset}),tx),{label:'Add existing image'});committed=result.status==='applied'&&result.operationIds.length>0;return result;}
try{const result=editor.transaction(tx=>callback(freeze({id,src:authorized.ref.asset.url,asset:authorized.ref.asset}),tx),{label:'Add existing image'});committed=result.status==='applied'&&result.operationIds.length>0;if(committed)adoptedFiles.set(id,{bytes:authorized.bytes,rootRevision:editor.getSnapshot().rootRevision});return result;}
finally{adopting=undefined;if(!committed){if(previous)adoptedFiles.set(id,previous);else adoptedFiles.delete(id);}}
})();
adoptionTasks.set(controller,task);void task.finally(()=>adoptionTasks.delete(controller)).catch(()=>{});return task;
@@ -955,7 +955,7 @@ export async function openProjectEngine(options: EngineOptions) {
const binding=editor.getSnapshot().runtime.assetBindings?.get(id);
const key=JSON.stringify(binding?['shared-asset',id,binding.authority,binding.digest]:retained?['asset',id,retained.operationId,retained.retirement?.purgedAt??null]:orphan?['orphan',id,orphan.digest]:source?['source',id,source.sha256??null]:['unmanaged',id]);
lease=displayCache.acquire(displayOwner,key,async signal=>{
let bytes=adoptedFiles.get(id)??fileWrites.get(id)??await options.storage.readFile(leader,id);
let bytes=adoptedFiles.get(id)?.bytes??fileWrites.get(id)??await options.storage.readFile(leader,id);
if(!bytes&&binding&&depot)bytes=await depot.read(leader,binding,signal);
assertDisplayIdentity();if(signal.aborted)throw fault('display_aborted');
if(orphan&&(!bytes||bytes.size!==orphan.byteLength||bytes.type!==orphan.mimeType||await sourceDigest(bytes,signal)!==orphan.digest))throw fault('file_inventory_changed',[id]);
src/lib/canvas-runtime/storage.ts
+11 -9
View File
@@ -1,4 +1,4 @@
import {depotStores,createDepotStorage,detachDepotCut,verifyDepotBytes,writeDepotCut} from './asset-depot-storage.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 } from './recovery-encoding.ts';
import type { Json } from '../canvas-core/types.ts';
@@ -46,7 +46,7 @@ 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', 3);
const request = factory.open(options.name ?? 'endoleap-headless', 4);
request.onupgradeneeded = () => {
for (const name of ['metadata', 'checkpoints', 'journal', 'files',...depotStores]) if (!request.result.objectStoreNames.contains(name)) request.result.createObjectStore(name);
};
@@ -158,13 +158,14 @@ export function createBrowserStorage(options: { indexedDB?: IDBFactory; locks?:
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');if(depot)await writeDepotCut(transaction,leader.scope,image.rootRevision,depot);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);await promoteDepotFiles(transaction,leader.scope,key,image,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]);
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(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;
@@ -178,21 +179,22 @@ export function createBrowserStorage(options: { indexedDB?: IDBFactory; locks?:
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)=>{
return execute(leader,['files',...depotStores],'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();
};
request.onsuccess=()=>{void(async()=>{const cursor=request.result;if(!cursor){resolve(files);return;}const parts=cursor.key;
if(!Array.isArray(parts)||parts.length!==2||parts[0]!==key||typeof parts[1]!=='string'||!parts[1]){throw fault('corrupt_file_store');}
const bytes=await resolveDepotFile(transaction,leader.scope,parts[1],cursor.value);
if(bytes)files.push(Object.freeze({id:parts[1],byteLength:bytes.size,mimeType:bytes.type}));if(files.length>limit)resolve(files);else cursor.continue();
})().catch(reject);};
});
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]))); },
readFile(leader: StorageLeader, id: string) { return execute(leader, ['files',...depotStores], 'readonly', async(transaction, _metadata, key) => resolveDepotFile(transaction,leader.scope,id,await requestValue<unknown>(transaction.objectStore('files').get([key, id])))); },
close() { closed = true; for (const handle of handles) void handle.release(); void opening.then(database => database.close()).catch(() => {}); },
});
}