diff --git a/src/lib/canvas-runtime/loader.ts b/src/lib/canvas-runtime/loader.ts index 50924d7..3948c22 100644 --- a/src/lib/canvas-runtime/loader.ts +++ b/src/lib/canvas-runtime/loader.ts @@ -19,7 +19,8 @@ export type LoadChunk = { readonly sequence: number; readonly coverage: readonly string[]; readonly complete: boolean; readonly byteLength: number; readonly value: T; }; -/** 加载状态及已收到块;initial-ready/partial 只说明有内容,ready Promise 要到 terminal+EOF 才完成。 */ +/** 加载状态及已收到块;initial-ready 表示收到首个合法块,可能只是空模型头,并不证明画面已有元素。 + * partial 继续累积前缀;ready Promise 要等 terminal+EOF,失败前缀仍保留在 failed-with-partial 快照中。 */ export type LoadSnapshot = { readonly scope: LoadScope; readonly generation: number; readonly cacheStatus: 'bypassed' | 'miss' | 'invalid' | 'hit'; @@ -44,7 +45,10 @@ export function loadQueryKey(scope: LoadScope): string { /** Read copies only. This coordinator never publishes a document or a server ACK. * Adapters validate payloads and declare their measured byte cost before yielding. */ -/** 装配有界加载池;默认缓存年龄0不保留空闲结果,启用缓存也要求宿主 acceptCached 明确同意。 */ +/** 建立共享 query 调度器:inflight 去重、queue 按 generation FIFO、active 计真实未退出源,cache 按 Map LRU。 + * 默认并发 4/等待 128、单响应声明 64MiB/4096 块;上限依赖 adapter 准确测量 byteLength,不测 T 或内存峰值。 + * 默认缓存年龄 0 不插入终态缓存;启用后仍须订阅 allow-validated 且 acceptCached 同意,不把缓存当授权事实。 + * 不重试、无读取超时,不深克隆/解释 T;current 应表达共享查询的授权生命周期,导航取消由各订阅管理。 */ export function createScopedLoader(options: { load: LoadPort; current(scope: LoadScope): boolean; concurrency?: number; maxQueued?: number; maxResponseBytes?: number; maxChunks?: number; maxCacheEntries?: number; maxCacheBytes?: number; @@ -54,7 +58,8 @@ export function createScopedLoader(options: { }) { /** 订阅自己的 scope/取消/通知状态,不把旧导航的取消写成整个 job 失败。 */ type Subscriber = { readonly scope: LoadScope; active: boolean; cancelled?: LoadSnapshot; lastState?: LoadSnapshot; snapshot?: LoadSnapshot;resolve(value: LoadSnapshot): void; reject(error: unknown): void; cleanup(): void; notify?(snapshot: LoadSnapshot): void }; - /** 共享 query 的执行句柄;generation 防重开混淆,controller 只在最后订阅释放时 abort。 */ + /** 共享执行句柄,scope 来自首个订阅;view 会将导航信息换成各订阅自己的 scope。 + * generation 区分重开,controller 在 retire(最后释放或 close)时 abort,不以订阅数为持久任务事实。 */ type Job = { readonly key: string; readonly scope: LoadScope; readonly generation: number; readonly controller: AbortController; readonly subscribers: Set; state: LoadSnapshot; 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; @@ -74,7 +79,8 @@ export function createScopedLoader(options: { 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); } }; - /** 幂等废弃 job:abort/出等待队列,仅当 inflight 仍指向自己时删除,旧任务不能删新代次。 */ + /** 幂等 retire:abort、移除排队/仍指向自己的 inflight,不删除缓存、不清前缀或提早扣 active。 + * 成功/失败后仍保留的句柄可独立 release;本函数不把已兑现的 ready Promise 改成拒绝。 */ 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); @@ -88,7 +94,9 @@ export function createScopedLoader(options: { void (async () => { try { current(job.scope); let terminal = false, projectId: Id | undefined, token: string | undefined, versions: Readonly> | undefined; - // 每块重新检查身份并核对连续序号、稳定 token/项目/版本向量、权限代次和声明预算。 + // 端口可收到已取消信号(例如 loading 通知期间最后订阅释放);它仍须在实际 I/O 前检查 signal。 + // 每块重新检查身份,核对连续序号、稳定 token/项目/版本、权限代次和声明预算;coverage 只是合法字符串集合。 + // complete/coverage 的领域含义、实际请求范围和 T 的 schema 由 adapter 负责,本层不从它们推断文档闭合。 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 @@ -103,7 +111,8 @@ export function createScopedLoader(options: { 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. + // 捕获标准元数据表/coverage,value 与其它嵌套扩展字段仍沿用端口引用,不在此深冻结任意 T。 + // adapter 应在 yield 前完成 payload 校验并交出不可变值;外层 Object.freeze 不证明内部所有权。 const owned = Object.freeze({ ...chunk, serverVersions: freeze(orderedVersions), coverage: freeze([...chunk.coverage]) }); // 只追加合法块并通知可显示前缀;即使当前块 complete=true,也要等源 EOF 才能成功 ready。 job.state = Object.freeze({ ...job.state, status: job.state.chunks.length ? 'partial' : 'initial-ready', chunks: Object.freeze([...job.state.chunks, owned]), @@ -116,6 +125,8 @@ export function createScopedLoader(options: { 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!); } + // 成功清理外部 abort 监听并兑现 ready,但不自动 release 句柄;后续 signal.abort 不再改变其快照。 + // subscriber 仍持有共享状态,调用方应在不再使用时显式 release;已发出的快照继续描述原切点。 notify(job); for (const subscriber of job.subscribers) if (subscriber.active) { subscriber.cleanup(); subscriber.resolve(view(job, subscriber)); } } catch (error) { if (job.retired) return; @@ -132,7 +143,10 @@ export function createScopedLoader(options: { } } return Object.freeze({ - /** 捕获 scope→检查缓存→共享/新建 job→建立独立订阅;本调用不直接打开/关闭 ProjectEngine。 */ + /** 捕获声明 scope 字段,校验身份/取消,再检查缓存并共享或新建 job;不打开 ProjectEngine。 + * fresh 会淘汰该 key 的旧缓存,但仍共享已有 inflight,并不保证为每次调用发新请求。 + * 缓存验证/时钟/容量等错误可同步抛出,成功才返回 ready/getSnapshot/release 句柄。 + * cacheStatus 属于共享 job 建立时的读取事实,复用在途 job 不会按新订阅的缓存偏好重写它。 */ acquire(scopeInput: LoadScope, settings: { signal?: AbortSignal; onUpdate?: (snapshot: LoadSnapshot) => void; cache?: 'fresh' | 'allow-validated' } = {}): LoadHandle { 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}); @@ -158,13 +172,19 @@ export function createScopedLoader(options: { } const shared = job; let subscriber: Subscriber; const ready = new Promise>((resolve, reject) => { subscriber = { scope, active: true, resolve, reject, notify: settings.onUpdate, cleanup: () => settings.signal?.removeEventListener('abort', release) }; }); - /** 冻结本订阅 cancelled 视图、清监听并拒绝等待;只有最后借用者离开才 retire 共享源。 */ + /** 为本订阅记录 cancelled 快照、移除监听/借用,尝试以 load_aborted 拒绝尚未兑现的 ready。 + * ready 已成功/失败时 Promise 结果不变;getSnapshot 则改读此 cancelled 切点,保留已有 chunks。 + * 最后借用者才 retire 共享源;取消一个订阅不会撤回其它订阅的前缀或删除成功缓存。 */ 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 }); + // 缓存命中的句柄不登记进 inflight,ready 立即兑现,外部 abort 监听仍保留到 release。 + // 因此 close 负责清共享缓存,却不追溯取消这种已返回句柄;调用方仍拥有其只读快照。 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 }); }, - /** 永久关闭并拒绝所有活订阅,清等待/缓存;底层迟到结果由 retired/current 围栏拒绝采纳。 */ + /** 永久关闭:只对当前 inflight 的订阅记录 cancelled 并以 loader_closed 拒绝,随后清 queue/cache。 + * 不追溯修改已从 inflight 退出的终态句柄或缓存命中句柄;其旧快照仍可读,显式 release 仍可调用。 + * 不等待底层迭代器退出,active 在各 finally 才归还;无视 abort 的源仍可能占着资源,但迟到块不被采纳。 */ 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(); }