Coordinate scoped model reads with bounded streaming state and identity fences
This commit is contained in:
@@ -11,7 +11,7 @@ import { createWorkspaceRuntime } from "@/lib/canvas-runtime/workspace";
|
||||
import { createBrowserDigests } from "@/lib/canvas-runtime/browser-digests";
|
||||
import {createBrowserImageDecoder} from "@/lib/canvas-runtime/browser-images";
|
||||
import { createBrowserStorage } from "@/lib/canvas-runtime/storage";
|
||||
import { createFoundryAssets, createFoundryPersistence, createFoundryTransport, createFoundryRuns } from "@/lib/canvas-service/index";
|
||||
import { createFoundryAssets, createFoundryPersistence, createFoundryTransport, createFoundryRuns,createFoundryModelLoader } from "@/lib/canvas-service/index";
|
||||
|
||||
const Context = createContext<RuntimeHost | null>(null);
|
||||
const Renderers=createContext(defaultCanvasRenderers);
|
||||
@@ -20,6 +20,7 @@ export function CanvasRuntimeHost({ children,extensions=[] }: { children: ReactN
|
||||
const [{host,renderers}] = useState(() => {
|
||||
const registry = createRegistry([...builtinComponents(),...extensions.flatMap(extension=>extension.components??[])]),renderers=createReactRenderers([...defaultCanvasRenderers.definitions,...extensions.flatMap(extension=>extension.renderers??[])],registry);
|
||||
const host=createRuntimeHost({ registry,
|
||||
modelLoader(accountId,workspaceId,_authEpoch,current){return createFoundryModelLoader({workspaceId,registry,transport:createFoundryTransport({accountId,isCurrentIdentity:current})});},
|
||||
workspace(accountId, workspaceId, authEpoch, isCurrentIdentity) {
|
||||
return createWorkspaceRuntime({ accountId, workspaceId, authEpoch, isCurrentIdentity, registry, storage: createBrowserStorage(),digests:createBrowserDigests(),images:createBrowserImageDecoder(),
|
||||
adapters(rootId, current) {
|
||||
|
||||
@@ -1,12 +1,11 @@
|
||||
"use client";
|
||||
|
||||
import { useEffect, useState } from "react";
|
||||
import { useEffect, useRef, useState } from "react";
|
||||
import { useRouter, useSearchParams } from "next/navigation";
|
||||
import { api, errorText } from "@/lib/api";
|
||||
import { readDemoBoards } from "@/lib/demo-workspace";
|
||||
import { matchesRef, type Board } from "@/lib/workspace";
|
||||
import { loadCanvasEntry } from "@/lib/canvas-entry";
|
||||
import { createFoundryTransport } from "@/lib/canvas-service/foundry";
|
||||
import type { ProjectEngine } from "@/lib/canvas-runtime/workspace";
|
||||
import { useCanvasRuntimeHost, useCanvasRenderers } from "./canvas-runtime-host";
|
||||
import type { CanvasSnapshot } from "@/lib/canvas-sync";
|
||||
@@ -27,6 +26,7 @@ export default function GraphPage({ workspaceId, boardId }: { workspaceId: strin
|
||||
const [loaded, setLoaded] = useState<Loaded | null>(null);
|
||||
const [error, setError] = useState("");
|
||||
const [retry, setRetry] = useState(0);
|
||||
const navigationGeneration=useRef(0);
|
||||
useEffect(() => {
|
||||
setLoaded(null); setError("");
|
||||
if (demo) {
|
||||
@@ -37,13 +37,14 @@ export default function GraphPage({ workspaceId, boardId }: { workspaceId: strin
|
||||
}
|
||||
let active = true;
|
||||
const controller = new AbortController();
|
||||
loadCanvasEntry({ request: api, workspaceRef: workspaceId, boardRef: boardId, registry: runtimeHost.registry, signal: controller.signal,
|
||||
transport(accountId) { return createFoundryTransport({ accountId, isCurrentIdentity: runtimeHost.bindAccount(accountId) }); },
|
||||
const startingIdentity=runtimeHost.identity();let currentIdentity:(()=>boolean)|undefined;
|
||||
loadCanvasEntry({ request: api, workspaceRef: workspaceId, boardRef: boardId, registry: runtimeHost.registry, signal: controller.signal,navigationGeneration:++navigationGeneration.current,
|
||||
readModel:input=>{const identity=runtimeHost.identity();if(identity.authEpoch!==startingIdentity.authEpoch&&identity.accountId!==input.accountId)throw new Error('identity_changed');currentIdentity=runtimeHost.bindAccount(input.accountId);return runtimeHost.loadModel(input.accountId,input.workspaceId,input.canvasId,{signal:input.signal,navigationGeneration:input.navigationGeneration});},
|
||||
}).then(async result => {
|
||||
if (!active) return;
|
||||
if (!active||currentIdentity&&!currentIdentity()) return;
|
||||
if (result.mode === 'legacy') { setLoaded({ board: result.board, snapshot: result.snapshot, route }); return; }
|
||||
const engine = await runtimeHost.workspace(result.profile.id, result.board.workspace_id).open(result.model.document, result.model.versions);
|
||||
if (active) setLoaded({ board: result.board, engine, route });
|
||||
if (active&&(!currentIdentity||currentIdentity())) setLoaded({ board: result.board, engine, route });
|
||||
})
|
||||
.catch((error) => { if (active) { setError(errorText(error)); if (error.message === "invalid_session") router.replace("/login"); } });
|
||||
return () => { active = false; controller.abort(); };
|
||||
|
||||
@@ -4,17 +4,17 @@ import { decodeSnapshot } from './canvas-sync.ts';
|
||||
import { CANVAS_ACCEPT } from './canvas-v2.ts';
|
||||
import { matchesRef, type Board } from './workspace.ts';
|
||||
import type { Registry } from './canvas-core/types.ts';
|
||||
import { loadFoundryModel, type ServiceTransport } from './canvas-service/foundry.ts';
|
||||
import { loadFoundryModel, type ModelLoad, type ServiceTransport } from './canvas-service/foundry.ts';
|
||||
|
||||
/** Read canonical first: a migrated project must never be opened through snapshot writers. */
|
||||
export async function loadCanvasEntry(options: { request: CanvasRequest; workspaceRef: string; boardRef: string; registry: Registry; transport(accountId: string): ServiceTransport; signal?: AbortSignal }) {
|
||||
export async function loadCanvasEntry(options: { request: CanvasRequest; workspaceRef: string; boardRef: string; registry: Registry; transport?(accountId: string): ServiceTransport; readModel?(input:{accountId:string;workspaceId:string;canvasId:string;signal?:AbortSignal;navigationGeneration:number}):Promise<ModelLoad>;signal?: AbortSignal;navigationGeneration?:number }) {
|
||||
const path = `workspaces/${encodeURIComponent(options.workspaceRef)}/materials/${encodeURIComponent(options.boardRef)}`;
|
||||
const [board, profile] = await Promise.all([options.request<Board>(path, { signal: options.signal }), options.request<Profile>('profile', { signal: options.signal })]);
|
||||
options.signal?.throwIfAborted();
|
||||
if (board.kind !== 'board' || !matchesRef(board, options.boardRef) || !board.workspace_id || !profile.id) throw new Error('invalid_reference');
|
||||
const transport = options.transport(profile.id);
|
||||
if(!options.readModel&&!options.transport)throw new Error('model_loader_unavailable');
|
||||
try {
|
||||
const model = await loadFoundryModel(transport, board.workspace_id, board.id, options.registry, options.signal);
|
||||
const model = options.readModel?await options.readModel({accountId:profile.id,workspaceId:board.workspace_id,canvasId:board.id,signal:options.signal,navigationGeneration:options.navigationGeneration??0}):await loadFoundryModel(options.transport!(profile.id), board.workspace_id, board.id, options.registry, options.signal);
|
||||
options.signal?.throwIfAborted();
|
||||
if (model.status === 'ready') return { mode: 'headless' as const, board, profile, model };
|
||||
// Existing projects keep their current editor until the user explicitly migrates them.
|
||||
|
||||
@@ -22,6 +22,7 @@ export type RuntimeState = {
|
||||
};
|
||||
export type EngineSnapshot = Snapshot<RuntimeState>;
|
||||
export type CanonicalProposal = { readonly document: Document; readonly versions: ReadonlyMap<Id, number> };
|
||||
export type CanonicalReadResult = ({ readonly status: 'ready' } & CanonicalProposal) | { readonly status: 'migration-required'; readonly rootId: Id; readonly sourceVersions: ReadonlyMap<Id, number> };
|
||||
/** A validated historical execution receipt; its old model is never published over a fresh read. */
|
||||
export type AppliedModelProof = { readonly versions:ReadonlyMap<Id,number>; assertCanonical(proposal:CanonicalProposal):void };
|
||||
export type PersistencePort = {
|
||||
|
||||
@@ -1,17 +1,20 @@
|
||||
import type { Id, Registry } from '../canvas-core/types.ts';
|
||||
import { fault, readMap } from '../canvas-core/values.ts';
|
||||
import type { WorkspaceRuntime } from './workspace.ts';
|
||||
import { createScopedLoader, type LoadPort, type ScopedLoader } from './loader.ts';
|
||||
import type { CanonicalReadResult } from './contracts.ts';
|
||||
|
||||
/** Host lifetime matches the signed-in application, independently of navigation. */
|
||||
export function createRuntimeHost(options: { registry: Registry; workspace(accountId: Id, workspaceId: Id, authEpoch: number, current: () => boolean): WorkspaceRuntime; onError?: (error: unknown) => void }) {
|
||||
export function createRuntimeHost(options: { registry: Registry; workspace(accountId: Id, workspaceId: Id, authEpoch: number, current: () => boolean): WorkspaceRuntime; modelLoader?(accountId:Id,workspaceId:Id,authEpoch:number,current:()=>boolean):LoadPort<CanonicalReadResult>; onError?: (error: unknown) => void }) {
|
||||
let accountId: Id | null = null, authEpoch = 0, disposed = false;
|
||||
const workspaces = new Map<Id, WorkspaceRuntime>(), retired = new Set<WorkspaceRuntime>();
|
||||
const readers = new Map<Id, ScopedLoader<CanonicalReadResult>>();
|
||||
const closeRetired = async () => {
|
||||
const results = await Promise.allSettled([...retired].map(async runtime => { await runtime.close(); retired.delete(runtime); }));
|
||||
const errors = results.flatMap(result => result.status === 'rejected' ? [result.reason] : []);
|
||||
if (errors.length) throw new AggregateError(errors, 'host_close_failed');
|
||||
};
|
||||
function retire() { authEpoch++; accountId = null; for (const runtime of workspaces.values()) retired.add(runtime); workspaces.clear(); }
|
||||
function retire() { authEpoch++; accountId = null; for (const reader of readers.values()) reader.close();readers.clear();for (const runtime of workspaces.values()) retired.add(runtime); workspaces.clear(); }
|
||||
function activate(account: Id) {
|
||||
if (disposed) throw fault('host_disposed');
|
||||
if (accountId !== account) {
|
||||
@@ -24,6 +27,15 @@ export function createRuntimeHost(options: { registry: Registry; workspace(accou
|
||||
bindAccount(account: Id) { activate(account); const epoch = authEpoch; return () => !disposed && accountId === account && authEpoch === epoch; },
|
||||
identity: () => Object.freeze({ accountId, authEpoch }),
|
||||
getWorkspaces: () => readMap(workspaces),
|
||||
async loadModel(account:Id,workspace:Id,canvas:Id,settings:{signal?:AbortSignal;navigationGeneration?:number}={}) {
|
||||
activate(account);
|
||||
const readEpoch=authEpoch;
|
||||
if(!options.modelLoader)throw fault('model_loader_unavailable');
|
||||
let reader=readers.get(workspace);
|
||||
if(!reader){const epoch=authEpoch,current=()=>!disposed&&accountId===account&&authEpoch===epoch;reader=createScopedLoader({load:options.modelLoader(account,workspace,epoch,current),current:()=>current(),maxCacheEntries:0});readers.set(workspace,reader);}
|
||||
const handle=reader.acquire({accountId:account,workspaceId:workspace,projectId:null,canvasId:canvas,viewKey:'canonical-project',format:'canonical-model-v1',queryFingerprint:'full-project',source:'foundry',authEpoch,navigationGeneration:settings.navigationGeneration??0},{signal:settings.signal});
|
||||
try{const snapshot=await handle.ready;if(disposed||accountId!==account||authEpoch!==readEpoch)throw fault('identity_changed');const last=snapshot.chunks.at(-1);if(!last||!last.complete)throw fault('incomplete_load');return last.value;}finally{handle.release();}
|
||||
},
|
||||
workspace(account: Id, workspace: Id) {
|
||||
activate(account);
|
||||
let runtime = workspaces.get(workspace);
|
||||
|
||||
@@ -0,0 +1,141 @@
|
||||
import type { Id } from '../canvas-core/types.ts';
|
||||
import { equal, fault, freeze } from '../canvas-core/values.ts';
|
||||
|
||||
export type LoadScope = {
|
||||
readonly accountId: Id; readonly workspaceId: Id; readonly projectId: Id | null; readonly canvasId: Id;
|
||||
readonly viewKey: string; readonly format: string; readonly queryFingerprint: string; readonly source: string;
|
||||
readonly authEpoch: number; readonly navigationGeneration: number;
|
||||
};
|
||||
export type LoadChunk<T> = {
|
||||
readonly queryKey: string; readonly projectId: Id; readonly snapshotToken: string;
|
||||
readonly serverVersions: Readonly<Record<Id, number>>; readonly permissionEpoch: number;
|
||||
readonly sequence: number; readonly coverage: readonly string[]; readonly complete: boolean;
|
||||
readonly byteLength: number; readonly value: T;
|
||||
};
|
||||
export type LoadSnapshot<T> = {
|
||||
readonly scope: LoadScope; readonly generation: number;
|
||||
readonly cacheStatus: 'bypassed' | 'miss' | 'invalid' | 'hit';
|
||||
readonly status: 'queued' | 'loading' | 'initial-ready' | 'partial' | 'complete' | 'failed' | 'failed-with-partial' | 'cancelled';
|
||||
readonly chunks: readonly LoadChunk<T>[]; readonly coverage: readonly string[]; readonly byteLength: number;
|
||||
readonly error?: { readonly code: string; readonly message: string };
|
||||
};
|
||||
export type LoadPort<T> = (scope: LoadScope, signal: AbortSignal) => AsyncIterable<LoadChunk<T>>;
|
||||
export type LoadHandle<T> = { readonly ready: Promise<LoadSnapshot<T>>; getSnapshot(): LoadSnapshot<T>; release(): void };
|
||||
|
||||
/** Navigation belongs to a subscriber, not the shared query identity. */
|
||||
export function loadQueryKey(scope: LoadScope): string {
|
||||
if (!scope || ![scope.accountId, scope.workspaceId, scope.canvasId, scope.viewKey, scope.format, scope.queryFingerprint, scope.source].every(value => typeof value === 'string' && value.length > 0)
|
||||
|| scope.projectId !== null && (typeof scope.projectId !== 'string' || !scope.projectId)
|
||||
|| ![scope.authEpoch, scope.navigationGeneration].every(value => Number.isSafeInteger(value) && value >= 0)) throw fault('invalid_load_scope');
|
||||
return JSON.stringify([scope.accountId, scope.workspaceId, scope.projectId, scope.canvasId, scope.viewKey, scope.format, scope.queryFingerprint, scope.source, scope.authEpoch]);
|
||||
}
|
||||
|
||||
/** Read copies only. This coordinator never publishes a document or a server ACK.
|
||||
* Adapters validate payloads and declare their measured byte cost before yielding.
|
||||
*/
|
||||
export function createScopedLoader<T>(options: {
|
||||
load: LoadPort<T>; current(scope: LoadScope): boolean; concurrency?: number; maxQueued?: number;
|
||||
maxResponseBytes?: number; maxChunks?: number; maxCacheEntries?: number; maxCacheBytes?: number;
|
||||
cacheMaxAgeMs?: number; now?: () => number;
|
||||
acceptCached?(scope: LoadScope, snapshot: LoadSnapshot<T>): boolean;
|
||||
onObserverError?(error: unknown): void;
|
||||
}) {
|
||||
type Subscriber = { readonly scope: LoadScope; active: boolean; cancelled?: LoadSnapshot<T>; lastState?: LoadSnapshot<T>; snapshot?: LoadSnapshot<T>;resolve(value: LoadSnapshot<T>): void; reject(error: unknown): void; cleanup(): void; notify?(snapshot: LoadSnapshot<T>): void };
|
||||
type Job = { readonly key: string; readonly scope: LoadScope; readonly generation: number; readonly controller: AbortController; readonly subscribers: Set<Subscriber>; state: LoadSnapshot<T>; retired: boolean };
|
||||
const concurrency = options.concurrency ?? 4, maxQueued = options.maxQueued ?? 128, maxBytes = options.maxResponseBytes ?? 64 * 1024 * 1024, maxChunks = options.maxChunks ?? 4096;
|
||||
const maxCacheEntries = options.maxCacheEntries ?? 16, maxCacheBytes = options.maxCacheBytes ?? 16 * 1024 * 1024, cacheMaxAge = options.cacheMaxAgeMs ?? 0;
|
||||
for (const [name, value] of Object.entries({ concurrency, maxQueued, maxBytes, maxChunks, maxCacheEntries, maxCacheBytes, cacheMaxAge })) if (!Number.isSafeInteger(value) || value < (['maxCacheEntries', 'maxCacheBytes', 'cacheMaxAge'].includes(name) ? 0 : 1)) throw fault('invalid_loader_budget');
|
||||
const inflight = new Map<string, Job>(), queue = new Map<number, Job>(), cache = new Map<string, { state: LoadSnapshot<T>; at: number }>();
|
||||
let active = 0, generation = 0, closed = false, cacheBytes = 0;
|
||||
const now = () => { const value = (options.now ?? Date.now)(); if (!Number.isSafeInteger(value) || value < 0) throw fault('invalid_loader_clock'); return value; };
|
||||
const current = (scope: LoadScope) => { if (closed) throw fault('loader_closed'); if (!options.current(scope)) throw fault('identity_changed'); };
|
||||
const view = (job: Job, subscriber: Subscriber): LoadSnapshot<T> => {if(subscriber.lastState!==job.state){subscriber.lastState=job.state;subscriber.snapshot=Object.freeze({...job.state,scope:subscriber.scope});}return subscriber.snapshot!;};
|
||||
const notifyOne = (job: Job, subscriber: Subscriber) => { if (subscriber.active) { try { subscriber.notify?.(view(job, subscriber)); } catch (error) { try { options.onObserverError?.(error); } catch { /* Observers cannot change scheduling. */ } } } };
|
||||
const notify = (job: Job) => { for (const subscriber of [...job.subscribers]) notifyOne(job, subscriber); };
|
||||
const removeCache = (key: string) => { const entry = cache.get(key); if (entry) { cacheBytes -= entry.state.byteLength; cache.delete(key); } };
|
||||
function retire(job: Job) {
|
||||
if (job.retired) return; job.retired = true; job.controller.abort(); queue.delete(job.generation);
|
||||
if (inflight.get(job.key) === job) inflight.delete(job.key);
|
||||
}
|
||||
function pump() {
|
||||
while (!closed && active < concurrency && queue.size) {
|
||||
const job = queue.values().next().value!; queue.delete(job.generation);
|
||||
if (job.retired || !job.subscribers.size) continue;
|
||||
active++; job.state = Object.freeze({ ...job.state, status: 'loading' }); notify(job);
|
||||
void (async () => {
|
||||
try {
|
||||
current(job.scope); let terminal = false, projectId: Id | undefined, token: string | undefined, versions: Readonly<Record<Id, number>> | undefined;
|
||||
for await (const chunk of options.load(job.scope, job.controller.signal)) {
|
||||
if (job.retired) return; current(job.scope);
|
||||
if (!chunk || terminal || chunk.queryKey !== job.key || chunk.permissionEpoch !== job.scope.authEpoch || typeof chunk.projectId !== 'string' || !chunk.projectId
|
||||
|| job.scope.projectId !== null && chunk.projectId !== job.scope.projectId || typeof chunk.snapshotToken !== 'string' || !chunk.snapshotToken
|
||||
|| chunk.sequence !== job.state.chunks.length || typeof chunk.complete !== 'boolean' || !Number.isSafeInteger(chunk.byteLength) || chunk.byteLength < 0
|
||||
|| !Array.isArray(chunk.coverage) || chunk.coverage.some(key => typeof key !== 'string' || !key) || new Set(chunk.coverage).size !== chunk.coverage.length
|
||||
|| !chunk.serverVersions || typeof chunk.serverVersions !== 'object' || Array.isArray(chunk.serverVersions)
|
||||
|| ![Object.prototype,null].includes(Object.getPrototypeOf(chunk.serverVersions))
|
||||
|| Object.entries(chunk.serverVersions).some(([id, version]) => !id || !Number.isSafeInteger(version) || version < 1)) throw fault('invalid_load_chunk');
|
||||
const orderedVersions = Object.fromEntries(Object.entries(chunk.serverVersions).sort(([a], [b]) => a.localeCompare(b)));
|
||||
if (projectId !== undefined && (chunk.projectId !== projectId || chunk.snapshotToken !== token || !equal(orderedVersions, versions))) throw fault('load_snapshot_changed');
|
||||
if (job.state.chunks.length >= maxChunks || chunk.byteLength > maxBytes - job.state.byteLength) throw fault('load_response_budget');
|
||||
projectId = chunk.projectId; token = chunk.snapshotToken; versions = orderedVersions; terminal = chunk.complete;
|
||||
// Own metadata; the adapter supplies an immutable/owned payload value.
|
||||
const owned = Object.freeze({ ...chunk, serverVersions: freeze(orderedVersions), coverage: freeze([...chunk.coverage]) });
|
||||
job.state = Object.freeze({ ...job.state, status: job.state.chunks.length ? 'partial' : 'initial-ready', chunks: Object.freeze([...job.state.chunks, owned]),
|
||||
coverage: freeze([...new Set([...job.state.coverage, ...chunk.coverage])]), byteLength: job.state.byteLength + chunk.byteLength }); notify(job);
|
||||
}
|
||||
if (job.retired) return; current(job.scope); if (!terminal) throw fault('incomplete_load');
|
||||
job.state = Object.freeze({ ...job.state, status: 'complete' });
|
||||
if (maxCacheEntries && maxCacheBytes && cacheMaxAge && job.state.byteLength <= maxCacheBytes) {
|
||||
removeCache(job.key); cache.set(job.key, { state: job.state, at: now() }); cacheBytes += job.state.byteLength;
|
||||
while (cache.size > maxCacheEntries || cacheBytes > maxCacheBytes) removeCache(cache.keys().next().value!);
|
||||
}
|
||||
notify(job); for (const subscriber of job.subscribers) if (subscriber.active) { subscriber.cleanup(); subscriber.resolve(view(job, subscriber)); }
|
||||
} catch (error) {
|
||||
if (job.retired) return;
|
||||
const failure = error as { code?: string; message?: string };
|
||||
job.state = Object.freeze({ ...job.state, status: job.state.chunks.length ? 'failed-with-partial' : 'failed', error: freeze({ code: typeof failure?.code === 'string' ? failure.code : 'load_failed', message: typeof failure?.message === 'string' ? failure.message : 'Load failed' }) });
|
||||
notify(job); for (const subscriber of job.subscribers) if (subscriber.active) { subscriber.cleanup(); subscriber.reject(error); }
|
||||
} finally {
|
||||
if (inflight.get(job.key) === job) inflight.delete(job.key);
|
||||
active--; pump();
|
||||
}
|
||||
})();
|
||||
}
|
||||
}
|
||||
return Object.freeze({
|
||||
acquire(scopeInput: LoadScope, settings: { signal?: AbortSignal; onUpdate?: (snapshot: LoadSnapshot<T>) => void; cache?: 'fresh' | 'allow-validated' } = {}): LoadHandle<T> {
|
||||
if(!scopeInput)throw fault('invalid_load_scope');
|
||||
const scope:LoadScope=Object.freeze({accountId:scopeInput.accountId,workspaceId:scopeInput.workspaceId,projectId:scopeInput.projectId,canvasId:scopeInput.canvasId,viewKey:scopeInput.viewKey,format:scopeInput.format,queryFingerprint:scopeInput.queryFingerprint,source:scopeInput.source,authEpoch:scopeInput.authEpoch,navigationGeneration:scopeInput.navigationGeneration});
|
||||
const key = loadQueryKey(scope); current(scope); if (settings.signal?.aborted) throw fault('load_aborted');
|
||||
let cacheStatus: LoadSnapshot<T>['cacheStatus'] = settings.cache === 'allow-validated' ? 'miss' : 'bypassed';
|
||||
let cached = cache.get(key);
|
||||
if (cached) {
|
||||
const age = now() - cached.at;
|
||||
if (age < 0 || age > cacheMaxAge || settings.cache !== 'allow-validated' || !options.acceptCached || options.acceptCached(scope, cached.state) !== true) { if(settings.cache==='allow-validated')cacheStatus='invalid';removeCache(key); cached = undefined; }
|
||||
else { cacheStatus='hit';cache.delete(key); cache.set(key, cached); }
|
||||
}
|
||||
current(scope);
|
||||
if(settings.signal?.aborted)throw fault('load_aborted');
|
||||
let job = inflight.get(key);
|
||||
if (!job) {
|
||||
if (!cached && queue.size >= maxQueued + Math.max(0, concurrency - active)) throw fault('load_capacity');
|
||||
if (generation >= Number.MAX_SAFE_INTEGER) throw fault('load_generation_exhausted');
|
||||
job = { key, scope, generation: ++generation, controller: new AbortController(), subscribers: new Set(), retired: false,
|
||||
state: cached ? Object.freeze({ ...cached.state, scope, generation,cacheStatus }) : Object.freeze({ scope, generation,cacheStatus,status: 'queued', chunks: Object.freeze([]), coverage: Object.freeze([]), byteLength: 0 }) };
|
||||
if (!cached) { inflight.set(key, job); queue.set(job.generation, job); }
|
||||
}
|
||||
const shared = job; let subscriber: Subscriber;
|
||||
const ready = new Promise<LoadSnapshot<T>>((resolve, reject) => { subscriber = { scope, active: true, resolve, reject, notify: settings.onUpdate, cleanup: () => settings.signal?.removeEventListener('abort', release) }; });
|
||||
const release = () => { if (!subscriber.active) return; subscriber.cancelled=Object.freeze({...view(shared,subscriber),status:'cancelled'});subscriber.active = false; subscriber.cleanup(); shared.subscribers.delete(subscriber); subscriber.reject(fault('load_aborted')); if (!shared.subscribers.size) retire(shared); };
|
||||
shared.subscribers.add(subscriber!); settings.signal?.addEventListener('abort', release, { once: true });
|
||||
if (cached) { subscriber!.resolve(view(shared, subscriber!)); notifyOne(shared, subscriber!); } else { notifyOne(shared, subscriber!); pump(); }
|
||||
return Object.freeze({ ready, getSnapshot: () => subscriber.cancelled??view(shared,subscriber), release });
|
||||
},
|
||||
close() {
|
||||
if (closed) return; closed = true;
|
||||
for (const job of [...inflight.values()]) { retire(job); for (const subscriber of job.subscribers) if (subscriber.active) { subscriber.cancelled=Object.freeze({...view(job,subscriber),status:'cancelled'});subscriber.active = false; subscriber.cleanup(); subscriber.reject(fault('loader_closed')); } job.subscribers.clear(); }
|
||||
queue.clear(); cache.clear(); cacheBytes = 0;
|
||||
},
|
||||
});
|
||||
}
|
||||
export type ScopedLoader<T> = ReturnType<typeof createScopedLoader<T>>;
|
||||
@@ -1,7 +1,7 @@
|
||||
import type { CanvasRecord, Document, ElementRecord, Id, Json, OperationIntent, Registry } from '../canvas-core/types.ts';
|
||||
import { createDocument, validateDocument } from '../canvas-core/document.ts';
|
||||
import { equal, fault, freeze, readMap, validJson } from '../canvas-core/values.ts';
|
||||
import type { AppliedModelProof, EngineSnapshot, PersistencePort } from '../canvas-runtime/contracts.ts';
|
||||
import type { AppliedModelProof, CanonicalReadResult, EngineSnapshot, PersistencePort } from '../canvas-runtime/contracts.ts';
|
||||
import type { FrozenRequest } from '../canvas-runtime/ledger.ts';
|
||||
import { projectBatch } from '../canvas-runtime/batch-projection.ts';
|
||||
import type { OperationStatus } from '../canvas-runtime/scheduler.ts';
|
||||
@@ -62,18 +62,20 @@ function modelChunk(value: unknown): ModelChunk {
|
||||
if (new Set(removed).size !== removed.length || removed.some(value => canvasIds.has(value) || !serverVersions.has(value)) || serverVersions.size !== canvasIds.size + removed.length) throw fault('invalid_canonical_response');
|
||||
return freeze({ projectId, complete: raw.complete, canvases, elements, tombstones: readMap(tombstones), versions: serverVersions, removed });
|
||||
}
|
||||
export type ModelLoad = { readonly status: 'ready'; readonly document: Document; readonly versions: ReadonlyMap<Id, number> } | { readonly status: 'migration-required'; readonly rootId: Id; readonly sourceVersions: ReadonlyMap<Id, number> };
|
||||
export type ModelLoad = CanonicalReadResult;
|
||||
export async function loadFoundryModel(transport: ServiceTransport, workspaceId: Id, canvasId: Id, registry: Registry, signal?: AbortSignal): Promise<ModelLoad> {
|
||||
const response = await transport({ method: 'GET', path: pathFor(workspaceId, canvasId), signal }), body = record(response.body);
|
||||
if (response.status === 409 && body.error === 'canonical_migration_required') {
|
||||
if (body.document_format_version !== 1) throw fault('unsupported_document_format');
|
||||
const rootId = id(body.root_id); if (!uuid.test(rootId)) throw fault('invalid_canonical_response');
|
||||
return freeze({ status: 'migration-required', rootId, sourceVersions: versions(body.source_versions) });
|
||||
const sourceVersions=versions(body.source_versions);if(!sourceVersions.has(canvasId)||!sourceVersions.has(rootId))throw fault('project_mismatch');
|
||||
return freeze({ status: 'migration-required', rootId, sourceVersions });
|
||||
}
|
||||
if (response.status !== 200) throw fault(typeof body.error === 'string' ? body.error : 'model_load_failed');
|
||||
const chunk = modelChunk(body.model);
|
||||
if (!chunk.complete) throw fault('incomplete_initial_model');
|
||||
const document = createDocument(chunk.projectId, chunk.canvases, chunk.elements, chunk.tombstones); validateDocument(document, registry);
|
||||
if(!document.canvases.has(canvasId))throw fault('project_mismatch');
|
||||
return Object.freeze({ status: 'ready', document, versions: chunk.versions });
|
||||
}
|
||||
|
||||
|
||||
@@ -2,3 +2,4 @@ export * from './foundry.ts';
|
||||
export * from './assets.ts';
|
||||
|
||||
export * from './runs.ts';
|
||||
export * from './model-loader.ts';
|
||||
|
||||
@@ -0,0 +1,24 @@
|
||||
import type { Id, Registry } from '../canvas-core/types.ts';
|
||||
import { fault } from '../canvas-core/values.ts';
|
||||
import type { CanonicalReadResult } from '../canvas-runtime/contracts.ts';
|
||||
import { loadQueryKey, type LoadPort } from '../canvas-runtime/loader.ts';
|
||||
import { loadFoundryModel, type ServiceTransport } from './foundry.ts';
|
||||
|
||||
/** The current endpoint is complete-only. One validated chunk is intentional;
|
||||
* this adapter does not claim to stream an already-buffered response progressively.
|
||||
*/
|
||||
export function createFoundryModelLoader(options: { workspaceId: Id; registry: Registry; transport: ServiceTransport }): LoadPort<CanonicalReadResult> {
|
||||
return async function* (scope, signal) {
|
||||
if (scope.workspaceId !== options.workspaceId || scope.format !== 'canonical-model-v1' || scope.source !== 'foundry' || scope.viewKey !== 'canonical-project') throw fault('invalid_model_load_scope');
|
||||
const value = await loadFoundryModel(options.transport, options.workspaceId, scope.canvasId, options.registry, signal);
|
||||
signal.throwIfAborted();
|
||||
const projectId = value.status === 'ready' ? value.document.projectId : value.rootId;
|
||||
const versions = Object.fromEntries([...(value.status === 'ready' ? value.versions : value.sourceVersions)].sort(([a], [b]) => a.localeCompare(b)));
|
||||
const token = JSON.stringify([projectId, versions]);
|
||||
const payload = value.status === 'ready' ? { projectId, canvases: [...value.document.canvases.values()], elements: [...value.document.elements.values()], tombstones: [...value.document.tombstones], versions } : { projectId, versions, migrationRequired: true };
|
||||
// A serialized payload cost, not a claim about pre-parse network/heap limits.
|
||||
const byteLength = new TextEncoder().encode(JSON.stringify(payload)).byteLength;
|
||||
yield Object.freeze({ queryKey: loadQueryKey(scope), projectId, snapshotToken: token, serverVersions: Object.freeze(versions), permissionEpoch: scope.authEpoch,
|
||||
sequence: 0, coverage: Object.freeze(value.status === 'ready' ? [...value.document.canvases.keys()].map(id => `canvas:${id}`) : [`migration:${scope.canvasId}`]), complete: true, byteLength, value });
|
||||
};
|
||||
}
|
||||
@@ -11,6 +11,28 @@ import { createFoundryRuns } from '../../src/lib/canvas-service/runs.ts';
|
||||
import { createFoundryAssets } from '../../src/lib/canvas-service/assets.ts';
|
||||
import { createFoundryPersistence } from '../../src/lib/canvas-service/foundry.ts';
|
||||
import { createResourceView } from '../../src/lib/canvas-bindings/resource-view.ts';
|
||||
import { createScopedLoader,loadQueryKey } from '../../src/lib/canvas-runtime/loader.ts';
|
||||
|
||||
async function scopedNetworkLoaderCase(assert) {
|
||||
const scope={accountId:'loader-account',workspaceId:'loader-workspace',projectId:'loader-project',canvasId:'loader-canvas',viewKey:'canvas',format:'fixture-ndjson-v1',queryFingerprint:'all',source:'network-fixture',authEpoch:1,navigationGeneration:1};
|
||||
async function run(fail) {
|
||||
const key=crypto.randomUUID();let signal,first;const initial=new Promise(resolve=>{first=resolve;});
|
||||
const loader=createScopedLoader({current:()=>true,async *load(query,abort){
|
||||
signal=abort;const response=await fetch(`/loader-stream?key=${key}`,{signal:abort}),reader=response.body.getReader(),decoder=new TextDecoder();let pending='',sequence=0;
|
||||
try{while(true){const next=await reader.read();pending+=decoder.decode(next.value??new Uint8Array(),{stream:!next.done});let end;while((end=pending.indexOf('\n'))>=0){const line=pending.slice(0,end);pending=pending.slice(end+1);if(!line)continue;const value=JSON.parse(line);yield{queryKey:loadQueryKey(query),projectId:'loader-project',snapshotToken:key,serverVersions:{'loader-canvas':1},permissionEpoch:query.authEpoch,sequence:sequence++,coverage:[`part-${value.part}`],complete:value.complete,byteLength:new TextEncoder().encode(line).byteLength,value:Object.freeze(value)};}if(next.done)break;}}finally{await reader.cancel().catch(()=>{});}
|
||||
}});
|
||||
const a=loader.acquire(scope,{onUpdate:state=>{if(state.status==='initial-ready')first();}}),b=loader.acquire({...scope,navigationGeneration:2});
|
||||
const rejected=fail?b.ready.then(()=>{throw Error('invalid tail became complete');},()=>{}):undefined;
|
||||
try{
|
||||
await initial;assert(a.getSnapshot().chunks.length===1&&b.getSnapshot().status==='initial-ready','first network chunk waited for the tail');
|
||||
const cancelled=a.ready.then(()=>{throw Error('cancelled subscriber completed');},error=>assert(error.code==='load_aborted','subscriber cancellation changed'));a.release();await cancelled;assert(!signal.aborted,'one subscriber cancelled the shared fetch');
|
||||
const response=await(await fetch(`/loader-release?key=${key}${fail?'&fail=1':''}`)).json();assert(response.reads===1,'subscribers fetched the same query twice');
|
||||
if(fail){await rejected;assert(b.getSnapshot().status==='failed-with-partial'&&b.getSnapshot().chunks.length===1,'invalid network tail discarded or completed the prefix');}
|
||||
else{const result=await b.ready;assert(result.status==='complete'&&result.chunks.length===2&&result.scope.navigationGeneration===2,'stream completion lost subscriber scope or chunks');assert(a.getSnapshot().status==='cancelled'&&a.getSnapshot().chunks.length===1,'cancelled subscriber observed its peer tail');}
|
||||
}finally{a.release();b.release();loader.close();}
|
||||
}
|
||||
await run(false);await run(true);
|
||||
}
|
||||
|
||||
async function sharedDisplayReadersCase(engine, storage, assert) {
|
||||
const resource = engine.getSnapshot().document.elements.get('picture').data.assetId, read = storage.readFile, createURL = URL.createObjectURL, revokeURL = URL.revokeObjectURL;
|
||||
@@ -727,6 +749,8 @@ export async function run() {
|
||||
results.push('scoped paged inventory finds legacy orphans without hiding references; orphan v5 deletion rolls back faults, retries, isolates projects and forbids byte identity reuse');
|
||||
await resourceRecoveryCase(name, assert);
|
||||
results.push('staged file and image recover from real IndexedDB; lost upload ACK queries stable resource identity');
|
||||
await scopedNetworkLoaderCase(assert);
|
||||
results.push('scoped loader shares one real streamed fetch, exposes the first chunk before tail release, keeps the other subscriber alive after cancellation and preserves partial data after a malformed tail');
|
||||
await defaultImageSizeCase(name, assert);
|
||||
results.push('real image Worker enriches creation history in recovery v6; IndexedDB reopening queries a lost size ACK; concurrent views share one read/URL, retain leases incrementally and release only the final display pin before sized undo/redo');
|
||||
await imageDecodeWorkerCase(assert);
|
||||
|
||||
@@ -8,9 +8,22 @@ import { stripTypeScriptTypes } from 'node:module';
|
||||
import { setTimeout as pause } from 'node:timers/promises';
|
||||
|
||||
const root = resolve(import.meta.dirname, '../..'), profile = await mkdtemp(join(tmpdir(), 'endoleap-storage-browser-'));
|
||||
const readStreams=new Map();
|
||||
const server = createServer(async (request, response) => {
|
||||
try {
|
||||
const path = new URL(request.url, 'http://localhost').pathname;
|
||||
if(path==='/loader-stream'){
|
||||
const url=new URL(request.url,'http://localhost'),key=url.searchParams.get('key');
|
||||
if(!key){response.writeHead(400).end();return;}
|
||||
const state=readStreams.get(key)??{reads:0};state.reads++;state.response=response;readStreams.set(key,state);
|
||||
response.writeHead(200,{'Content-Type':'application/x-ndjson'});response.write(JSON.stringify({part:0,complete:false})+'\n');return;
|
||||
}
|
||||
if(path==='/loader-release'){
|
||||
const url=new URL(request.url,'http://localhost'),state=readStreams.get(url.searchParams.get('key'));
|
||||
if(!state){response.writeHead(404).end();return;}
|
||||
if(!state.response.destroyed)state.response.end(url.searchParams.get('fail')==='1'?'{invalid tail}\n':JSON.stringify({part:1,complete:true})+'\n');
|
||||
response.setHeader('Content-Type','application/json');response.end(JSON.stringify({reads:state.reads}));return;
|
||||
}
|
||||
if (path === '/') { response.setHeader('Content-Type', 'text/html'); response.end('<title>Headless storage checks</title>'); return; }
|
||||
const filename = resolve(root, `.${path}`);
|
||||
if (!filename.startsWith(`${root}/src/lib/`) && !filename.startsWith(`${root}/tests/headless/`)) { response.writeHead(404).end(); return; }
|
||||
|
||||
@@ -0,0 +1,102 @@
|
||||
import { test } from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { createScopedLoader, loadQueryKey } from '../../src/lib/canvas-runtime/loader.ts';
|
||||
const scope = { accountId: 'account', workspaceId: 'workspace', projectId: 'project', canvasId: 'canvas', viewKey: 'canvas', format: 'v1', queryFingerprint: 'all', source: 'fixture', authEpoch: 1, navigationGeneration: 1 };
|
||||
const deferred = () => { let resolve, reject; const promise = new Promise((yes, no) => { resolve = yes; reject = no; }); return { promise, resolve, reject }; };
|
||||
const tick = () => new Promise(resolve => setImmediate(resolve));
|
||||
const chunk = (query, sequence = 0, complete = true, extra = {}) => ({ queryKey: loadQueryKey(query), projectId: 'project', snapshotToken: 'snapshot', serverVersions: { canvas: 3 }, permissionEpoch: query.authEpoch, sequence, coverage: [`part-${sequence}`], complete, byteLength: 2, value: Object.freeze({ part: sequence }), ...extra });
|
||||
|
||||
test('subscribers share a progressive read while retaining navigation scopes; cancelling one preserves the other', async () => {
|
||||
const gate = deferred(), updates = []; let calls = 0, signal;
|
||||
const loader = createScopedLoader({ current: () => true, async *load(query, abort) { calls++; signal = abort; yield chunk(query, 0, false); await gate.promise; yield chunk(query, 1); } });
|
||||
const a = loader.acquire(scope, { onUpdate: value => updates.push(value.status) }); await tick();
|
||||
assert.equal(a.getSnapshot().status, 'initial-ready'); assert.equal(a.getSnapshot().chunks.length, 1);
|
||||
assert.strictEqual(a.getSnapshot(),a.getSnapshot());
|
||||
const b = loader.acquire({ ...scope, navigationGeneration: 2 }); assert.equal(b.getSnapshot().scope.navigationGeneration, 2);
|
||||
const cancelled = assert.rejects(a.ready, { code: 'load_aborted' }); a.release(); await cancelled;const stopped=a.getSnapshot(); assert.equal(signal.aborted, false);
|
||||
gate.resolve(); const result = await b.ready; assert.equal(calls, 1); assert.equal(result.status, 'complete'); assert.deepEqual(result.coverage, ['part-0', 'part-1']);
|
||||
assert.equal(a.getSnapshot().status,'cancelled');assert.equal(a.getSnapshot().chunks.length,1);assert.deepEqual(a.getSnapshot().coverage,['part-0']);
|
||||
assert.strictEqual(a.getSnapshot(),stopped);assert.strictEqual(b.getSnapshot(),result);
|
||||
assert.deepEqual(updates, ['queued', 'loading', 'initial-ready']); b.release(); loader.close();
|
||||
});
|
||||
test('tail failure keeps the accepted prefix and a fresh retry replaces the failed inflight entry', async () => {
|
||||
let calls = 0;
|
||||
const loader = createScopedLoader({ current: () => true, async *load(query) { calls++; yield chunk(query, 0, false); if (calls === 1) throw Object.assign(Error('tail_failed'), { code: 'tail_failed' }); yield chunk(query, 1); } });
|
||||
const a = loader.acquire(scope); await assert.rejects(a.ready, { code: 'tail_failed' }); assert.equal(a.getSnapshot().status, 'failed-with-partial'); assert.equal(a.getSnapshot().chunks[0].value.part, 0);
|
||||
const b = loader.acquire(scope); assert.equal((await b.ready).chunks.length, 2); assert.equal(calls, 2); a.release(); b.release(); loader.close();
|
||||
});
|
||||
test('a complete marker is not authoritative before iterator EOF; a checksum/trailing failure retains partial status', async () => {
|
||||
const statuses = [], loader = createScopedLoader({ current: () => true, async *load(query) { yield chunk(query); throw Error('trailing_checksum'); } });
|
||||
const handle = loader.acquire(scope, { onUpdate: value => statuses.push(value.status) }); await assert.rejects(handle.ready, /trailing_checksum/);
|
||||
assert.equal(handle.getSnapshot().status, 'failed-with-partial'); assert.equal(statuses.includes('complete'), false); handle.release(); loader.close();
|
||||
});
|
||||
test('snapshot token, version vector and project changes reject the tail without mixing it into accepted coverage', async () => {
|
||||
for (const extra of [{ snapshotToken: 'other' }, { serverVersions: { canvas: 4 } }, { projectId: 'other' }]) {
|
||||
const loader = createScopedLoader({ current: () => true, async *load(query) { yield chunk(query, 0, false); yield chunk(query, 1, true, extra); } });
|
||||
const handle = loader.acquire(scope); await assert.rejects(handle.ready); assert.equal(handle.getSnapshot().status, 'failed-with-partial'); assert.equal(handle.getSnapshot().chunks.length, 1); assert.deepEqual(handle.getSnapshot().coverage, ['part-0']); handle.release(); loader.close();
|
||||
}
|
||||
});
|
||||
test('query, permission, sequence and malformed coverage headers reject before exposing a chunk', async () => {
|
||||
for (const extra of [{ queryKey: 'foreign' }, { permissionEpoch: 2 }, { sequence: 3 }, { coverage: ['x', 'x'] }, { byteLength: -1 }, { serverVersions: { canvas: 0 } }]) {
|
||||
const loader = createScopedLoader({ current: () => true, async *load(query) { yield chunk(query, 0, true, extra); } });
|
||||
const handle = loader.acquire(scope); await assert.rejects(handle.ready, { code: 'invalid_load_chunk' }); assert.equal(handle.getSnapshot().chunks.length, 0); assert.equal(handle.getSnapshot().status, 'failed'); handle.release(); loader.close();
|
||||
}
|
||||
});
|
||||
test('an incomplete EOF or extra chunk after a complete marker cannot become a complete result', async () => {
|
||||
for (const extra of [false, true]) {
|
||||
const loader = createScopedLoader({ current: () => true, async *load(query) { yield chunk(query, 0, extra); if (extra) yield chunk(query, 1); } });
|
||||
const handle = loader.acquire(scope); await assert.rejects(handle.ready); assert.equal(handle.getSnapshot().status, 'failed-with-partial'); handle.release(); loader.close();
|
||||
}
|
||||
});
|
||||
test('response byte and chunk budgets bound retained read copies while preserving earlier chunks', async () => {
|
||||
for (const limits of [{ maxResponseBytes: 3 }, { maxChunks: 1 }]) {
|
||||
const loader = createScopedLoader({ ...limits, current: () => true, async *load(query) { yield chunk(query, 0, false); yield chunk(query, 1); } });
|
||||
const handle = loader.acquire(scope); await assert.rejects(handle.ready, { code: 'load_response_budget' }); assert.equal(handle.getSnapshot().byteLength, 2); assert.equal(handle.getSnapshot().chunks.length, 1); handle.release(); loader.close();
|
||||
}
|
||||
});
|
||||
test('cancelled active work retains its concurrency slot; its late cleanup cannot delete the newer generation', async () => {
|
||||
const first = deferred(), second = deferred(), entered = []; let calls = 0;
|
||||
const loader = createScopedLoader({ concurrency: 1, maxQueued: 1, current: () => true, async *load(query) { const index = ++calls; entered.push(index); await (index === 1 ? first.promise : second.promise); yield chunk(query); } });
|
||||
const a = loader.acquire(scope); await tick(); const cancelled = assert.rejects(a.ready, { code: 'load_aborted' }); a.release(); await cancelled;
|
||||
const b = loader.acquire(scope); await tick(); assert.equal(calls, 1); assert.equal(b.getSnapshot().status, 'queued');
|
||||
assert.throws(() => loader.acquire({ ...scope, canvasId: 'third' }), { code: 'load_capacity' });
|
||||
first.resolve(); await tick(); assert.equal(calls, 2); const peer = loader.acquire(scope); assert.equal(peer.getSnapshot().generation, b.getSnapshot().generation);
|
||||
second.resolve(); await Promise.all([b.ready, peer.ready]); assert.deepEqual(entered, [1, 2]); b.release(); peer.release(); loader.close();
|
||||
});
|
||||
test('queued cancellation frees admission capacity and never executes the abandoned source', async () => {
|
||||
const gate = deferred(), called = [], loader = createScopedLoader({ concurrency: 1, maxQueued: 1, current: () => true, async *load(query) { called.push(query.canvasId); await gate.promise; yield chunk(query); } });
|
||||
const a = loader.acquire(scope), b = loader.acquire({ ...scope, canvasId: 'cancelled' }); const cancelled = assert.rejects(b.ready, { code: 'load_aborted' }); b.release(); await cancelled;
|
||||
const c = loader.acquire({ ...scope, canvasId: 'next' }); gate.resolve(); await Promise.all([a.ready, c.ready]); assert.deepEqual(called, ['canvas', 'next']); a.release(); c.release(); loader.close();
|
||||
});
|
||||
test('read cache requires explicit source validation and is bounded by byte LRU and permission age', async () => {
|
||||
let time = 100, accepted = true; const calls = new Map();
|
||||
const loader = createScopedLoader({ maxCacheEntries: 10, maxCacheBytes: 4, cacheMaxAgeMs: 100, now: () => time, acceptCached: () => accepted, current: () => true,
|
||||
async *load(query) { calls.set(query.canvasId, (calls.get(query.canvasId) ?? 0) + 1); yield chunk(query); } });
|
||||
const read = async (canvasId, cached = false) => { const handle = loader.acquire({ ...scope, canvasId }, { cache: cached ? 'allow-validated' : 'fresh' }); await handle.ready; handle.release(); };
|
||||
await read('a'); await read('b'); await read('a', true); assert.equal(calls.get('a'), 1); await read('c'); await read('b', true); assert.equal(calls.get('b'), 2);
|
||||
accepted = false; await read('b', true); assert.equal(calls.get('b'), 3); accepted = true; time = 201; await read('b', true); assert.equal(calls.get('b'), 4);
|
||||
time = 0; await read('b', true); assert.equal(calls.get('b'), 5); loader.close();
|
||||
});
|
||||
test('different query fingerprints and permission epochs do not share requests', async () => {
|
||||
let calls = 0; const loader = createScopedLoader({ current: () => true, async *load(query) { calls++; yield chunk(query); } });
|
||||
const handles = [scope, { ...scope, queryFingerprint: 'filtered' }, { ...scope, authEpoch: 2 }].map(query => loader.acquire(query));
|
||||
await Promise.all(handles.map(handle => handle.ready)); assert.equal(calls, 3); handles.forEach(handle => handle.release()); loader.close();
|
||||
});
|
||||
test('scope capture owns only declared metadata without freezing unrelated caller state', async () => {
|
||||
const extra={filter:'mutable'},input={...scope,extra};const loader=createScopedLoader({current:()=>true,async *load(query){yield chunk(query);}});
|
||||
const handle=loader.acquire(input);assert.equal(Object.isFrozen(extra),false);input.canvasId='later';extra.filter='changed';
|
||||
const result=await handle.ready;assert.equal(result.scope.canvasId,'canvas');assert.equal(Object.hasOwn(result.scope,'extra'),false);assert.equal(Object.isFrozen(result.scope),true);handle.release();loader.close();
|
||||
});
|
||||
test('identity changes discard late chunks, and close promptly rejects waiting readers even if the source ignores abort', async () => {
|
||||
const gate = deferred(); let current = true;
|
||||
const loader = createScopedLoader({ current: () => current, async *load(query) { await gate.promise; yield chunk(query); } });
|
||||
const a = loader.acquire(scope); const rejected = assert.rejects(a.ready, { code: 'identity_changed' }); current = false; gate.resolve(); await rejected; assert.equal(a.getSnapshot().chunks.length, 0); a.release(); loader.close();
|
||||
const pending = deferred(), closed = createScopedLoader({ current: () => true, async *load(query) { await pending.promise; yield chunk(query); } });
|
||||
const b = closed.acquire(scope); const stopped = assert.rejects(b.ready, { code: 'loader_closed' }); closed.close(); await stopped; pending.resolve(); await tick(); assert.equal(b.getSnapshot().status, 'cancelled'); assert.throws(() => closed.acquire(scope), { code: 'loader_closed' });
|
||||
});
|
||||
test('observer reentrancy does not re-notify old subscribers or allow observer exceptions to fail a read', async () => {
|
||||
const errors = []; let peer, updates = 0;
|
||||
const loader = createScopedLoader({ current: () => true, onObserverError: error => errors.push(error), async *load(query) { yield chunk(query); } });
|
||||
const a = loader.acquire(scope, { onUpdate: state => { updates++; if (state.status === 'initial-ready' && !peer) peer = loader.acquire(scope, { onUpdate: () => { throw Error('observer'); } }); } });
|
||||
await a.ready; await peer.ready; assert.equal(updates, 4); assert.ok(errors.length > 0); a.release(); peer.release(); loader.close();
|
||||
});
|
||||
@@ -0,0 +1,58 @@
|
||||
import { test } from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { createRegistry, builtinComponents } from '../../src/lib/canvas-core/index.ts';
|
||||
import { createRuntimeHost } from '../../src/lib/canvas-runtime/host.ts';
|
||||
import { loadQueryKey } from '../../src/lib/canvas-runtime/loader.ts';
|
||||
import { createFoundryModelLoader } from '../../src/lib/canvas-service/model-loader.ts';
|
||||
import { loadCanvasEntry } from '../../src/lib/canvas-entry.ts';
|
||||
const id = n => `10000000-0000-4000-8000-${String(n).padStart(12, '0')}`;
|
||||
const account = id(1), workspace = id(2), root = id(3), registry = createRegistry(builtinComponents());
|
||||
const scope = { accountId: account, workspaceId: workspace, projectId: null, canvasId: root, authEpoch: 1, navigationGeneration: 7, format: 'canonical-model-v1', source: 'foundry', viewKey: 'canonical-project', queryFingerprint: 'full-project' };
|
||||
const deferred = () => { let resolve; const promise = new Promise(yes => { resolve = yes; }); return { promise, resolve }; };
|
||||
const tick = () => new Promise(resolve => setImmediate(resolve));
|
||||
const reply = (version = 1, title = account) => ({ status: 200, body: { model: { documentFormatVersion: 1, projectId: root, complete: true, canvases: [{ id: root, parent: null }], elements: [{ id: 'node', incarnation: 1, ownerCanvasId: root, componentId: 'text', dataVersion: 1, structure: 'node', data: { title, text: '' }, geometry: { x: 0, y: 0, width: 100, height: 80, rotation: 0 } }], tombstones: {}, versions: { [root]: version }, removedCanvasIds: [] } } });
|
||||
|
||||
test('Foundry adapter emits one validated complete chunk with bound query, project, coverage and version metadata', async () => {
|
||||
const abort = new AbortController(), calls = [];
|
||||
const port = createFoundryModelLoader({ workspaceId: workspace, registry, transport: async input => { calls.push(input); return reply(3); } });
|
||||
const chunks = []; for await (const chunk of port(scope, abort.signal)) chunks.push(chunk);
|
||||
assert.equal(chunks.length, 1); assert.equal(chunks[0].complete, true); assert.equal(chunks[0].queryKey, loadQueryKey(scope)); assert.equal(chunks[0].projectId, root);
|
||||
assert.deepEqual(chunks[0].coverage, [`canvas:${root}`]); assert.equal(chunks[0].serverVersions[root], 3); assert.ok(chunks[0].byteLength > 0); assert.equal(chunks[0].value.document.elements.get('node').data.title, account);
|
||||
assert.strictEqual(calls[0].signal, abort.signal); assert.equal(calls[0].method, 'GET');
|
||||
});
|
||||
test('partial canonical replies, foreign requested canvases and unbound migration headers cannot masquerade as initial readiness', async () => {
|
||||
for (const [response, expected] of [[{ ...reply(), body: { model: { ...reply().body.model, complete: false } } }, 'incomplete_initial_model'], [reply(), 'project_mismatch'], [{ status: 409, body: { error: 'canonical_migration_required', document_format_version: 1, root_id: root, source_versions: { [root]: 1 } } }, 'project_mismatch']]) {
|
||||
const port = createFoundryModelLoader({ workspaceId: workspace, registry, transport: async () => response }), query = expected === 'incomplete_initial_model' ? scope : { ...scope, canvasId: id(99) };
|
||||
await assert.rejects(async () => { for await (const _ of port(query, new AbortController().signal)) assert.fail('invalid response yielded a chunk'); }, { code: expected });
|
||||
}
|
||||
});
|
||||
test('Host shares concurrent model reads per identity/workspace but never reuses an idle authoritative snapshot', async () => {
|
||||
const gate = deferred(), calls = []; let factories = 0, version = 1;
|
||||
const host = createRuntimeHost({ registry, workspace() { assert.fail('readonly load opened a project/storage runtime'); }, modelLoader(accountId, workspaceId, _epoch, current) {
|
||||
factories++; assert.equal(accountId, account); return createFoundryModelLoader({ workspaceId, registry, transport: async input => { calls.push(input); if (calls.length === 1) await gate.promise; assert.equal(current(), true); return reply(version); } });
|
||||
} });
|
||||
try {
|
||||
const abort = new AbortController(), a = host.loadModel(account, workspace, root, { signal: abort.signal, navigationGeneration: 1 }), b = host.loadModel(account, workspace, root, { navigationGeneration: 2 });
|
||||
await tick(); const rejected = assert.rejects(a, { code: 'load_aborted' }); abort.abort(); await rejected; assert.equal(calls[0].signal.aborted, false);
|
||||
gate.resolve(); assert.equal((await b).versions.get(root), 1); assert.equal(calls.length, 1); assert.equal(factories, 1);
|
||||
version = 2; assert.equal((await host.loadModel(account, workspace, root)).versions.get(root), 2); assert.equal(calls.length, 2); assert.equal(host.getWorkspaces().size, 0);
|
||||
} finally { gate.resolve(); await host.close(); }
|
||||
});
|
||||
test('Host retirement aborts the old read and a late response cannot pollute the new account reader', async () => {
|
||||
const gate = deferred(), calls = [];
|
||||
const host = createRuntimeHost({ registry, workspace() { assert.fail('unexpected project open'); }, modelLoader(accountId, workspaceId) {
|
||||
return createFoundryModelLoader({ workspaceId, registry, transport: async input => { calls.push({ accountId, input }); if (accountId === account) await gate.promise; return reply(1, accountId); } });
|
||||
} });
|
||||
try {
|
||||
const old = host.loadModel(account, workspace, root); await tick(); const rejected = assert.rejects(old, { code: 'loader_closed' }); await host.retireIdentity(); await rejected;
|
||||
const next = await host.loadModel(id(8), workspace, root); assert.equal(next.document.elements.get('node').data.title, id(8));
|
||||
gate.resolve(); await tick(); assert.equal(host.identity().accountId, id(8)); assert.equal(calls[0].input.signal.aborted, true); assert.equal(calls.length, 2);
|
||||
} finally { gate.resolve(); await host.close(); }
|
||||
});
|
||||
test('entry injection uses resolved IDs and navigation context while preserving canonical-first legacy policy', async () => {
|
||||
const board = { id: root, workspace_id: workspace, slug: 'board', kind: 'board', title: 'Board', version: 1 }, inputs = [], reads = [];
|
||||
const options = { workspaceRef: 'workspace-slug', boardRef: 'board', registry, navigationGeneration: 9, request: async path => { reads.push(path); return path === 'profile' ? { id: account } : board; },
|
||||
readModel: async input => { inputs.push(input); const port = createFoundryModelLoader({ workspaceId: workspace, registry, transport: async () => reply() }); for await (const chunk of port(scope, new AbortController().signal)) return chunk.value; }, transport() { assert.fail('entry bypassed the installed loader'); } };
|
||||
const result = await loadCanvasEntry(options); assert.equal(result.mode, 'headless'); assert.deepEqual(inputs[0], { accountId: account, workspaceId: workspace, canvasId: root, signal: undefined, navigationGeneration: 9 }); assert.equal(reads.some(path => path.endsWith('/canvas')), false);
|
||||
await assert.rejects(() => loadCanvasEntry({ ...options, readModel: async () => { throw Object.assign(Error('forbidden'), { code: 'forbidden' }); } }), { code: 'forbidden' });
|
||||
});
|
||||
Reference in New Issue
Block a user