diff --git a/headless/README.md b/headless/README.md index 8052b3b..aee6ff4 100644 --- a/headless/README.md +++ b/headless/README.md @@ -277,7 +277,8 @@ Import from `/react` for React only, or the re-export in `/react-flow`. its own lease. Controller replacements keep independent token slots. Preview geometry remains shared across views; active ownership and controls belong to the owning view. Host bindings still handle pointer capture and events. -- `useOperation(id)` observes both the logical record and its frozen request. +- `useOperation(id)` observes both the logical record and its frozen request; + request-only changes are included without observing unrelated Root fields. - `useAsset(id)` resolves a releasable display handle, returns metadata plus loading/ready/error/url/retry, and releases on ref/owner change or unmount. Ready means the display handle resolved; it does not prove upload ACK, image @@ -542,3 +543,23 @@ subscription removes its bucket and releases the upstream listener. A controller identity change (engine/view/target/incarnation) creates a private lease slot, so old cleanup cannot consume the replacement controller's token. This does not change the one shared gesture owner, its history unit or durable task lifecycle. + + +Operation, asset metadata and sync-count channels share one upstream editor +listener per engine. Their dependency tuples contain only the tables they read +(and the loading flag for counts). Table identities use weak object IDs, so a +channel caches a primitive version signature and selected values rather than an +old Root or full table. Changed tables recheck subscribed IDs; unchanged tables +skip those reads. Read caches and last-notified values stay separate so an early +`getSnapshot()` cannot consume another observer's notification. + +Sync counts build one scalar index per operations/requests/loading version: +project totals plus canvas→counts. Requests spanning multiple pending operations +count once per matching canvas. Operations with no canvas remain in project +counts. Unknown canvas queries return the shared zero count without growing the +index. This preserves the existing blocked/unknown definitions; table changes +still require a traversal, not a per-record ledger write-set update. + +Asset display effects keep their previous ownership, abort/release and retry +rules. Durable-only completion/error/close is still read from the independent +durability source, so saving and synchronization remain distinct. diff --git a/src/lib/canvas-bindings/react.ts b/src/lib/canvas-bindings/react.ts index e4c19a1..2c33388 100644 --- a/src/lib/canvas-bindings/react.ts +++ b/src/lib/canvas-bindings/react.ts @@ -3,12 +3,13 @@ import {createContext,createElement,useCallback,useContext,useEffect,useMemo,use import {createSelectorStore,type ExternalSource} from './selectors.ts'; import {createElementSources} from './element-sources.ts'; import {createSessionSources,type GestureSelection} from './session-sources.ts'; +import {createRuntimeSources,type OperationView} from './runtime-sources.ts'; +export type {OperationView} from './runtime-sources.ts'; import type {ProjectEngine} from '../canvas-runtime/workspace.ts'; import type {EngineSnapshot} from '../canvas-runtime/contracts.ts'; import type {Target,Id,Data,Geometry,ElementRecord,Availability,Result} from '../canvas-core/types.ts'; import type {ElementLookup} from '../canvas-core/queries.ts'; import {equal,fault} from '../canvas-core/values.ts'; -import type {OperationRecord,FrozenRequest} from '../canvas-runtime/ledger.ts'; import type {DisplayHandle} from '../canvas-runtime/resources.ts'; export type CanvasBinding={readonly engine:ProjectEngine;readonly viewId?:Id}; @@ -68,24 +69,22 @@ export function useGestureBindings(target:Target){ },[engine,viewId,id,incarnation]); useEffect(()=>()=>controls.cancel(),[controls]);return useMemo(()=>Object.freeze({...state,...controls}),[state,controls]); } -export type OperationView={readonly operation:OperationRecord;readonly request?:FrozenRequest}|undefined; -export function useOperation(id:Id):OperationView{return useCanvas(useCallback(root=>{const operation=root.runtime.ledger.operations.get(id);return operation?{operation,...(operation.requestId?{request:root.runtime.ledger.requests.get(operation.requestId)}:{})}:undefined;},[id]),equal);} +const runtimeSources=new WeakMap>(); +function runtimeSourcesFor(engine:ProjectEngine){let sources=runtimeSources.get(engine);if(!sources){sources=createRuntimeSources({getSnapshot:engine.getSnapshot,subscribe:notify=>engine.getDurability().closed?()=>{}:engine.editor.subscribe(notify)});runtimeSources.set(engine,sources);}return sources;} +export function useOperation(id:Id):OperationView{const {engine}=useCanvasBinding(),source=useMemo(()=>runtimeSourcesFor(engine).operation(id),[engine,id]);return useSyncExternalStore(source.subscribe,source.getSnapshot,source.getSnapshot);} type Durability=ReturnType; const syncDurability=(value:Durability)=>({dirty:value.rootRevision>value.durableRootRevision,error:value.error,closing:value.closing,closed:value.closed}); const inactiveOwner=(value:Durability)=>value.closing||value.closed; function useDurability(selector:(value:Durability)=>T,isEqual:(a:T,b:T)=>boolean=Object.is){const {engine}=useCanvasBinding();const source=useMemo(()=>({getSnapshot:engine.getDurability,subscribe:engine.subscribeDurability}),[engine]);return useSelected(source,selector,isEqual);} export function useSyncStatus(canvasId?:Id){ - const durable=useDurability(syncDurability,equal),counts=useCanvas(useCallback(root=>{ - const operations=[...root.runtime.ledger.operations.values()].filter(row=>row.phase!=='resolved'&&(canvasId===undefined||row.intent.canvasIds.includes(canvasId))),ids=new Set(operations.map(row=>row.intent.id)),requests=[...root.runtime.ledger.requests.values()].filter(row=>row.phase!=='resolved'&&row.operationIds.some(id=>ids.has(id))); - const blocked=operations.filter(row=>row.phase==='blocked'&&row.blockers.some(reason=>reason!=='waiting_digest')).length+requests.filter(row=>row.error&&row.retryAt===undefined).length,unknown=requests.filter(row=>row.outcome==='unresolved'&&['dispatch-marked','waiting'].includes(row.phase)).length; - return {pending:operations.length,blocked,unknown,loading:!!root.runtime.loading}; - },[canvasId]),equal); + const {engine}=useCanvasBinding(),source=useMemo(()=>runtimeSourcesFor(engine).counts(canvasId),[engine,canvasId]); + const durable=useDurability(syncDurability,equal),counts=useSyncExternalStore(source.subscribe,source.getSnapshot,source.getSnapshot); const dirty=durable.dirty; return useMemo(()=>Object.freeze({...counts,dirty,error:durable.error,status:durable.closed?'closed' as const:durable.error||counts.blocked?'blocked' as const:durable.closing?'closing' as const:dirty?'saving' as const:counts.pending?'syncing' as const:counts.loading?'loading' as const:'synced' as const}),[counts,dirty,durable.error,durable.closing,durable.closed]); } export function useAsset(id:Id){ const {engine}=useCanvasBinding(),inactive=useDurability(inactiveOwner); - const metadata=useCanvas(useCallback(root=>({record:root.runtime.resources.get(id),binding:root.runtime.assetBindings?.get(id)}),[id]),equal); + const source=useMemo(()=>runtimeSourcesFor(engine).asset(id),[engine,id]),metadata=useSyncExternalStore(source.subscribe,source.getSnapshot,source.getSnapshot); const key=JSON.stringify([metadata.record?.operationId,metadata.record?.retirement?.purgedAt,metadata.binding?.authority,metadata.binding?.digest]); const [attempt,setAttempt]=useState(0),[state,setState]=useState<{engine:ProjectEngine;id:Id;key:string;status:'loading'|'ready'|'error';url?:string;error?:unknown}>(); useEffect(()=>{if(inactive){setState({engine,id,key,status:'error',error:fault('engine_closing')});return;}const controller=new AbortController();let handle:DisplayHandle|undefined;setState({engine,id,key,status:'loading'});void engine.displayResource(id,{signal:controller.signal}).then(value=>{if(controller.signal.aborted){value.release();return;}handle=value;setState({engine,id,key,status:'ready',url:value.url});},error=>{if(!controller.signal.aborted)setState({engine,id,key,status:'error',error});});return()=>{controller.abort();handle?.release();};},[engine,id,key,attempt,inactive]); diff --git a/src/lib/canvas-bindings/runtime-sources.ts b/src/lib/canvas-bindings/runtime-sources.ts new file mode 100644 index 0000000..6a5a1d7 --- /dev/null +++ b/src/lib/canvas-bindings/runtime-sources.ts @@ -0,0 +1,79 @@ +import type {Id} from '../canvas-core/types.ts'; +import type {EngineSnapshot} from '../canvas-runtime/contracts.ts'; +import type {OperationRecord,FrozenRequest} from '../canvas-runtime/ledger.ts'; +import type {ResourceRecord} from '../canvas-runtime/resources.ts'; +import type {SharedAssetBinding} from '../canvas-runtime/asset-depot.ts'; +import {equal} from '../canvas-core/values.ts'; +import type {ExternalSource} from './selectors.ts'; + +export type OperationView={readonly operation:OperationRecord;readonly request?:FrozenRequest}|undefined; +export type AssetMetadata={readonly record:ResourceRecord|undefined;readonly binding:SharedAssetBinding|undefined}; +export type SyncCounts={readonly pending:number;readonly blocked:number;readonly unknown:number;readonly loading:boolean}; +type Dependencies=readonly(object|boolean|undefined)[]; + +/** Read-only table projections: one upstream listener, no per-hook Root/table snapshots. */ +export function createRuntimeSources(source:ExternalSource){ + type Group={update(root:EngineSnapshot,notifications:Set<()=>void>):void}; + const active=new Set(),identities=new WeakMap();let nextIdentity=0,stop:(()=>void)|undefined; + function signature(values:Dependencies){return values.map(value=>{if(value===undefined)return'u';if(typeof value==='boolean')return value?'t':'f';let id=identities.get(value);if(id===undefined){id=++nextIdentity;identities.set(value,id);}return`o${id}`;}).join(':');} + function changed(){ + if(!active.size)return;const root=source.getSnapshot(),notifications=new Set<()=>void>(); + for(const group of active)group.update(root,notifications); + let failed=false,error:unknown;for(const notify of notifications)try{notify();}catch(caught){if(!failed){failed=true;error=caught;}}if(failed)throw error; + } + function group(dependencies:(root:EngineSnapshot)=>Dependencies,read:(root:EngineSnapshot,key:K)=>V){ + // Direct reads may precede notification, so read caches cannot replace observed baselines. + type Bucket={cached:V;cachedSignature:string;observed:V;listeners:Set<()=>void>}; + const buckets=new Map(),failed=new Set();let observedSignature:string|undefined; + function readCached(root:EngineSnapshot,key:K,bucket:Bucket,stamp:string){ + if(bucket.cachedSignature!==stamp){const value=read(root,key);if(!equal(bucket.cached,value))bucket.cached=value;bucket.cachedSignature=stamp;} + return bucket.cached; + } + const owner:Group={update(root,notifications){ + const stamp=signature(dependencies(root)),all=stamp!==observedSignature;if(!all&&!failed.size)return;observedSignature=stamp; + for(const key of all?buckets.keys():[...failed]){ + const bucket=buckets.get(key);if(!bucket)continue; + try{const value=readCached(root,key,bucket,stamp);if(!failed.has(key)&&equal(bucket.observed,value))continue;bucket.observed=value;failed.delete(key);}catch{failed.add(key);} + for(const notify of bucket.listeners)notifications.add(notify); + } + }}; + return{forKey(key:K):ExternalSource{ + let initialized=false,local:V,localSignature:string; + function getSnapshot(){ + const root=source.getSnapshot(),stamp=signature(dependencies(root)),bucket=buckets.get(key);if(bucket)return readCached(root,key,bucket,stamp); + if(!initialized||stamp!==localSignature){const value=read(root,key);if(!initialized||!equal(local,value))local=value;localSignature=stamp;initialized=true;} + return local; + } + return Object.freeze({getSnapshot,subscribe(notify:()=>void){ + let bucket=buckets.get(key);if(!bucket){const value=getSnapshot();if(!buckets.size)observedSignature=signature(dependencies(source.getSnapshot()));bucket={cached:value,cachedSignature:signature(dependencies(source.getSnapshot())),observed:value,listeners:new Set()};buckets.set(key,bucket);active.add(owner);} + const listener=()=>notify();bucket.listeners.add(listener); + const remove=()=>{bucket!.listeners.delete(listener);if(!bucket!.listeners.size){buckets.delete(key);failed.delete(key);}if(!buckets.size){active.delete(owner);observedSignature=undefined;}}; + if(!stop)try{stop=source.subscribe(changed);}catch(error){remove();throw error;} + let live=true;return()=>{if(!live)return;live=false;remove();if(!active.size){stop?.();stop=undefined;}}; + }}); + }}; + } + const operations=group(root=>[root.runtime.ledger.operations,root.runtime.ledger.requests],(root,id)=>{const operation=root.runtime.ledger.operations.get(id);return operation?Object.freeze({operation,...(operation.requestId?{request:root.runtime.ledger.requests.get(operation.requestId)}:{})}):undefined;}); + const assets=group(root=>[root.runtime.resources,root.runtime.assetBindings],(root,id)=>Object.freeze({record:root.runtime.resources.get(id),binding:root.runtime.assetBindings?.get(id)})); + let countStamp:string|undefined,projectCounts:SyncCounts,emptyCounts:SyncCounts,canvasCounts=new Map(); + const countDependencies=(root:EngineSnapshot):Dependencies=>[root.runtime.ledger.operations,root.runtime.ledger.requests,!!root.runtime.loading]; + const counts=group(countDependencies,(root,canvasId)=>{ + const stamp=signature(countDependencies(root)); + if(stamp!==countStamp){ + const loading=!!root.runtime.loading,empty=()=>({pending:0,blocked:0,unknown:0,loading}),project=empty(),byCanvas=new Map>(); + const scope=(id:Id)=>{let row=byCanvas.get(id);if(!row){row=empty();byCanvas.set(id,row);}return row;}; + for(const operation of root.runtime.ledger.operations.values())if(operation.phase!=='resolved'){ + const ids=operation.intent.canvasIds.length>1?new Set(operation.intent.canvasIds):operation.intent.canvasIds,blocked=operation.phase==='blocked'&&operation.blockers.some(reason=>reason!=='waiting_digest');project.pending++;if(blocked)project.blocked++; + for(const id of ids){const row=scope(id);row.pending++;if(blocked)row.blocked++;} + } + for(const request of root.runtime.ledger.requests.values())if(request.phase!=='resolved'){ + let matched=false;const ids=new Set();for(const id of request.operationIds){const operation=root.runtime.ledger.operations.get(id);if(operation&&operation.phase!=='resolved'){matched=true;for(const canvas of operation.intent.canvasIds)ids.add(canvas);}} + if(!matched)continue;const blocked=!!request.error&&request.retryAt===undefined,unknown=request.outcome==='unresolved'&&['dispatch-marked','waiting'].includes(request.phase); + if(blocked)project.blocked++;if(unknown)project.unknown++;for(const id of ids){const row=scope(id);if(blocked)row.blocked++;if(unknown)row.unknown++;} + } + projectCounts=Object.freeze(project);emptyCounts=Object.freeze(empty());canvasCounts=new Map([...byCanvas].map(([id,row])=>[id,Object.freeze(row)]));countStamp=stamp; + } + return canvasId===undefined?projectCounts:canvasCounts.get(canvasId)??emptyCounts; + }); + return Object.freeze({operation:operations.forKey,asset:assets.forKey,counts:(canvasId?:Id)=>counts.forKey(canvasId)}); +} diff --git a/tests/headless/package.integration.mjs b/tests/headless/package.integration.mjs index 394e398..77ce7e7 100644 --- a/tests/headless/package.integration.mjs +++ b/tests/headless/package.integration.mjs @@ -78,10 +78,10 @@ try { import React from 'react';import {renderToStaticMarkup} from 'react-dom/server'; import {createReactRenderers,CanvasRendererCommandsContext,useCanvasRendererCommands} from '@endoleap/headless-canvas/react-flow'; import {createRegistry,builtinComponents} from '@endoleap/headless-canvas/core'; - import {CanvasProvider,useCanvas,useSyncStatus,useElement,useMode,useGestureBindings} from '@endoleap/headless-canvas/react'; - const snapshot={rootRevision:0,runtime:{ledger:{operations:new Map(),requests:new Map()}},document:{projectId:'react-consumer'}};const durable={rootRevision:0,durableRootRevision:0,closed:false,closing:false};const hookEngine={getSnapshot:()=>snapshot,getDurability:()=>durable,subscribeDurability:()=>()=>{},editor:{subscribe:()=>()=>{},getQueries:()=>({lookupElement:()=>({status:'absent'})})},sessions:{getSnapshot:()=>({views:new Map([['ssr',{modes:new Map()}]]),gesture:null}),subscribe:()=>()=>{}}}; - function HookView(){const id=useCanvas(value=>value.document.projectId),status=useSyncStatus(),elementStatus=useElement('missing',view=>view.status),mode=useMode({id:'missing',incarnation:1}),gesture=useGestureBindings({id:'missing',incarnation:1});return React.createElement('span',null,id+':'+status.status+':'+elementStatus+':'+mode.mode+':'+gesture.active);} - assert.equal(renderToStaticMarkup(React.createElement(CanvasProvider,{engine:hookEngine,viewId:'ssr'},React.createElement(HookView))),'react-consumer:synced:absent:null:false'); + import {CanvasProvider,useCanvas,useSyncStatus,useElement,useMode,useGestureBindings,useOperation,useAsset} from '@endoleap/headless-canvas/react'; + const snapshot={rootRevision:0,runtime:{ledger:{operations:new Map(),requests:new Map()},resources:new Map()},document:{projectId:'react-consumer'}};const durable={rootRevision:0,durableRootRevision:0,closed:false,closing:false};const hookEngine={getSnapshot:()=>snapshot,getDurability:()=>durable,subscribeDurability:()=>()=>{},editor:{subscribe:()=>()=>{},getQueries:()=>({lookupElement:()=>({status:'absent'})})},sessions:{getSnapshot:()=>({views:new Map([['ssr',{modes:new Map()}]]),gesture:null}),subscribe:()=>()=>{}}}; + function HookView(){const id=useCanvas(value=>value.document.projectId),status=useSyncStatus(),elementStatus=useElement('missing',view=>view.status),mode=useMode({id:'missing',incarnation:1}),gesture=useGestureBindings({id:'missing',incarnation:1}),operation=useOperation('missing'),asset=useAsset('missing');return React.createElement('span',null,id+':'+status.status+':'+elementStatus+':'+mode.mode+':'+gesture.active+':'+(operation===undefined)+':'+asset.status);} + assert.equal(renderToStaticMarkup(React.createElement(CanvasProvider,{engine:hookEngine,viewId:'ssr'},React.createElement(HookView))),'react-consumer:synced:absent:null:false:true:loading'); const registry=createRegistry(builtinComponents());let dispatched=0,seen;const commands={dispatch(){dispatched++;return{status:'applied',revision:1,operationIds:[]};},transaction(){throw Error('unused');}}; function View(){seen=useCanvasRendererCommands();return React.createElement('span',null,'peer renderer');} const renderers=createReactRenderers([{componentId:'text',dataVersion:1,structure:'node',rendererId:'fixture',render:View}],registry); @@ -92,9 +92,9 @@ try { await writeFile(join(packageRoot,'react-consumer-types.mts'),` import {createElement} from 'react';import type {NodeProps,EdgeProps} from '@xyflow/react'; import {createReactRenderers,type ReactFlowProjectionNode,type ReactFlowProjectionEdge} from '@endoleap/headless-canvas/react-flow'; - import {CanvasProvider,useCanvas,useElement,useActions,useMode,useGestureBindings,useOperation,useAsset,useSyncStatus} from '@endoleap/headless-canvas/react'; + import {CanvasProvider,useCanvas,useElement,useActions,useMode,useGestureBindings,useOperation,useAsset,useSyncStatus,type OperationView} from '@endoleap/headless-canvas/react'; import type {ProjectEngine} from '@endoleap/headless-canvas/runtime'; - function Hooks(){const ref={id:'a',incarnation:1};const title=useCanvas(root=>root.document.projectId);useElement(ref);const selectedId:string|undefined=useElement(ref,view=>view.status==='present'?view.element.id:undefined);useElement(ref,view=>({status:view.status}),(a,b)=>a.status===b.status);void selectedId;const actions=useActions(ref);useMode(ref);useGestureBindings(ref);useOperation('operation');useAsset('resource');useSyncStatus('canvas');return createElement('button',{onClick:()=>actions.rename(title,{label:'Rename'})},title);} + function Hooks(){const ref={id:'a',incarnation:1};const title=useCanvas(root=>root.document.projectId);useElement(ref);const selectedId:string|undefined=useElement(ref,view=>view.status==='present'?view.element.id:undefined);useElement(ref,view=>({status:view.status}),(a,b)=>a.status===b.status);void selectedId;const actions=useActions(ref);useMode(ref);useGestureBindings(ref);const operation:OperationView=useOperation('operation');void operation;useAsset('resource');useSyncStatus('canvas');return createElement('button',{onClick:()=>actions.rename(title,{label:'Rename'})},title);} export function BoundHooks({engine}:{engine:ProjectEngine}){return createElement(CanvasProvider,{engine,viewId:'view'},createElement(Hooks));} function Card({data}:NodeProps){return createElement('div',null,data.title);} function Link({data}:EdgeProps){return createElement('span',null,data?.__headlessRecord?.id);} diff --git a/tests/headless/runtime-sources.test.mjs b/tests/headless/runtime-sources.test.mjs new file mode 100644 index 0000000..3e41e5f --- /dev/null +++ b/tests/headless/runtime-sources.test.mjs @@ -0,0 +1,51 @@ +import {test} from 'node:test';import assert from 'node:assert/strict'; +import {createRuntimeSources} from '../../src/lib/canvas-bindings/runtime-sources.ts'; +class Tracked extends Map{gets=0;scans=0;broken=false;get(id){this.gets++;if(this.broken)throw Error('table unavailable');return super.get(id);}values(){this.scans++;return super.values();}reset(){this.gets=0;this.scans=0;}} +const operation=(id,canvasIds=['a'],phase='pending',blockers=[])=>({intent:{id,canvasIds},phase,blockers}); +function fixture(count=3){let root={document:{},runtime:{ledger:{operations:new Tracked(Array.from({length:count},(_,index)=>[String(index),operation(String(index))])),requests:new Tracked()},resources:new Tracked(Array.from({length:count},(_,index)=>[String(index),{id:String(index),name:'image',operationId:'upload-'+index}])),assetBindings:new Tracked()}},broken=false;const listeners=new Set();const source={getSnapshot:()=>root,subscribe(fn){if(broken)throw Error('subscribe failed');listeners.add(fn);return()=>listeners.delete(fn);}};return{source,get:()=>root,get size(){return listeners.size;},broken(value){broken=value;},emit(next){root=next;for(const fn of [...listeners])fn();}};} + +test('2000 asset/operation subscribers share one listener and unrelated document cuts read no tables',()=>{ + const f=fixture(1000),sources=createRuntimeSources(f.source),seen=[],stops=[];for(let index=0;index<1000;index++){const id=String(index);stops.push(sources.asset(id).subscribe(()=>seen.push('asset'+id)),sources.operation(id).subscribe(()=>seen.push('op'+id)));}assert.equal(f.size,1);const {resources,ledger}=f.get().runtime;resources.reset();ledger.operations.reset();ledger.requests.reset();f.emit({...f.get(),document:{edited:true}});assert.equal(resources.gets+ledger.operations.gets+ledger.requests.gets,0);assert.deepEqual(seen,[]); + const next=new Tracked(resources);next.set('300',{...resources.get('300'),name:'new image'});f.emit({...f.get(),runtime:{...f.get().runtime,resources:next}});assert.deepEqual(seen,['asset300']);assert.equal(ledger.operations.gets+ledger.requests.gets,0);for(const stop of stops)stop();assert.equal(f.size,0); +}); + +test('request-only updates notify the linked operation while unrelated operation views stay stable',()=>{ + const f=fixture(),ops=new Tracked(f.get().runtime.ledger.operations),requests=new Tracked([['request',{id:'request',phase:'waiting',outcome:'accepted'}]]);ops.set('0',{...ops.get('0'),requestId:'request'});f.emit({...f.get(),runtime:{...f.get().runtime,ledger:{operations:ops,requests}}});const sources=createRuntimeSources(f.source),a=sources.operation('0'),b=sources.operation('1'),seen=[],before=b.getSnapshot(),stops=[a.subscribe(()=>seen.push('0')),b.subscribe(()=>seen.push('1'))];const changed=new Tracked(requests);changed.set('request',{id:'request',phase:'waiting',outcome:'unresolved',error:'transport'});f.emit({...f.get(),runtime:{...f.get().runtime,ledger:{operations:ops,requests:changed}}});assert.deepEqual(seen,['0']);assert.equal(a.getSnapshot().request.error,'transport');assert.strictEqual(b.getSnapshot(),before);for(const stop of stops)stop(); +}); + +test('early snapshot reads cannot consume a binding notification before other observers receive it',()=>{ + const f=fixture();let channel;const earlier=f.source.subscribe(()=>channel?.getSnapshot()),sources=createRuntimeSources(f.source);channel=sources.asset('shared');let seen=0;const stop=channel.subscribe(()=>seen++),binding={authority:'scope',digest:'a'.repeat(64),asset:{id:'shared'}};f.emit({...f.get(),runtime:{...f.get().runtime,assetBindings:new Tracked([['shared',binding]])}});assert.equal(seen,1);assert.equal(channel.getSnapshot().binding.digest,binding.digest);stop();earlier();assert.equal(f.size,0); +}); + +test('sync counts index shared multi-canvas requests once and handles project-only operations',()=>{ + const f=fixture(0),ops=new Tracked([['one',operation('one',['a','b','a'])],['two',operation('two',['b'],'blocked',['permission'])],['digest',operation('digest',['a'],'blocked',['waiting_digest'])],['done',operation('done',['a'],'resolved')],['project',operation('project',[])]]),requests=new Tracked([['r1',{phase:'waiting',outcome:'unresolved',error:'transport',operationIds:['one','two']}],['r2',{phase:'waiting',outcome:'unresolved',error:'old',operationIds:['done']}],['r3',{phase:'waiting',outcome:'unresolved',operationIds:['project']}],['r4',{phase:'waiting',outcome:'unresolved',error:'retry',retryAt:100,operationIds:['digest']}]]);f.emit({...f.get(),runtime:{...f.get().runtime,ledger:{operations:ops,requests},loading:{version:6}}});const sources=createRuntimeSources(f.source); + assert.deepEqual(sources.counts().getSnapshot(),{pending:4,blocked:2,unknown:3,loading:true});assert.deepEqual(sources.counts('a').getSnapshot(),{pending:2,blocked:1,unknown:2,loading:true});assert.deepEqual(sources.counts('b').getSnapshot(),{pending:2,blocked:2,unknown:1,loading:true});for(let index=0;index<1000;index++)assert.deepEqual(sources.counts('missing-'+index).getSnapshot(),{pending:0,blocked:0,unknown:0,loading:true});assert.equal(ops.scans,1);assert.equal(requests.scans,1); + const channel=sources.counts('a'),stop=channel.subscribe(()=>{});ops.reset();requests.reset();f.emit({...f.get(),document:{unrelated:true}});channel.getSnapshot();assert.equal(ops.scans+requests.scans,0);f.emit({...f.get(),runtime:{...f.get().runtime,loading:undefined}});assert.equal(channel.getSnapshot().loading,false);assert.equal(ops.scans,1);assert.equal(requests.scans,1);stop(); +}); + +test('cloned equal tables preserve view identity and unsubscribed channels still read current facts',()=>{ + const f=fixture(),sources=createRuntimeSources(f.source),channel=sources.asset('0'),before=channel.getSnapshot();let calls=0;const stop=channel.subscribe(()=>calls++);f.emit({...f.get(),runtime:{...f.get().runtime,resources:new Tracked([...f.get().runtime.resources].map(([id,row])=>[id,structuredClone(row)]))}});assert.equal(calls,0);assert.strictEqual(channel.getSnapshot(),before);stop();const resources=new Tracked(f.get().runtime.resources);resources.delete('0');f.emit({...f.get(),runtime:{...f.get().runtime,resources}});assert.equal(channel.getSnapshot().record,undefined);assert.equal(f.size,0); +}); + +test('lookup failures retry without poisoning cached facts, and failed subscription retains no live group',()=>{ + const f=fixture(),sources=createRuntimeSources(f.source),channel=sources.asset('0');f.broken(true);assert.throws(()=>channel.subscribe(()=>{}),/subscribe failed/);f.broken(false);let calls=0;const stop=channel.subscribe(()=>calls++),before=channel.getSnapshot(),resources=new Tracked(f.get().runtime.resources);resources.broken=true;f.emit({...f.get(),runtime:{...f.get().runtime,resources}});assert.equal(calls,1);assert.throws(channel.getSnapshot,/table unavailable/);resources.broken=false;f.emit({...f.get(),document:{metadata:true}});assert.equal(calls,2);assert.strictEqual(channel.getSnapshot(),before);stop();stop();assert.equal(f.size,0); +}); + +test('observer failures do not starve another collection and late captured callbacks after cleanup are safe',()=>{ + const f=fixture(),sources=createRuntimeSources(f.source);let other=0;const a=sources.asset('0').subscribe(()=>{throw Error('observer failed');}),b=sources.operation('0').subscribe(()=>other++),resources=new Tracked(f.get().runtime.resources),ops=new Tracked(f.get().runtime.ledger.operations);resources.set('0',{id:'0',operationId:'upload',name:'changed'});ops.set('0',operation('0',['a'],'blocked',['permission']));assert.throws(()=>f.emit({...f.get(),runtime:{...f.get().runtime,resources,ledger:{...f.get().runtime.ledger,operations:ops}}}),/observer failed/);assert.equal(other,1);a();b();assert.equal(f.size,0); + let stop;const earlier=f.source.subscribe(()=>stop());stop=sources.counts().subscribe(()=>{throw Error('detached observer');});assert.doesNotThrow(()=>f.emit({...f.get(),runtime:{...f.get().runtime,loading:{}}}));earlier();assert.equal(f.size,0); +}); + +test('request status changes update scoped sync counts and unrelated requests preserve cached count identity',()=>{ + const f=fixture(1),request={phase:'waiting',outcome:'accepted',operationIds:['0']};f.emit({...f.get(),runtime:{...f.get().runtime,ledger:{...f.get().runtime.ledger,requests:new Tracked([['r',request]])}}});const sources=createRuntimeSources(f.source),channel=sources.counts('a');let calls=0;const stop=channel.subscribe(()=>calls++);assert.deepEqual(channel.getSnapshot(),{pending:1,blocked:0,unknown:0,loading:false});const requests=new Tracked([['r',{...request,outcome:'unresolved',error:'transport'}]]);f.emit({...f.get(),runtime:{...f.get().runtime,ledger:{...f.get().runtime.ledger,requests}}});assert.equal(calls,1);const counted=channel.getSnapshot();assert.deepEqual(counted,{pending:1,blocked:1,unknown:1,loading:false});const unrelated=new Tracked(requests);unrelated.set('other',{phase:'waiting',outcome:'unresolved',error:'other',operationIds:['missing']});f.emit({...f.get(),runtime:{...f.get().runtime,ledger:{...f.get().runtime.ledger,requests:unrelated}}});assert.equal(calls,1);assert.strictEqual(sources.counts('a').getSnapshot(),counted);stop(); +}); + +test('indexed counts preserve the previous filter definition over mixed request and operation states',()=>{ + const f=fixture(0),sources=createRuntimeSources(f.source);let seed=7;const random=n=>{seed=(Math.imul(seed,1664525)+1013904223)>>>0;return seed%n;}; + for(let sample=0;sample<40;sample++){ + const operations=new Tracked(Array.from({length:24},(_,index)=>{const id='op'+index,canvases=Array.from({length:random(4)},()=>['a','b','c'][random(3)]);return[id,operation(id,canvases,['pending','blocked','frozen','resolved'][random(4)],random(2)?['waiting_digest']:['permission'])];})),requests=new Tracked(Array.from({length:12},(_,index)=>['r'+index,{phase:['prepared','dispatch-marked','waiting','resolved'][random(4)],outcome:[undefined,'unresolved','accepted','applied'][random(4)],error:random(2)?'transport':undefined,retryAt:random(2)?100:undefined,operationIds:Array.from({length:random(5)},()=>'op'+random(26))}]));f.emit({...f.get(),runtime:{...f.get().runtime,ledger:{operations,requests},loading:sample%2?{}:undefined}}); + for(const scope of [undefined,'a','b','c','missing']){ + const selected=[...operations.values()].filter(row=>row.phase!=='resolved'&&(scope===undefined||row.intent.canvasIds.includes(scope))),ids=new Set(selected.map(row=>row.intent.id)),visible=[...requests.values()].filter(row=>row.phase!=='resolved'&&row.operationIds.some(id=>ids.has(id))),expected={pending:selected.length,blocked:selected.filter(row=>row.phase==='blocked'&&row.blockers.some(reason=>reason!=='waiting_digest')).length+visible.filter(row=>row.error&&row.retryAt===undefined).length,unknown:visible.filter(row=>row.outcome==='unresolved'&&['dispatch-marked','waiting'].includes(row.phase)).length,loading:!!f.get().runtime.loading};assert.deepEqual(sources.counts(scope).getSnapshot(),expected,`sample ${sample}, scope ${scope}`); + } + } +});