Compact reconciled document receipts through an explicit retention policy
This commit is contained in:
@@ -874,3 +874,21 @@ ordered comparison and no-op rules.
|
||||
Synchronous document validation loops directly over the same per-record checks
|
||||
used by the cooperative iterator. The iterator retains its original yield points;
|
||||
the atomic publication path does not pay a generator handoff per unchanged record.
|
||||
|
||||
Resolved Foundry document receipts have an explicit retention policy. The SDK
|
||||
keeps raw receipts by default; CanvasRuntimeHost opts into `summary`. Only a
|
||||
successfully reconciled protocol-2 document change with known model/record fields
|
||||
can become the adapter's `foundry.reconciled-document` version-1 summary. It keeps
|
||||
the request ID, covered logical operation IDs, project ID and actual returned
|
||||
canvas versions. Frozen request bytes, ledger outcome, ACK base and history remain
|
||||
in their existing owners. This summary does not retain the complete raw response
|
||||
for audit, and cannot be used as a fresh server proof or replay input.
|
||||
|
||||
The engine prepares the summary for the same Root publication that resolves the
|
||||
ledger and installs the ACK/document/history result. If that publication rejects,
|
||||
the original raw receipt remains available with the unresolved request. A failed
|
||||
or invalid summary codec reports the error and retains raw data without failing a
|
||||
valid ACK. Unknown fields/components/versions, unresolved requests and run/asset
|
||||
receipts stay raw. Existing stored raw receipts are not migrated by this policy.
|
||||
The recovery envelope remains unchanged: receipts are opaque JSON, and the
|
||||
adapter-local summary has its own version and explicit replay rejection.
|
||||
|
||||
@@ -22,7 +22,7 @@ export function CanvasRuntimeHost({ children,extensions=[] }: { children: ReactN
|
||||
return createWorkspaceRuntime({ accountId, workspaceId, authEpoch, isCurrentIdentity, registry, storage: createBrowserStorage(),digests:createBrowserDigests(),images:createBrowserImageDecoder(),
|
||||
adapters(rootId, current) {
|
||||
const transport = createFoundryTransport({ accountId, isCurrentIdentity: current });
|
||||
return { runs: createFoundryRuns({ workspaceId, rootId, registry, transport }), persistence: createFoundryPersistence({ workspaceId, rootId, registry, transport }), assets: createFoundryAssets({ workspaceId, transport, project: process.env.NEXT_PUBLIC_FOUNDRY_PROJECT || "endoleap", origin: process.env.NEXT_PUBLIC_FOUNDRY_API_ORIGIN }) };
|
||||
return { runs: createFoundryRuns({ workspaceId, rootId, registry, transport }), persistence: createFoundryPersistence({ workspaceId, rootId, registry, transport, resolvedReceiptPolicy:'summary' }), assets: createFoundryAssets({ workspaceId, transport, project: process.env.NEXT_PUBLIC_FOUNDRY_PROJECT || "endoleap", origin: process.env.NEXT_PUBLIC_FOUNDRY_API_ORIGIN }) };
|
||||
},
|
||||
onError: error => notify({ id: `canvas-runtime:${accountId}:${workspaceId}`, title: "画布同步需要处理", description: errorText(error), level: "error" }),
|
||||
});
|
||||
|
||||
@@ -17,6 +17,7 @@ export type RuntimeState = {
|
||||
readonly runs: ReadonlyMap<Id, RunRecord>;
|
||||
readonly acknowledged: Document;
|
||||
readonly versions: ReadonlyMap<Id, number>;
|
||||
/** Raw recovery inputs, or adapter-owned summaries for fully reconciled document requests. */
|
||||
readonly receipts: ReadonlyMap<Id, Json>;
|
||||
readonly conflicts: ReadonlyMap<Id, string>;
|
||||
readonly resources: ReadonlyMap<Id,ResourceRecord>;
|
||||
@@ -40,6 +41,8 @@ export type PersistencePort = {
|
||||
compileBatch?: (intents: readonly OperationIntent[], requestId: Id, snapshot: EngineSnapshot) => CompiledRequest;
|
||||
submit(request: FrozenRequest, signal: AbortSignal): Promise<OperationStatus>;
|
||||
query(request: FrozenRequest, signal: AbortSignal): Promise<OperationStatus>;
|
||||
/** Optional pure codec for a joint resolved publication. Rejection/failure retains the raw recovery input. */
|
||||
compactResolvedReceipt?: (request:FrozenRequest,receipt:Json,applied:CanonicalProposal)=>Json;
|
||||
/** Pure codec, merging only participants into the latest base, never an old working snapshot. */
|
||||
canonical(request: FrozenRequest, status: OperationStatus, snapshot: EngineSnapshot): CanonicalProposal;
|
||||
};
|
||||
|
||||
@@ -31,7 +31,7 @@ import type { Document, Id, Registry, Result, RunInput, RunApplyInput, RunCancel
|
||||
import { createEditor, type EditorOptions, type Transaction } from '../canvas-core/editor.ts';
|
||||
import { createDocument } from '../canvas-core/document.ts';
|
||||
import { createSessions } from '../canvas-core/gestures.ts';
|
||||
import { clone, equal, fault, freeze, readMap, reject } from '../canvas-core/values.ts';
|
||||
import { clone, equal, fault, freeze, readMap, reject, validJson } from '../canvas-core/values.ts';
|
||||
import type { CanonicalProposal, EngineSnapshot, PersistencePort, RuntimeState } from './contracts.ts';
|
||||
import { createRunTracker, validateRunInput, runContainsTarget, type RunPort, type RunRecord, type RunPage } from './runs.ts';
|
||||
import { createNetworkGate } from './network.ts';
|
||||
@@ -448,7 +448,16 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
if (!existing) runs.set(admission.run.id, admission.run);
|
||||
}
|
||||
if(action){const run=runs.get(action.runId);if(!run)throw fault('run_identity_mismatch');runs.set(run.id,freeze({...run,controlEpoch:(run.controlEpoch??0)+1,appliedOutputIds:[...new Set([...(run.appliedOutputIds??[]),action.outputId!])],hasMore:true,blocked:false,error:undefined,pendingPage:undefined,nextPollAt:runClock.now()}));}
|
||||
const runtime: RuntimeState = Object.freeze({ ...previous, runs: readMap(runs), ledger: ledger.getSnapshot(), acknowledged, versions: readMap(proposal.versions), receipts: readMap(receipts), conflicts: readMap(conflicts) });
|
||||
const appliedVersions=readMap(proposal.versions);let resolvedReceipts=receipts;
|
||||
if(request.kind==='document.change'&&status.canonical!==undefined&&options.persistence.compactResolvedReceipt){
|
||||
try{
|
||||
const compact=options.persistence.compactResolvedReceipt(ledger.getSnapshot().requests.get(request.id)!,status.canonical,Object.freeze({document:acknowledged,versions:appliedVersions}));
|
||||
if(!validJson(compact))throw fault('invalid_receipt_compaction');
|
||||
if(compact!==status.canonical){const owned=freeze(clone(compact));if(!validJson(owned))throw fault('invalid_receipt_compaction');resolvedReceipts=new Map(receipts);resolvedReceipts.set(request.id,owned);}
|
||||
}catch(error){report(error);}
|
||||
}
|
||||
// Keep the original map intact for a rejected final publication below.
|
||||
const runtime: RuntimeState = Object.freeze({ ...previous, runs: readMap(runs), ledger: ledger.getSnapshot(), acknowledged, versions: appliedVersions, receipts: readMap(resolvedReceipts), conflicts: readMap(conflicts) });
|
||||
const oldAccepting=acceptingLoading;acceptingLoading=oldAccepting||(previous.loading?.version??0)>=3;let result:Result;try{result=editor.reconcile(document, runtime, root.rootRevision);}finally{acceptingLoading=oldAccepting;}
|
||||
if (result.status === 'rejected') throw fault(result.code, result.targets);
|
||||
if(action)queueMicrotask(()=>runTracker.notify([action.runId]));
|
||||
|
||||
@@ -125,8 +125,16 @@ function validateFrozen(request: FrozenRequest) {
|
||||
}
|
||||
|
||||
/** Operations/conditional fields are sent; full snapshots are only reads and ACK participant views. */
|
||||
export function createFoundryPersistence(options: { workspaceId: Id; rootId: Id; registry: Registry; transport: ServiceTransport }): PersistencePort {
|
||||
export type FoundryResolvedReceiptSummary={readonly localReceiptFormat:'foundry.reconciled-document';readonly localReceiptVersion:1;readonly protocol_version:2;readonly request_id:Id;readonly status:'applied';readonly operation_ids:readonly Id[];readonly projectId:Id;readonly versions:Readonly<Record<Id,number>>};
|
||||
const receiptKeys=new Set(['protocol_version','request_id','status','operation_ids','model']);
|
||||
const modelKeys=new Set(['documentFormatVersion','projectId','complete','canvases','elements','tombstones','versions','removedCanvasIds']);
|
||||
const elementKeys=new Set(['id','incarnation','ownerCanvasId','componentId','dataVersion','structure','geometry','data','source','target','path']);
|
||||
function hasExtraKeys(value:Record<string,unknown>,known:ReadonlySet<string>){for(const key in value)if(Object.hasOwn(value,key)&&!known.has(key))return true;return false;}
|
||||
function assertRawReceipt(value:unknown){if(value&&typeof value==='object'&&!Array.isArray(value)&&(value as Record<string,unknown>).localReceiptFormat==='foundry.reconciled-document'&&!Object.hasOwn(value,'model')){if((value as Record<string,unknown>).localReceiptVersion!==1)throw fault('receipt_summary_interpreter_unavailable');throw fault('receipt_already_reconciled');}}
|
||||
|
||||
export function createFoundryPersistence(options: { workspaceId: Id; rootId: Id; registry: Registry; transport: ServiceTransport;resolvedReceiptPolicy?:'raw'|'summary' }): PersistencePort {
|
||||
if (!uuid.test(options.workspaceId) || !uuid.test(options.rootId)) throw fault('invalid_service_scope');
|
||||
if(options.resolvedReceiptPolicy!==undefined&&!['raw','summary'].includes(options.resolvedReceiptPolicy))throw fault('invalid_receipt_policy');
|
||||
const path = pathFor(options.workspaceId, options.rootId);
|
||||
async function execute(request: FrozenRequest, query: boolean, signal: AbortSignal): Promise<OperationStatus> {
|
||||
try {
|
||||
@@ -155,7 +163,25 @@ export function createFoundryPersistence(options: { workspaceId: Id; rootId: Id;
|
||||
}
|
||||
return Object.freeze({
|
||||
supportsRequest: (request: FrozenRequest) => request.protocolVersion === 2,
|
||||
...(options.resolvedReceiptPolicy==='summary'?{compactResolvedReceipt(request:FrozenRequest,receipt:Json,applied:import('../canvas-runtime/contracts.ts').CanonicalProposal):Json{
|
||||
if(request.kind!=='document.change'||request.protocolVersion!==2||request.phase!=='resolved'||request.outcome!=='applied')return receipt;
|
||||
validateFrozen(request);const body=record(receipt);
|
||||
if(hasExtraKeys(body,receiptKeys))return receipt;
|
||||
if(body.protocol_version!==2)return receipt;
|
||||
if(body.request_id!==request.id||body.status!=='applied')throw fault('receipt_identity_mismatch');
|
||||
const covered=list(body.operation_ids).map(id);if(covered.length!==request.operationIds.length||new Set(covered).size!==covered.length||covered.some(id=>!request.operationIds.includes(id)))throw fault('receipt_coverage_mismatch');
|
||||
const model=record(body.model);
|
||||
if(hasExtraKeys(model,modelKeys)||model.documentFormatVersion!==1||model.projectId!==options.rootId||model.complete!==false)return receipt;
|
||||
const resultVersions=versions(model.versions);
|
||||
if(resultVersions.size!==request.canvasIds.length||request.canvasIds.some(id=>!resultVersions.has(id)||(resultVersions.get(id)??0)<=request.expectedVersions[id]||resultVersions.get(id)!==applied.versions.get(id)))return receipt;
|
||||
for(const value of list(model.elements)){const element=record(value);if(options.registry.resolve(element as unknown as ElementRecord)!=='known'||hasExtraKeys(element,elementKeys))return receipt;}
|
||||
const canvasKeys=new Set(['id','parent']),parentKeys=new Set(['canvasId','entryElementId']);
|
||||
for(const value of list(model.canvases)){const canvas=record(value);if(hasExtraKeys(canvas,canvasKeys)||canvas.parent!==null&&hasExtraKeys(record(canvas.parent),parentKeys))return receipt;}
|
||||
return freeze({localReceiptFormat:'foundry.reconciled-document',localReceiptVersion:1,protocol_version:2,request_id:request.id,status:'applied',operation_ids:covered,projectId:options.rootId,versions:Object.fromEntries(resultVersions)});
|
||||
}}:{}),
|
||||
|
||||
verifyApplied(request:FrozenRequest,receipt:Json){
|
||||
assertRawReceipt(receipt);
|
||||
validateFrozen(request);
|
||||
if(statusFor(request,{status:200,body:receipt},false).outcome!=='applied')throw fault('receipt_identity_mismatch');
|
||||
const proof=verifyFoundryModelProof(record(receipt).model,request.canvasIds,options.rootId);
|
||||
@@ -173,7 +199,7 @@ export function createFoundryPersistence(options: { workspaceId: Id; rootId: Id;
|
||||
submit: (request: FrozenRequest, signal: AbortSignal) => execute(request, false, signal),
|
||||
query: (request: FrozenRequest, signal: AbortSignal) => execute(request, true, signal),
|
||||
canonical(request: FrozenRequest, status: OperationStatus, snapshot: EngineSnapshot) {
|
||||
const body = record(status.canonical);
|
||||
assertRawReceipt(status.canonical);const body = record(status.canonical);
|
||||
if (body.request_id !== request.id || body.protocol_version !== 2) throw fault('receipt_identity_mismatch');
|
||||
return mergeFoundryModel(body.model, request.canvasIds, snapshot, options.registry, options.rootId);
|
||||
},
|
||||
|
||||
@@ -334,3 +334,17 @@ test('a durable dispatch prefix can send while a later local Root is still waiti
|
||||
await engine.persist();storage.commit=async(...args)=>{const image=args[1].image;if(bodyCut===undefined&&image.payload.runtime.ledger.requests.some(([,row])=>row.phase==='dispatch-marked')){bodyCut=image.rootRevision;bodyEntered.resolve();await body.promise;}if(image.rootRevision>after){laterEntered.resolve();await later.promise;}return commit(...args);};engine.editor.dispatch({type:'element.data',target,patch:{title:'first'}});const saving=engine.persist().then(()=>{flushed=true;});await bodyEntered.promise;after=engine.getSnapshot().rootRevision;engine.editor.dispatch({type:'element.geometry',target,patch:{x:42}});body.resolve();await laterEntered.promise;await new Promise((resolve,reject)=>{const timer=setTimeout(()=>reject(Error('durable request waited for a later Root')),2000);sent.promise.then(()=>{clearTimeout(timer);resolve();},reject);});assert.equal(flushed,false);assert.equal(engine.getSnapshot().document.elements.get('a').geometry.x,42);later.resolve();await saving;assert.equal(engine.getDurability().durableRootRevision,engine.getSnapshot().rootRevision);
|
||||
}finally{body.resolve();later.resolve();network.resolve();storage.commit=commit;await engine.close();}
|
||||
});
|
||||
|
||||
test('an opted-in resolved receipt summary publishes with ACK/history/ledger and reopens without retransmission',async()=>{
|
||||
const storage=storageFixture(),port=persistenceFixture({compactResolvedReceipt(request,receipt,applied){assert.equal(request.phase,'resolved');assert.equal(request.outcome,'applied');assert.equal(receipt.receipt,request.id);assert.equal(applied.document.elements.get('a').data.title,'summary');return{localReceiptVersion:1,id:request.id,versions:[...applied.versions]};}});let{engine}=await open(storage,port);
|
||||
try{engine.editor.dispatch({type:'element.data',target,patch:{title:'summary'}});await engine.settle();const snapshot=engine.getSnapshot(),request=[...snapshot.runtime.ledger.requests.values()][0],body=request.body;assert.equal(request.phase,'resolved');assert.deepEqual(snapshot.runtime.receipts.get(request.id),{localReceiptVersion:1,id:request.id,versions:[['canvas',2]]});assert.equal(snapshot.history.undo.length,1);for(const image of storage.saved){const restored=decodeRecovery(image,registry),saved=restored.runtime.ledger.requests.get(request.id);if(restored.runtime.receipts.get(request.id)?.localReceiptVersion===1){assert.equal(saved.phase,'resolved');assert.equal(restored.document.elements.get('a').data.title,'summary');assert.equal(restored.history.undo.length,1);}if(saved)assert.equal(saved.body,body);}await engine.close();({engine}=await open(storage,port));await engine.settle();assert.equal(engine.getSnapshot().runtime.ledger.requests.get(request.id).body,body);assert.equal(engine.getSnapshot().document.elements.get('a').data.title,'summary');assert.equal(port.calls.filter(([kind])=>kind==='submit').length,1);
|
||||
}finally{await engine.close();}
|
||||
});
|
||||
test('a failed final publication keeps the raw receipt and waiting claim instead of the prepared summary',async()=>{
|
||||
let blocked=false,compactions=0;const custom=createRegistry(builtinComponents().map(definition=>definition.id==='text'?{...definition,validate:data=>typeof data.title==='string'&&typeof data.text==='string'&&(!blocked||data.title!=='local')}:definition)),storage=storageFixture(),port=persistenceFixture({compactResolvedReceipt(request){compactions++;if(compactions===1)blocked=true;return{summary:request.id};}});let{engine}=await open(storage,port,{registry:custom});
|
||||
try{engine.editor.dispatch({type:'element.data',target,patch:{title:'local'}});await engine.settle();const snapshot=engine.getSnapshot(),request=[...snapshot.runtime.ledger.requests.values()][0];assert.equal(compactions,1);assert.equal(request.outcome,'applied');assert.equal(request.phase,'waiting');assert.deepEqual(snapshot.runtime.receipts.get(request.id),{receipt:request.id});assert.equal(snapshot.document.elements.get('a').data.title,'local');assert.equal(snapshot.history.undo.length,1);blocked=false;await engine.close();({engine}=await open(storage,port,{registry:custom}));await engine.retry();await engine.settle();assert.equal(engine.getSnapshot().runtime.ledger.requests.get(request.id).phase,'resolved');assert.deepEqual(engine.getSnapshot().runtime.receipts.get(request.id),{summary:request.id});assert.equal(port.calls.filter(([kind])=>kind==='submit').length,1);assert.ok(port.calls.some(([kind])=>kind==='query'));
|
||||
}finally{blocked=false;await engine.close();}
|
||||
});
|
||||
test('receipt compaction is optional and invalid codecs cannot block a valid ACK or discard its raw payload',async()=>{
|
||||
for(const compactResolvedReceipt of [undefined,()=>{throw Error('compact failed');},()=>undefined]){const storage=storageFixture(),port=persistenceFixture(compactResolvedReceipt?{compactResolvedReceipt}:{}),{engine,errors}=await open(storage,port);try{engine.editor.dispatch({type:'element.data',target,patch:{title:'keep raw'}});await engine.settle();const request=[...engine.getSnapshot().runtime.ledger.requests.values()][0];assert.equal(request.phase,'resolved');assert.deepEqual(engine.getSnapshot().runtime.receipts.get(request.id),{receipt:request.id});assert.equal(engine.getSnapshot().runtime.conflicts.size,0);assert.equal(errors.length,compactResolvedReceipt?1:0);}finally{await engine.close();}}
|
||||
});
|
||||
|
||||
@@ -152,3 +152,10 @@ test('Foundry compiles the proven projection without changing logical intents or
|
||||
assert.equal(ledger.getSnapshot().operations.get(first.id).intent.changes[0].after,middle);
|
||||
assert.deepEqual(wire.map(op=>op.id),request.operationIds);assert.equal((await port.submit(request,new AbortController().signal)).outcome,'applied');
|
||||
});
|
||||
|
||||
test('Foundry receipt summaries are opt-in, retain execution metadata and cannot masquerade as raw recovery proofs',()=>{
|
||||
const{request,port:rawPort}=setup();assert.equal(rawPort.compactResolvedReceipt,undefined);const port=createFoundryPersistence({workspaceId:workspace,rootId:root,registry,transport:async()=>{},resolvedReceiptPolicy:'summary'}),receipt={protocol_version:2,request_id:requestId,status:'applied',operation_ids:[operationId],model:chunk(false)},applied=port.canonical(request,{outcome:'applied',canonical:receipt},snapshot),resolved={...request,phase:'resolved',outcome:'applied'},summary=port.compactResolvedReceipt(resolved,receipt,applied);assert.strictEqual(port.compactResolvedReceipt(request,receipt,applied),receipt);assert.deepEqual(summary,{localReceiptFormat:'foundry.reconciled-document',localReceiptVersion:1,protocol_version:2,request_id:requestId,status:'applied',operation_ids:[operationId],projectId:root,versions:{[root]:4}});assert.equal(Object.hasOwn(summary,'model'),false);assert.equal(receipt.model.elements.length,2);assert.throws(()=>port.verifyApplied(request,summary),/receipt_already_reconciled/);assert.throws(()=>port.canonical(request,{outcome:'applied',canonical:summary},snapshot),/receipt_already_reconciled/);assert.throws(()=>port.verifyApplied(request,{...summary,localReceiptVersion:2}),/receipt_summary_interpreter_unavailable/);assert.deepEqual([...port.verifyApplied(request,receipt).versions],[[root,4]]);
|
||||
});
|
||||
test('summary policy preserves opaque, future and unfamiliar receipt data unchanged',()=>{
|
||||
const{request}=setup(),port=createFoundryPersistence({workspaceId:workspace,rootId:root,registry,transport:async()=>{},resolvedReceiptPolicy:'summary'}),raw={protocol_version:2,request_id:requestId,status:'applied',operation_ids:[operationId],model:chunk(false)},applied=port.canonical(request,{outcome:'applied',canonical:raw},snapshot),cases=[{...raw,extension:{keep:true}},{...raw,protocol_version:3},{...raw,model:{...raw.model,documentFormatVersion:2}},{...raw,model:{...raw.model,futureField:'keep'}},{...raw,model:{...raw.model,elements:[...raw.model.elements,{...future,id:'opaque',ownerCanvasId:root}]}},{...raw,model:{...raw.model,elements:[{...a,extension:{keep:true}},f]}}];for(const value of cases)assert.strictEqual(port.compactResolvedReceipt({...request,phase:'resolved',outcome:'applied'},value,applied),value);assert.strictEqual(port.compactResolvedReceipt({...request,kind:'run.apply'},raw,applied),raw);assert.throws(()=>createFoundryPersistence({workspaceId:workspace,rootId:root,registry,transport:async()=>{},resolvedReceiptPolicy:'invalid'}),/invalid_receipt_policy/);
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user