Yield recovery container encoding without changing durable cuts
This commit is contained in:
@@ -37,7 +37,7 @@ import { createRunTracker, validateRunInput, runContainsTarget, type RunPort, ty
|
||||
import { createNetworkGate } from './network.ts';
|
||||
import { createLedger, type FrozenRequest } from './ledger.ts';
|
||||
import { operationCanvases, replayPending, validateCanonicalRead } from './reconcile.ts';
|
||||
import { decodeRecovery, encodeRecovery } from './recovery.ts';
|
||||
import { decodeRecovery, encodeRecoveryAsync } from './recovery.ts';
|
||||
import { createScheduler, type Clock, type OperationStatus } from './scheduler.ts';
|
||||
import { planConflictRecovery, type ConflictPlan, type RecoveryChoice } from './conflicts.ts';
|
||||
import { stageAppliedRecovery } from './applied-recovery.ts';
|
||||
@@ -213,7 +213,7 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
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 depot=options.assets?.authority&&options.storage.assetDepot?await prepareDepotCut({snapshot,registry:options.registry,storage:options.storage,leader,authority:options.assets.authority.scope,files:fileWrites,adoptedFiles,displays:displayPins,temporary:admissionPins(snapshot)}):undefined;
|
||||
const receipt = await options.storage.commit(leader, { image: encodeRecovery(snapshot), files,deleteFiles,...(depot?{depot}:{}) });
|
||||
const receipt = await options.storage.commit(leader, { image: await encodeRecoveryAsync(snapshot), files,deleteFiles,...(depot?{depot}:{}) });
|
||||
if (receipt.durableRootRevision !== revision) throw fault('invalid_durable_receipt');
|
||||
confirmedLedgerGeneration=Math.max(confirmedLedgerGeneration,snapshot.runtime.ledger.generation);
|
||||
durableRootRevision = revision;durableImportSeals=importSeals(snapshot.runtime.imports); pending.delete(revision);notifyDurability();
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
import type { Json } from '../canvas-core/types.ts';
|
||||
import { fault, freeze, validJson } from '../canvas-core/values.ts';
|
||||
import {isDataContainer} from '../canvas-core/data-containers.ts';
|
||||
import {createImportBudget, type ImportScheduling} from './import-budget.ts';
|
||||
import type { RecoveryImage } from './storage-port.ts';
|
||||
|
||||
type Encoded = {value:Json|undefined;height:number;reusable:boolean;bytes:number};
|
||||
@@ -8,7 +10,7 @@ function utf8Length(text:string){let bytes=0;for(let i=0;i<text.length;i++){cons
|
||||
const scalarBytes=(value:unknown)=>utf8Length(JSON.stringify(value));
|
||||
|
||||
/** JSON value semantics, with reuse only for transitively frozen plain data properties. */
|
||||
export function encodeRecoveryValue(input:unknown):Json {
|
||||
function createEncoder() {
|
||||
const visiting=new Set<object>();
|
||||
function visit(value:unknown,depth:number,key:string):Encoded {
|
||||
if(value===null||typeof value==='string'||typeof value==='boolean'){if(depth>64)throw fault('invalid_recovery_payload');return{value,height:0,reusable:true,bytes:scalarBytes(value)};}
|
||||
@@ -50,13 +52,81 @@ export function encodeRecoveryValue(input:unknown):Json {
|
||||
const encoded={value:Object.freeze(result) as Json,height,reusable,bytes};encodedSizes.set(result,bytes);if(reusable)immutable.set(object,encoded);return encoded;
|
||||
}finally{visiting.delete(object);}
|
||||
}
|
||||
const result=visit(input,0,'');if(result.value===undefined)throw fault('invalid_recovery_payload');return result.value;
|
||||
return visit;
|
||||
}
|
||||
export function encodeRecoveryValue(input:unknown):Json {
|
||||
const result=createEncoder()(input,0,'');if(result.value===undefined)throw fault('invalid_recovery_payload');return result.value;
|
||||
}
|
||||
|
||||
/** Only codec-owned, validated immutable images qualify for the storage copy fast path. */
|
||||
export function sealRecoveryImage(formatVersion:1|2|3|4|5|6|7|8|9|10|11|12|13|14|15|16|17,rootRevision:number,localRevision:number,payload:unknown):RecoveryImage {
|
||||
type ImageVersion = RecoveryImage['formatVersion'];
|
||||
function validateHeader(formatVersion:ImageVersion,rootRevision:number,localRevision:number){
|
||||
if(![1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17].includes(formatVersion)||!Number.isSafeInteger(rootRevision)||rootRevision<0||!Number.isSafeInteger(localRevision)||localRevision<0||localRevision>rootRevision)throw fault('corrupt_recovery_image');
|
||||
const encoded=encodeRecoveryValue(payload),image=Object.freeze({formatVersion,rootRevision,localRevision,payload:encoded});sealed.add(image);imageSizes.set(image,JSON.stringify({...image,payload:null}).length-4+(encoded&&typeof encoded==='object'?encodedSizes.get(encoded)??scalarBytes(encoded):scalarBytes(encoded)));return image;
|
||||
}
|
||||
function finishImage(formatVersion:ImageVersion,rootRevision:number,localRevision:number,encoded:Json):RecoveryImage {
|
||||
const image=Object.freeze({formatVersion,rootRevision,localRevision,payload:encoded});sealed.add(image);imageSizes.set(image,JSON.stringify({...image,payload:null}).length-4+(encoded&&typeof encoded==='object'?encodedSizes.get(encoded)??scalarBytes(encoded):scalarBytes(encoded)));return image;
|
||||
}
|
||||
/** Only codec-owned, validated immutable images qualify for the storage copy fast path. */
|
||||
export function sealRecoveryImage(formatVersion:ImageVersion,rootRevision:number,localRevision:number,payload:unknown):RecoveryImage {
|
||||
validateHeader(formatVersion,rootRevision,localRevision);
|
||||
return finishImage(formatVersion,rootRevision,localRevision,encodeRecoveryValue(payload));
|
||||
}
|
||||
|
||||
type Prepared = Encoded | {source:object;depth:number;children:Prepared[];keys?:string[]};
|
||||
// Read unknown objects synchronously, before the first suspension. Shape-owned
|
||||
// containers can include external values; each such occurrence is captured now.
|
||||
// Only our fresh container shape and previously proven encoded values defer.
|
||||
function prepareEncoding(input:unknown) {
|
||||
const normalize=createEncoder(),visiting=new Set<object>();
|
||||
function prepare(value:unknown,depth:number,key:string):Prepared {
|
||||
if(!value||typeof value!=='object')return normalize(value,depth,key);
|
||||
const cached=immutable.get(value);
|
||||
if(cached){if(depth+cached.height>64)throw fault('invalid_recovery_payload');return cached;}
|
||||
// Prototype toJSON changes must retain ordinary JSON semantics, including
|
||||
// accessors. Inspect descriptors without evaluating them an extra time.
|
||||
let prototype:object|null=value,hasToJSON=false;
|
||||
while(prototype){if(Object.getOwnPropertyDescriptor(prototype,'toJSON')){hasToJSON=true;break;}prototype=Object.getPrototypeOf(prototype);}
|
||||
if(!isDataContainer(value)||hasToJSON)return normalize(value,depth,key);
|
||||
if(depth>64||visiting.has(value))throw fault('invalid_recovery_payload');
|
||||
visiting.add(value);
|
||||
try {
|
||||
const keys=Array.isArray(value)?undefined:Object.keys(value),children:Prepared[]=[];
|
||||
if(keys)for(const name of keys)children.push(prepare((value as Record<string,unknown>)[name],depth+1,name));
|
||||
else for(let index=0;index<(value as unknown[]).length;index++)children.push(prepare((value as unknown[])[index],depth+1,String(index)));
|
||||
return {source:value,depth,children,keys};
|
||||
}finally{visiting.delete(value);}
|
||||
}
|
||||
return prepare(input,0,'');
|
||||
}
|
||||
|
||||
/** Capture external data immediately; cooperatively encode stable owned tables.
|
||||
* Time slicing is soft: synchronous capture of an unknown subtree remains indivisible.
|
||||
* The returned image has exactly the synchronous codec's wire representation. */
|
||||
export async function sealRecoveryImageAsync(formatVersion:ImageVersion,rootRevision:number,localRevision:number,payload:unknown,options:ImportScheduling={}):Promise<RecoveryImage> {
|
||||
validateHeader(formatVersion,rootRevision,localRevision);
|
||||
const plan=prepareEncoding(payload);
|
||||
const scheduler=(globalThis as typeof globalThis & {scheduler?:{yield?:()=>Promise<void>}}).scheduler;
|
||||
const budget=createImportBudget(new AbortController().signal,{...options,yieldTask:options.yieldTask??(typeof scheduler?.yield==='function'?()=>scheduler.yield!():undefined)});
|
||||
function* encode(plan:Prepared):Generator<Promise<void>,Encoded,void>{
|
||||
if('height' in plan)return plan;
|
||||
let height=0,bytes=2,reusable=true,properties=0;
|
||||
const array=plan.keys===undefined,result:Json[]|Record<string,Json>=array?[]:{};
|
||||
for(let index=0;index<plan.children.length;index++){
|
||||
const entry=plan.children[index],child='height' in entry?entry:yield* encode(entry);
|
||||
reusable&&=child.reusable;
|
||||
if(array){
|
||||
if(child.value===undefined&&plan.depth+1>64)throw fault('invalid_recovery_payload');
|
||||
(result as Json[]).push(child.value??null);bytes+=(index?1:0)+(child.value===undefined?4:child.bytes);height=Math.max(height,1+child.height);
|
||||
}else if(child.value!==undefined){
|
||||
const name=plan.keys![index];bytes+=(properties++?1:0)+scalarBytes(name)+1+child.bytes;
|
||||
Object.defineProperty(result,name,{value:child.value,enumerable:true});height=Math.max(height,1+child.height);
|
||||
}
|
||||
const pause=budget.step();if(pause)yield pause;
|
||||
}
|
||||
const encoded={value:Object.freeze(result) as Json,height,reusable,bytes};encodedSizes.set(result,bytes);if(reusable)immutable.set(plan.source,encoded);return encoded;
|
||||
}
|
||||
const iterator=encode(plan);let step=iterator.next();while(!step.done){await step.value;step=iterator.next();}
|
||||
if(step.value.value===undefined)throw fault('invalid_recovery_payload');
|
||||
return finishImage(formatVersion,rootRevision,localRevision,step.value.value);
|
||||
}
|
||||
export function isSealedRecoveryImage(image:RecoveryImage):boolean{return sealed.has(image);}
|
||||
|
||||
|
||||
@@ -4,7 +4,9 @@ import {validateLoadingRuntime,type LoadingState} from './loading.ts';
|
||||
import {validateOrphanFile,type OrphanFile} from './orphan-files.ts';
|
||||
import {importFiles} from './import-file-retention.ts';
|
||||
import {resourceRetentionPins} from './resource-retention.ts';
|
||||
import { sealRecoveryImage } from './recovery-encoding.ts';
|
||||
import {dataArray,dataRecord} from '../canvas-core/data-containers.ts';
|
||||
import type {ImportScheduling} from './import-budget.ts';
|
||||
import { sealRecoveryImage, sealRecoveryImageAsync } from './recovery-encoding.ts';
|
||||
import { validateCreationSizeEffect } from '../canvas-core/creation-size-plan.ts';
|
||||
import { validateDefaultImageSizing } from './default-image-size.ts';
|
||||
import { validateImportJob, type ImportJob } from './imports.ts';
|
||||
@@ -24,19 +26,19 @@ import type { RecoveryImage } from './storage-port.ts';
|
||||
// Membership is stable only for core-owned read views. Never infer ownership
|
||||
// from Object.freeze: a frozen native Map can still change through .set().
|
||||
const entryViews=new WeakMap<object,readonly (readonly [unknown,unknown])[]>();
|
||||
const noEntries=Object.freeze([]);
|
||||
const noEntries=dataArray([]);
|
||||
function mapValue<K,V>(map?:ReadonlyMap<K,V>):readonly (readonly [K,V])[]{
|
||||
if(!map)return noEntries;
|
||||
if(!isOwnedMapView(map))return [...map];
|
||||
let value=entryViews.get(map);
|
||||
if(!value){value=Object.freeze([...map].map(pair=>Object.freeze(pair)));entryViews.set(map,value);}
|
||||
if(!value){value=dataArray([...map].map(pair=>dataArray(pair) as readonly [K,V]));entryViews.set(map,value);}
|
||||
return value as readonly (readonly [K,V])[];
|
||||
}
|
||||
|
||||
function encodeDocumentValue(document:Document){
|
||||
// Freeze our own containers only. The codec still independently proves whether
|
||||
// each nested value can be reused, including external mutable records.
|
||||
return Object.freeze({projectId:document.projectId,canvases:Object.freeze([...document.canvases.values()]),elements:Object.freeze([...document.elements.values()]),tombstones:Object.freeze([...document.tombstones].map(pair=>Object.freeze(pair)))});
|
||||
return dataRecord({projectId:document.projectId,canvases:dataArray(document.canvases.values()),elements:dataArray(document.elements.values()),tombstones:dataArray([...document.tombstones].map(pair=>dataArray(pair)))});
|
||||
}
|
||||
const documentViews=new WeakMap<Document,ReturnType<typeof encodeDocumentValue>>();
|
||||
function documentValue(document:Document){
|
||||
@@ -44,14 +46,20 @@ function documentValue(document:Document){
|
||||
let value=documentViews.get(document);if(!value){value=encodeDocumentValue(document);documentViews.set(document,value);}return value;
|
||||
}
|
||||
/** Plain records only: functions, credentials, renderer objects, URLs and Map methods never serialize. */
|
||||
export function encodeRecovery(snapshot: EngineSnapshot): RecoveryImage {
|
||||
const payload = {
|
||||
function recoveryInput(snapshot: EngineSnapshot) {
|
||||
const payload = dataRecord({
|
||||
document: documentValue(snapshot.document),
|
||||
history: { undo: snapshot.history.undo, redo: snapshot.history.redo, entries: mapValue(snapshot.history.entries) },
|
||||
history: dataRecord({ undo: snapshot.history.undo, redo: snapshot.history.redo, entries: mapValue(snapshot.history.entries) }),
|
||||
operations: mapValue(snapshot.operations),
|
||||
runtime: { ...(snapshot.runtime.assetBindings?.size?{assetBindings:mapValue(snapshot.runtime.assetBindings)}:{}), ...(snapshot.runtime.loadingRead?{loadingRead:{document:documentValue(snapshot.runtime.loadingRead.document),versions:mapValue(snapshot.runtime.loadingRead.versions)}}:{}), ...(snapshot.runtime.loading?{loading:snapshot.runtime.loading}:{}), ...(snapshot.runtime.orphanFiles?.size?{orphanFiles:mapValue(snapshot.runtime.orphanFiles)}:{}), imports:mapValue(snapshot.runtime.imports), recoveries:mapValue(snapshot.runtime.recoveries), runs: mapValue(snapshot.runtime.runs), resources: mapValue(snapshot.runtime.resources), resourcePins: mapValue(snapshot.runtime.resourcePins), acknowledged: documentValue(snapshot.runtime.acknowledged), versions: mapValue(snapshot.runtime.versions), receipts: mapValue(snapshot.runtime.receipts), conflicts: mapValue(snapshot.runtime.conflicts), ledger: { ...snapshot.runtime.ledger, operations: mapValue(snapshot.runtime.ledger.operations), requests: mapValue(snapshot.runtime.ledger.requests) } },
|
||||
};
|
||||
return sealRecoveryImage([...(snapshot.runtime.imports?.values()??[])].some(job=>job.source?.integrityVersion===2)?17:[...(snapshot.runtime.imports?.values()??[])].some(job=>job.source?.kind==='native-document'&&job.parserVersion===5)?16:[...(snapshot.runtime.imports?.values()??[])].some(job=>job.source?.kind==='native-document')?15:[...(snapshot.runtime.imports?.values()??[])].some(job=>job.source?.kind==='native-json-array')?14:snapshot.runtime.assetBindings?.size?13:snapshot.runtime.loading?(snapshot.runtime.loading.version===6?12:snapshot.runtime.loading.version===5?11:snapshot.runtime.loading.version===4?10:snapshot.runtime.loading.version===3?9:snapshot.runtime.loading.version===2?8:7):[...snapshot.operations.values()].some(intent=>intent.historyEffect)||[...snapshot.runtime.resources.values()].some(resource=>resource.sizing)?6:snapshot.runtime.orphanFiles?.size?5:[...(snapshot.runtime.imports?.values()??[])].some(job=>job.source?.retirement||job.source?.previousFiles?.length)?4:[...snapshot.runtime.resources.values()].some(resource=>resource.retirement)?3:snapshot.runtime.imports?.size ? 2 : 1,snapshot.rootRevision,snapshot.revision,payload);
|
||||
runtime: dataRecord({ ...(snapshot.runtime.assetBindings?.size?{assetBindings:mapValue(snapshot.runtime.assetBindings)}:{}), ...(snapshot.runtime.loadingRead?{loadingRead:dataRecord({document:documentValue(snapshot.runtime.loadingRead.document),versions:mapValue(snapshot.runtime.loadingRead.versions)})}:{}), ...(snapshot.runtime.loading?{loading:snapshot.runtime.loading}:{}), ...(snapshot.runtime.orphanFiles?.size?{orphanFiles:mapValue(snapshot.runtime.orphanFiles)}:{}), imports:mapValue(snapshot.runtime.imports), recoveries:mapValue(snapshot.runtime.recoveries), runs: mapValue(snapshot.runtime.runs), resources: mapValue(snapshot.runtime.resources), resourcePins: mapValue(snapshot.runtime.resourcePins), acknowledged: documentValue(snapshot.runtime.acknowledged), versions: mapValue(snapshot.runtime.versions), receipts: mapValue(snapshot.runtime.receipts), conflicts: mapValue(snapshot.runtime.conflicts), ledger: dataRecord({ ...snapshot.runtime.ledger, operations: mapValue(snapshot.runtime.ledger.operations), requests: mapValue(snapshot.runtime.ledger.requests) }) }),
|
||||
});
|
||||
return [[...(snapshot.runtime.imports?.values()??[])].some(job=>job.source?.integrityVersion===2)?17:[...(snapshot.runtime.imports?.values()??[])].some(job=>job.source?.kind==='native-document'&&job.parserVersion===5)?16:[...(snapshot.runtime.imports?.values()??[])].some(job=>job.source?.kind==='native-document')?15:[...(snapshot.runtime.imports?.values()??[])].some(job=>job.source?.kind==='native-json-array')?14:snapshot.runtime.assetBindings?.size?13:snapshot.runtime.loading?(snapshot.runtime.loading.version===6?12:snapshot.runtime.loading.version===5?11:snapshot.runtime.loading.version===4?10:snapshot.runtime.loading.version===3?9:snapshot.runtime.loading.version===2?8:7):[...snapshot.operations.values()].some(intent=>intent.historyEffect)||[...snapshot.runtime.resources.values()].some(resource=>resource.sizing)?6:snapshot.runtime.orphanFiles?.size?5:[...(snapshot.runtime.imports?.values()??[])].some(job=>job.source?.retirement||job.source?.previousFiles?.length)?4:[...snapshot.runtime.resources.values()].some(resource=>resource.retirement)?3:snapshot.runtime.imports?.size ? 2 : 1,snapshot.rootRevision,snapshot.revision,payload] as const;
|
||||
}
|
||||
export function encodeRecovery(snapshot:EngineSnapshot):RecoveryImage {
|
||||
return sealRecoveryImage(...recoveryInput(snapshot));
|
||||
}
|
||||
export function encodeRecoveryAsync(snapshot:EngineSnapshot,options:ImportScheduling={}):Promise<RecoveryImage> {
|
||||
return sealRecoveryImageAsync(...recoveryInput(snapshot),options);
|
||||
}
|
||||
type ObjectValue = Record<string, unknown>;
|
||||
function object(value: unknown): ObjectValue { if (!value || Array.isArray(value) || typeof value !== 'object') throw fault('corrupt_recovery_image'); return value as ObjectValue; }
|
||||
|
||||
Reference in New Issue
Block a user