57 lines
5.3 KiB
TypeScript
57 lines
5.3 KiB
TypeScript
import type { Id } from '../canvas-core/types.ts';
|
|
import { equal, fault, freeze, readMap } from '../canvas-core/values.ts';
|
|
import type { AppliedModelProof, CanonicalProposal, EngineSnapshot, PersistencePort } from './contracts.ts';
|
|
import { createLedger, type FrozenRequest } from './ledger.ts';
|
|
import { runTerminal, type RunPort } from './runs.ts';
|
|
import type { AssetPort } from './resources.ts';
|
|
|
|
/** Stage acknowledgements only. The caller must validate a complete fresh read and durably publish it. */
|
|
export function stageAppliedRecovery(snapshot:EngineSnapshot,ports:{persistence:PersistencePort;runs?:RunPort;assets?:AssetPort;now:number}) {
|
|
const waiting=[...snapshot.runtime.ledger.requests.values()].filter(request=>request.phase!=='resolved');
|
|
const unknown=waiting.filter(request=>request.outcome!=='applied');
|
|
if(unknown.length)throw fault('recovery_waiting_for_receipts',unknown.map(request=>request.id));
|
|
const sequence=(request:FrozenRequest)=>request.operationIds.reduce((result,id)=>Math.min(result,snapshot.runtime.ledger.operations.get(id)?.intent.sequence??Infinity),Infinity);
|
|
waiting.sort((a,b)=>sequence(a)-sequence(b));
|
|
const ledger=createLedger(snapshot.runtime.ledger),runs=new Map(snapshot.runtime.runs),resources=new Map(snapshot.runtime.resources),conflicts=new Map(snapshot.runtime.conflicts);
|
|
const proofs:AppliedModelProof[]=[],runIds=new Set<Id>();
|
|
for(const request of waiting){
|
|
const receipt=snapshot.runtime.receipts.get(request.id);if(receipt===undefined)throw fault('recovery_receipt_missing',[request.id]);
|
|
let proof:AppliedModelProof|undefined;
|
|
if(request.kind==='asset.upload'){
|
|
const intent=snapshot.runtime.ledger.operations.get(request.operationIds[0])?.intent,resource=intent?.operationType==='asset.upload'?resources.get(intent.resourceId):undefined;
|
|
if(!ports.assets||!resource)throw fault('asset_interpreter_unavailable',[request.id]);
|
|
const stored=freeze(ports.assets.accept(request,{outcome:'applied',canonical:receipt},resource));
|
|
if(stored.id!==resource.id||stored.bytes!==resource.byteLength)throw fault('asset_receipt_mismatch');
|
|
resources.set(resource.id,freeze({...resource,stored}));
|
|
}else if(['run.start','run.cancel','run.apply'].includes(request.kind??'')){
|
|
if(!ports.runs?.verifyApplied)throw fault('run_receipt_interpreter_unavailable',[request.id]);
|
|
const verified=ports.runs.verifyApplied(request,receipt);proof=verified.model;
|
|
if(request.kind==='run.start'){
|
|
const run=verified.admission,intent=snapshot.runtime.ledger.operations.get(request.operationIds[0])?.intent;
|
|
if(!run||verified.action||intent?.operationType!=='run.start'||run.operationId!==intent.id||run.requestId!==request.id||run.boardId!==intent.run.boardId||!equal(run.target,intent.run.target))throw fault('run_identity_mismatch');
|
|
const existing=runs.get(run.id);
|
|
if(existing&&(existing.requestId!==run.requestId||existing.operationId!==run.operationId||!equal(existing.plan,run.plan)))throw fault('run_identity_mismatch');
|
|
if(!existing)runs.set(run.id,freeze(run));runIds.add(run.id);
|
|
}else{
|
|
const action=verified.action,run=action?runs.get(action.runId):undefined;
|
|
if(!action||verified.admission||action.runId!==request.runId||!run)throw fault('run_identity_mismatch');
|
|
if(request.kind==='run.cancel'){
|
|
if(!action.runStatus||action.outputId||runTerminal(run.status)&&runTerminal(action.runStatus)&&run.status!==action.runStatus)throw fault('run_status_conflict');
|
|
}else if(!action.outputId||action.runStatus)throw fault('run_identity_mismatch');
|
|
runs.set(run.id,freeze({...run,controlEpoch:(run.controlEpoch??0)+1,...(action.runStatus?{status:action.runStatus}:{}),...(action.outputId?{appliedOutputIds:[...new Set([...(run.appliedOutputIds??[]),action.outputId])]}:{}),hasMore:true,blocked:false,error:undefined,pendingPage:undefined,nextPollAt:ports.now}));runIds.add(run.id);
|
|
}
|
|
}else{
|
|
if(!ports.persistence.verifyApplied)throw fault('document_receipt_interpreter_unavailable',[request.id]);
|
|
proof=ports.persistence.verifyApplied(request,receipt);
|
|
}
|
|
const writes=request.kind==='run.start'?[request.ownerCanvasId!]:request.kind==='run.cancel'||request.kind==='asset.upload'?[]:request.canvasIds;
|
|
if(writes.length!==(proof?.versions.size??0)||writes.some(id=>!proof?.versions.has(id)))throw fault('receipt_coverage_mismatch',[request.id]);
|
|
if(proof){for(const [canvas,version]of proof.versions)if(!Number.isSafeInteger(version)||version<=request.expectedVersions[canvas])throw fault('receipt_version_mismatch',[canvas]);proofs.push(proof);}
|
|
ledger.result(request.id,'applied');conflicts.delete(request.id);
|
|
}
|
|
const staged:EngineSnapshot=Object.freeze({...snapshot,runtime:Object.freeze({...snapshot.runtime,ledger:ledger.getSnapshot(),runs:readMap(runs),resources:readMap(resources),conflicts:readMap(conflicts)})});
|
|
return Object.freeze({snapshot:staged,requestIds:Object.freeze(waiting.map(request=>request.id)),runIds:Object.freeze([...runIds]),
|
|
assertCanonical(canonical:CanonicalProposal){for(const proof of proofs){for(const [canvas,version]of proof.versions)if((canonical.versions.get(canvas)??-1)<version)throw fault('recovery_read_behind_receipt',[canvas]);proof.assertCanonical(canonical);}},
|
|
});
|
|
}
|