docs(canvas): explain streamed loading and certified scope admission

This commit is contained in:
2026-11-12 00:25:26 +08:00
parent b471d88520
commit 68e655b52c
6 changed files with 138 additions and 0 deletions
+35
View File
@@ -1,17 +1,25 @@
/**
* 作用域化只读加载调度器:同 query identity 合并进行中的请求,每个导航订阅仍有独立取消/快照。
* queue 控制等待、active 限并发、cache 可选且必须额外验证;不拥有 working/history/ACK,也不直接发布业务文档。
* 端口必须验证并提供不可变 payload 和测量 byteLength,这里校验捕获元数据、连续块与预算,不猜任意 T 的结构。
*/
import type { Id } from '../canvas-core/types.ts';
import { equal, fault, freeze } from '../canvas-core/values.ts';
/** 账号/工作区/项目/画布/查询/格式/来源/authEpoch 决定共享身份,navigationGeneration 属于订阅者。 */
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;
};
/** 一次捕获的连续块:queryKey/token/版本/权限代次固定,coverage 由协议适配器解释。 */
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;
};
/** 加载状态及已收到块;initial-ready/partial 只说明有内容,ready Promise 要到 terminal+EOF 才完成。 */
export type LoadSnapshot<T> = {
readonly scope: LoadScope; readonly generation: number;
readonly cacheStatus: 'bypassed' | 'miss' | 'invalid' | 'hit';
@@ -19,10 +27,13 @@ export type LoadSnapshot<T> = {
readonly chunks: readonly LoadChunk<T>[]; readonly coverage: readonly string[]; readonly byteLength: number;
readonly error?: { readonly code: string; readonly message: string };
};
/** 可取消的异步只读块源,负责实际 HTTP/解码与 payload 校验,不能在这里偷偷写项目状态。 */
export type LoadPort<T> = (scope: LoadScope, signal: AbortSignal) => AsyncIterable<LoadChunk<T>>;
/** 单订阅句柄:ready 等完整读取,getSnapshot 可看合法前缀,release 仅解除自己的借用。 */
export type LoadHandle<T> = { readonly ready: Promise<LoadSnapshot<T>>; getSnapshot(): LoadSnapshot<T>; release(): void };
/** Navigation belongs to a subscriber, not the shared query identity. */
/** 验证 scope 并稳定编码共享 key;刻意排除导航代次,让两个 view 可复用同一读取。 */
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)
@@ -33,6 +44,7 @@ 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 明确同意。 */
export function createScopedLoader<T>(options: {
load: LoadPort<T>; current(scope: LoadScope): boolean; concurrency?: number; maxQueued?: number;
maxResponseBytes?: number; maxChunks?: number; maxCacheEntries?: number; maxCacheBytes?: number;
@@ -40,23 +52,34 @@ export function createScopedLoader<T>(options: {
acceptCached?(scope: LoadScope, snapshot: LoadSnapshot<T>): boolean;
onObserverError?(error: unknown): void;
}) {
/** 订阅自己的 scope/取消/通知状态,不把旧导航的取消写成整个 job 失败。 */
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 };
/** 共享 query 的执行句柄;generation 防重开混淆,controller 只在最后订阅释放时 abort。 */
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');
// inflight 按 query 去重,queue 按 generation FIFO;cache 的 Map 插入顺序用于 LRU 淘汰。
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; };
/** 每个异步采纳边界检查 loader 生命周期与精确身份,不用 HTTP 成功替代授权仍有效。 */
const current = (scope: LoadScope) => { if (closed) throw fault('loader_closed'); if (!options.current(scope)) throw fault('identity_changed'); };
/** 共享 job 状态不变时复用该订阅快照;scope 替换为订阅者自己的导航信息。 */
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!;};
/** 单独隔离 observer 异常,报告失败也不改变共享执行和其它订阅者的通知。 */
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); } };
/** 幂等废弃 job:abort/出等待队列,仅当 inflight 仍指向自己时删除,旧任务不能删新代次。 */
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);
}
/** FIFO 启动有订阅的工作并登记真实 active;即使源无视 abort,只有它 finally 结束才归还槽位。 */
function pump() {
while (!closed && active < concurrency && queue.size) {
const job = queue.values().next().value!; queue.delete(job.generation);
@@ -65,6 +88,7 @@ export function createScopedLoader<T>(options: {
void (async () => {
try {
current(job.scope); let terminal = false, projectId: Id | undefined, token: string | undefined, versions: Readonly<Record<Id, number>> | undefined;
// 每块重新检查身份并核对连续序号、稳定 token/项目/版本向量、权限代次和声明预算。
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
@@ -74,17 +98,20 @@ export function createScopedLoader<T>(options: {
|| !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]) });
// 只追加合法块并通知可显示前缀;即使当前块 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]),
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' });
// 完整结果才进缓存,且三项缓存开关都开启;按条数和测量字节双重 LRU 上限淘汰。
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!);
@@ -92,10 +119,12 @@ export function createScopedLoader<T>(options: {
notify(job); for (const subscriber of job.subscribers) if (subscriber.active) { subscriber.cleanup(); subscriber.resolve(view(job, subscriber)); }
} catch (error) {
if (job.retired) return;
// 尾部失败仍保留已校验 chunks,状态 failed-with-partial;不伪造 complete,也不清掉已显示前缀。
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 {
// 迟到旧源结束时只清自己,不能误删相同 key 新订阅创建的 job;随后归还 active 并泵队列。
if (inflight.get(job.key) === job) inflight.delete(job.key);
active--; pump();
}
@@ -103,12 +132,14 @@ export function createScopedLoader<T>(options: {
}
}
return Object.freeze({
/** 捕获 scope→检查缓存→共享/新建 job→建立独立订阅;本调用不直接打开/关闭 ProjectEngine。 */
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);
// 缓存需 allow-validated、年龄有效且 acceptCached===true;时钟倒退或 fresh 都使旧缓存失效。
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; }
@@ -116,6 +147,7 @@ export function createScopedLoader<T>(options: {
}
current(scope);
if(settings.signal?.aborted)throw fault('load_aborted');
// 已有 inflight 优先共享;新 job 的容量包含可立即开跑的空槽,generation 溢出明确拒绝。
let job = inflight.get(key);
if (!job) {
if (!cached && queue.size >= maxQueued + Math.max(0, concurrency - active)) throw fault('load_capacity');
@@ -126,11 +158,13 @@ export function createScopedLoader<T>(options: {
}
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) }; });
/** 冻结本订阅 cancelled 视图、清监听并拒绝等待;只有最后借用者离开才 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 });
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 围栏拒绝采纳。 */
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(); }
@@ -138,4 +172,5 @@ export function createScopedLoader<T>(options: {
},
});
}
/** 公共泛型读取器 API 由工厂推导,加载结果仍需协议层转换为具体 canonical 数据。 */
export type ScopedLoader<T> = ReturnType<typeof createScopedLoader<T>>;
+12
View File
@@ -1,3 +1,7 @@
/**
* 部分加载v6的上传/跨项目资源采纳准入证据。资源存在或URL相同不代表当前源可安全使用,
* 必须绑定 sourceToken、认证canvas范围和对应逻辑操作;原上传body/key也要与资源事实一致。
*/
import type {Document,Id,Registry,OperationIntent} from '../canvas-core/types.ts';
import {equal,fault} from '../canvas-core/values.ts';
import type {RuntimeState} from './contracts.ts';
@@ -5,7 +9,9 @@ import {structuralRequestScope} from './loading.ts';
import {referencedResources,type LoadingAssetAdmission} from './resources.ts';
/** Proven local upload identities; never infer authorization from an arbitrary URL. */
// source身份+各proof身份一致时复用资源URL表;不因浏览器缓存有URL就授予准入。
const namespaces=new WeakMap<object,{proofs:Map<Id,LoadingAssetAdmission>;urls:ReadonlyMap<Id,string>}>();
/** 从已记录loadingAdmission派生resource→规范来源URL,只在v6和保留源存在时提供。 */
export function loadingAssetURLs(runtime:RuntimeState):ReadonlyMap<Id,string>{
const source=runtime.loadingRead;if(!source||runtime.loading?.version!==6)return new Map();
const proofs=new Map<Id,LoadingAssetAdmission>();for(const resource of runtime.resources.values())if(resource.loadingAdmission)proofs.set(resource.id,resource.loadingAdmission);
@@ -13,19 +19,23 @@ export function loadingAssetURLs(runtime:RuntimeState):ReadonlyMap<Id,string>{
const previous=namespaces.get(source);if(previous&&previous.proofs.size===proofs.size&&[...proofs].every(([id,proof])=>previous.proofs.get(id)===proof))return previous.urls;
const urls=new Map([...proofs].map(([id,proof])=>[id,proof.sourceURL]));namespaces.set(source,{proofs,urls});return urls;
}
/** 联合验证上传/采纳与源范围、操作顺序、引用和stored证明;最后核对冻结上传协议字段。 */
export function validateLoadingAssetAdmissions(runtime:RuntimeState,registry:Registry):void{
const loading=runtime.loading;if(loading?.version!==6)return;
// upload先分配资源,后续document.change才能引用;早于/同于upload的引用不具备可靠身份因果。
for(const resource of runtime.resources.values()){
const proof=resource.loadingAdmission,record=runtime.ledger.operations.get(resource.operationId);
if(!proof||proof.version!==1||proof.sourceToken!==loading.sourceToken||typeof proof.sourceURL!=='string'||!proof.sourceURL||proof.sourceURL.length>8192||!Array.isArray(proof.canvasIds)||!proof.canvasIds.length||new Set(proof.canvasIds).size!==proof.canvasIds.length||proof.canvasIds.some(id=>typeof id!=='string'||!structuralRequestScope(runtime,runtime.loadingRead!,id))||record?.intent.operationType!=='asset.upload'||record.intent.resourceId!==resource.id||resource.stored&&record.resolution!=='applied')throw fault('invalid_loading_asset',[resource.id]);
if([...runtime.ledger.operations.values()].some(candidate=>candidate.intent.operationType==='document.change'&&candidate.intent.sequence<=record.intent.sequence&&referencedResources(candidate.intent.changes,registry).has(resource.id)))throw fault('invalid_loading_asset',[resource.id]);
if(![...runtime.ledger.operations.values()].some(candidate=>candidate.intent.operationType==='document.change'&&candidate.intent.sequence>record.intent.sequence&&referencedResources(candidate.intent.changes,registry).has(resource.id)))throw fault('invalid_loading_asset',[resource.id]);
}
// adoption绑定自己的文档意图,sourceURL等于已授权asset.url,不凭后来的另一个引用补造准入。
for(const [id,binding]of runtime.assetBindings??[]){
const proof=binding.loadingAdmission,record=runtime.ledger.operations.get(binding.operationId);
if(!proof||proof.version!==1||proof.operationId!==binding.operationId||proof.sourceToken!==loading.sourceToken||proof.sourceURL!==binding.asset.url||!Array.isArray(proof.canvasIds)||!proof.canvasIds.length||new Set(proof.canvasIds).size!==proof.canvasIds.length||proof.canvasIds.some(owner=>!structuralRequestScope(runtime,runtime.loadingRead!,owner))||record?.intent.operationType!=='document.change'||!referencedResources(record.intent.changes,registry).has(id))throw fault('invalid_loading_asset',[id]);
if([...runtime.ledger.operations.values()].some(candidate=>candidate.intent.operationType==='document.change'&&candidate.intent.sequence<record.intent.sequence&&referencedResources(candidate.intent.changes,registry).has(id)))throw fault('invalid_loading_asset',[id]);
}
// 冻结请求必须是原资源/operation/digest/长度/MIME,且不伪造canvas版本参与者。
for(const request of runtime.ledger.requests.values())if(request.kind==='asset.upload'){
const intent=request.operationIds.length===1?runtime.ledger.operations.get(request.operationIds[0])?.intent:undefined,resource=intent?.operationType==='asset.upload'?runtime.resources.get(intent.resourceId):undefined;
if(!resource?.digest||request.protocolVersion!==2||request.canvasIds.length||Object.keys(request.expectedVersions).length)throw fault('invalid_loading_request');
@@ -34,7 +44,9 @@ export function validateLoadingAssetAdmissions(runtime:RuntimeState,registry:Reg
}
}
// 每registry+文档身份缓存资源的实际owner集合,完整比较首用范围时避免重复全表扫描。
const ownership=new WeakMap<Registry,WeakMap<Document,Map<Id,readonly Id[]>>>();
/** 资源首次出现在逻辑回放中时,实际owner集合须精确匹配准入范围;seen只在本次验证中去重。 */
export function validateLoadingAssetUse(intent:OperationIntent,document:Document,runtime:RuntimeState,registry:Registry,seen:Set<Id>):void{
if(runtime.loading?.version!==6)return;
for(const id of referencedResources(intent.changes,registry)){
+15
View File
@@ -1,3 +1,7 @@
/**
* 部分加载v5/v6允许闭合范围内导入/复制的准入与恢复校验。
* 源token、开始时逻辑sequence和必要canvas集合一起捕获,避免后来的未读树或新ACK改变原复制源含义。
*/
import type {CanvasRecord,Document,Id,Target} from '../canvas-core/types.ts';
import {fault,freeze,equal} from '../canvas-core/values.ts';
import type {EngineSnapshot} from './contracts.ts';
@@ -7,18 +11,24 @@ import {copyTopology} from './copy-snapshot-index.ts';
import {structuralRequestScope} from './loading.ts';
import {ownerReady} from './loading-structure.ts';
// 不可变文档的owner成员/入口子canvas关系索引,复用扫描结果但不另存第二份文档。
const relations=new WeakMap<Document,{members:Map<Id,Id[]>;children:Map<Id,Id[]>}>();
/** 构造直接成员和entry→子canvas映射,供选区DFS扩张使用,不从节点可见性推断拥有范围。 */
function index(document:Document){let value=relations.get(document);if(!value){const members=new Map<Id,Id[]>(),children=new Map<Id,Id[]>();for(const element of document.elements.values()){const ids=members.get(element.ownerCanvasId)??[];ids.push(element.id);members.set(element.ownerCanvasId,ids);}for(const canvas of document.canvases.values())if(canvas.parent){const ids=children.get(canvas.parent.entryElementId)??[];ids.push(canvas.id);children.set(canvas.parent.entryElementId,ids);}value={members,children};relations.set(document,value);}return value;}
/** The caller captures this proof with the same document that becomes source bytes. */
/** target先闭合;复制还要求全部选中后代/必要祖先/上级入边owner完整,然后冻结准入范围与sequence。 */
export function loadingImportAdmission(snapshot:EngineSnapshot,target:Id,selection?:readonly Target[]):LoadingImportAdmission|undefined {
const state=snapshot.runtime.loading;if(!state)return;
if((state.version!==5&&state.version!==6)||!snapshot.runtime.loadingRead)throw fault('model_loading');
const source=snapshot.runtime.loadingRead,document=snapshot.document,required=new Set<Id>([target]);
/** 每个必要canvas既加入证明,也必须当前ownerReady,不能只记名字就声称已经加载完整。 */
const requireScope=(id:Id)=>{required.add(id);if(!ownerReady(state,source,document,id))throw fault('loading_scope_incomplete',[id]);};requireScope(target);
if(selection){
if(!selection.length)throw fault('empty_copy_selection');const tables=index(document),pending=selection.map(target=>{const element=document.elements.get(target.id);if(!element||element.incarnation!==target.incarnation)throw fault('stale_incarnation',[target.id]);return target.id;}),copied=new Set<Id>(),canvases=new Map<Id,CanvasRecord>(document.canvases);
// Missing captured children cannot be mistaken for an empty/leaf selection.
// 完整sourceCanvases揭示未收到的子分区,不能把尚未到齐的F当成叶子,误复制空分区。
const missingChildren=new Map<Id,Id[]>();for(const canvas of state.sourceCanvases)if(canvas.parent&&!source.document.canvases.has(canvas.id)&&!document.canvases.has(canvas.id)){canvases.set(canvas.id,canvas);const ids=missingChildren.get(canvas.parent.entryElementId)??[];ids.push(canvas.id);missingChildren.set(canvas.parent.entryElementId,ids);}
// 显式stack去重深复制选区及全部后代,对每层直属owner检查闭合。
while(pending.length){const id=pending.pop()!;if(copied.has(id))continue;copied.add(id);const element=document.elements.get(id);if(!element)throw fault('loading_scope_incomplete',[id]);requireScope(element.ownerCanvasId);
for(const child of [...(tables.children.get(id)??[]),...(missingChildren.get(id)??[])]){requireScope(child);pending.push(...(tables.members.get(child)??[]));}
}
@@ -28,9 +38,11 @@ export function loadingImportAdmission(snapshot:EngineSnapshot,target:Id,selecti
// Edges involving born objects are entirely local to this captured namespace.
for(const element of document.elements.values())if(element.structure==='connection'&&copied.has(element.target.elementId)&&!copied.has(element.source.elementId))requireScope(element.ownerCanvasId);
}
// 用当前最大逻辑sequence固定复制源时点,后续用户编辑不应倒灌成原源内容。
let operationSequence=0;for(const record of snapshot.runtime.ledger.operations.values())operationSequence=Math.max(operationSequence,record.intent.sequence);
return freeze({version:1 as const,sourceToken:state.sourceToken,operationSequence,requiredCanvasIds:[...required].sort()});
}
/** 恢复时job须同项目/同token,证明sequence不超账本,target与全部必要canvas仍是可证明范围。 */
export function validateLoadingImportAdmissions(runtime:EngineSnapshot['runtime']):void {
const state=runtime.loading;if(!state||(state.version!==5&&state.version!==6))return;
let maxSequence=0;for(const record of runtime.ledger.operations.values())maxSequence=Math.max(maxSequence,record.intent.sequence);
@@ -39,13 +51,16 @@ export function validateLoadingImportAdmissions(runtime:EngineSnapshot['runtime'
/** Revalidate the frozen copy topology before planning or resuming its source.
* Later source chunks may add unrelated owners, so compare only the closed cut. */
/** 恢复规划前按准入sequence重建预期源,只比较原闭合cut的topology,不要求后到无关owner保持不增长。 */
export function validateLoadingCopySource(snapshot:EngineSnapshot,job:ImportJob,document:Document):void {
const state=snapshot.runtime.loading,proof=job.source?.loadingAdmission;if(!state||(state.version!==5&&state.version!==6)||!proof)return;
const source=snapshot.runtime.loadingRead!;let expected=source.document;
// 原captured源叠当时已有的逻辑变化重建源时点,不把当前working当成冻结复制快照。
for(const record of [...snapshot.runtime.ledger.operations.values()].sort((a,b)=>a.intent.sequence-b.intent.sequence))if(record.intent.sequence<=proof.operationSequence)expected=applyChanges(expected,record.intent.changes,false);
const selection=job.preparation?.selection??job.source?.plan?.selection;if(!selection)throw fault('invalid_loading_import',[job.id]);
const required=loadingImportAdmission({...snapshot,document:expected},job.source!.targetCanvasId,selection)!;
if(!equal(required.requiredCanvasIds,proof.requiredCanvasIds))throw fault('loading_copy_scope_changed',[job.id]);
// 逐owner比较canvas与copyTopology;普通payload并非这里完整性证明,源字节/摘要由其它导入层校验。
for(const id of proof.requiredCanvasIds){if(!equal(expected.canvases.get(id),document.canvases.get(id)))throw fault('loading_copy_scope_changed',[id]);
const wanted=Object.fromEntries([...expected.elements.values()].filter(element=>element.ownerCanvasId===id).map(element=>[element.id,copyTopology(element)])),actual=Object.fromEntries([...document.elements.values()].filter(element=>element.ownerCanvasId===id).map(element=>[element.id,copyTopology(element)]));if(!equal(wanted,actual))throw fault('loading_copy_scope_changed',[id]);
}
@@ -1,3 +1,7 @@
/**
* 部分加载中的结构修改证明:源命名空间区分 captured 与本地 born,源直属计数证明 owner 已闭合。
* 命令预检与实际 changes 双重校验,保护未收到的后代/引用;已确认结构效果另投影回新读取,避免尾块回退 ACK。
*/
import type {Change,Command,Document,ElementRecord,Id,Registry,Availability,OperationIntent} from '../canvas-core/types.ts';
import {applyChanges} from '../canvas-core/document.ts';
import {copyReferenceIds} from '../canvas-core/copy-paths.ts';
@@ -5,12 +9,19 @@ import {equal,fault} from '../canvas-core/values.ts';
import type {CanonicalProposal,RuntimeState} from './contracts.ts';
import type {LoadingState} from './loading.ts';
/** 同 token 的原 canonical 捕获及其版本向量,生命周期内保持不可变,不是当前 working。 */
type Capture=CanonicalProposal;
// WeakMap 缓存源文档成员数,复用同快照时不必每条命令扫描全表。
const counts=new WeakMap<Document,ReadonlyMap<Id,number>>();
/** 直属owner计数,包括节点与边;不使用DOM或视口可见性。 */
function membership(document:Document){let result=counts.get(document);if(!result){const next=new Map<Id,number>();for(const element of document.elements.values())next.set(element.ownerCanvasId,(next.get(element.ownerCanvasId)??0)+1);result=next;counts.set(document,result);}return result;}
/** 全部声明元素/canvas 均到齐才算源闭合;这里只做数量闭合判断,领域合法性由其它校验共同保证。 */
export function sourceClosed(state:LoadingState,source:Capture){return source.document.elements.size===state.totalElements&&source.document.canvases.size===state.declaredCanvasIds.length;}
/** working中owner存在,且是明确新canvas或源成员数已达头部证明,才能作为闭合操作范围。 */
export function ownerReady(state:LoadingState,source:Capture,document:Document,id:Id){return document.canvases.has(id)&&(!source.versions.has(id)||state.version!==1&&(membership(source.document).get(id)??0)===state.canvasElementCounts[id]);}
/** 只有v4+完整ID命名空间才能证明一个ID不属于原捕获;“当前还没看到”本身不够。 */
function born(state:LoadingState,id:Id){return (state.version===4||state.version===5||state.version===6)&&!Object.hasOwn(state.elementOwners,id);}
/** 禁止凭早期payload引入新任务状态或任意资源URL;新资源须有准入表,已有资源须能在已知数据证明。 */
function safeData(data:ElementRecord['data'],registry:Registry,component:Id,document:Document,source?:Document,prior?:ElementRecord['data'],assets?:ReadonlyMap<Id,string>){
const definition=registry.components.get(component);if(!definition?.copy)return false;
for(const key of ['ai','taskId','runId'])if(data[key]!==undefined&&data[key]!==null&&!equal(data[key],prior?.[key]))return false;
@@ -20,14 +31,18 @@ function safeData(data:ElementRecord['data'],registry:Registry,component:Id,docu
}
return true;
}
/** 最终候选所有已知组件声明引用都必须可解到已有元素/canvas;不猜未知payload中的字符串。 */
function references(document:Document,registry:Registry){for(const element of document.elements.values())if(registry.resolve(element)==='known'){const schema=registry.components.get(element.componentId)?.copy;for(const reference of schema?.references??[])for(const id of copyReferenceIds(element.data,reference))if(!(reference.target==='element'?document.elements:document.canvases).has(id))throw fault('loading_reference_incomplete',[element.id,id]);}}
/** v6受控资源替换的字段分类:允许资源键和安全普通键,不夹带受保护的表现/端口/任务/引用修改。 */
function assetPatch(patch:ElementRecord['data'],registry:Registry,component:Id){
const definition=registry.components.get(component),fields=['src','assetId',...(definition?.resourceFields??[])],keys=Object.keys(patch);
return keys.some(key=>fields.includes(key))&&keys.every(key=>fields.includes(key)||!['presentation','ports','ai','taskId','runId'].includes(key)&&!definition?.copy?.references.some(ref=>ref.path[0]===key||ref.path[0]==='*'));
}
/** 返回结构命令的预检结果,undefined交普通字段门;准入仍需后续核心policy/schema与实际changes检查。 */
export function structuralCommand(command:Command,document:Document,registry:Registry,state:LoadingState,source:Capture,assets?:ReadonlyMap<Id,string>):Availability|undefined {
if(state.version<4)return;
const deny:Availability={enabled:false,code:'model_loading',targets:[]},yes:Availability={enabled:true};
// 本地born对象可在闭合owner创建;captured对象的删除/重挂等需完整源,防未读后代被当作不存在。
if(command.type==='element.create'){const element=command.element;return born(state,element.id)&&!source.document.tombstones.has(element.id)&&ownerReady(state,source,document,element.ownerCanvasId)&&registry.resolve(element)==='known'&&safeData(element.data,registry,element.componentId,document,source.document,undefined,assets)?yes:deny;}
if(command.type==='canvas.create'){const canvas=command.canvas,parent=canvas.parent;return !source.versions.has(canvas.id)&&!!parent&&born(state,parent.entryElementId)&&ownerReady(state,source,document,parent.canvasId)?yes:deny;}
if(command.type==='canvas.reparent')return (!source.versions.has(command.canvasId)||sourceClosed(state,source))&&!!command.parent&&ownerReady(state,source,document,command.parent.canvasId)?yes:deny;
@@ -37,8 +52,10 @@ export function structuralCommand(command:Command,document:Document,registry:Reg
if(command.type==='element.data'&&(born(state,command.target.id)||sourceClosed(state,source)||state.version===6&&assetPatch(command.patch,registry,document.elements.get(command.target.id)?.componentId??'')&&ownerReady(state,source,document,document.elements.get(command.target.id)?.ownerCanvasId??''))){const element=document.elements.get(command.target.id);return element&&registry.resolve(element)==='known'&&safeData({...element.data,...command.patch},registry,element.componentId,document,source.document,element.data,assets)?yes:deny;}
}
/** Actual changes, not a command name, establish the mutation's required closure. */
/** 逐条在私有cursor验证实际变化的必要闭合范围/数据,并检查最终声明引用,防handler副作用绕过。 */
export function assertStructuralChanges(changes:readonly Change[],document:Document,registry:Registry,state:LoadingState,source:Capture,assets?:ReadonlyMap<Id,string>):Document {
if(state.version<4)throw fault('model_loading');
// 按变化顺序推进,同批先创建owner/端点可满足后续依赖,但原源命名空间不会因此被改写。
let cursor=document;const closed=sourceClosed(state,source);
for(const change of changes){
if(change.kind==='canvas'){
@@ -54,6 +71,7 @@ export function assertStructuralChanges(changes:readonly Change[],document:Docum
if(change.field==='geometry'){if(element.structure!=='node')throw fault('model_loading',[element.id]);}
else if(change.field==='path'){if(element.structure!=='connection'||!ownerReady(state,source,cursor,element.ownerCanvasId))throw fault('model_loading',[element.id]);}
else if(['source','target'].includes(change.field)){if(element.structure!=='connection'||!ownerReady(state,source,cursor,element.ownerCanvasId))throw fault('model_loading',[element.id]);}
// 跨owner还要两侧owner都完整;原captured对象在源未齐时不允许盲目移动。
else if(change.field==='ownerCanvasId'){if(!born(state,element.id)&&!closed||!ownerReady(state,source,cursor,element.ownerCanvasId)||!ownerReady(state,source,cursor,change.after as Id))throw fault('model_loading',[element.id]);}
else if(change.field.startsWith('data.')){
const key=change.field.slice(5),definition=registry.components.get(element.componentId),protectedField=['presentation','ports','src','assetId','ai','taskId','runId'].includes(key)||definition?.resourceFields?.includes(key)||definition?.copy?.references.some(ref=>ref.path[0]===key||ref.path[0]==='*');
@@ -63,10 +81,13 @@ export function assertStructuralChanges(changes:readonly Change[],document:Docum
}
cursor=applyChanges(cursor,[change]);
}
// 即使命令名字获准,最终仍不能产生悬空声明引用;整批失败不给调用方半份合法结果。
references(cursor,registry);return cursor;
}
/** Protocol-2 receipts contain exactly these applied logical changes. */
// 缓存同一源上的已确认结构投影,只在applied意图前缀身份未变时从缓存后缀继续。
const projectionCache=new WeakMap<Capture,{intents:readonly OperationIntent[];document:Document}>();
/** 将resolved/applied意图按sequence叠到读取候选,并按精确协议2的参与者版本推进;不混入pending。 */
export function appliedSource(runtime:RuntimeState,source:Capture):Capture {
let document=source.document;const versions=new Map(source.versions);
const applied=[...runtime.ledger.operations.values()].filter(record=>record.phase==='resolved'&&record.resolution==='applied').sort((a,b)=>a.intent.sequence-b.intent.sequence);
@@ -77,7 +98,9 @@ export function appliedSource(runtime:RuntimeState,source:Capture):Capture {
return{document,versions};
}
// source/registry/intent/before/assets全身份绑定校验缓存,不能把另一捕获或不同资源授权的结果拿来复用。
const intentProofs=new WeakMap<Capture,WeakMap<Registry,WeakMap<OperationIntent,{before:Document;after:Document;assets?:ReadonlyMap<Id,string>}>>>();
/** 对同一候选上下文复用已验证结构结果,否则重新assert全部实际changes;不是跳过当前源范围验证。 */
export function validateStructuralIntent(intent:OperationIntent,document:Document,registry:Registry,state:LoadingState,source:Capture,assets?:ReadonlyMap<Id,string>):Document {
let registries=intentProofs.get(source);if(!registries){registries=new WeakMap();intentProofs.set(source,registries);}let intents=registries.get(registry);if(!intents){intents=new WeakMap();registries.set(registry,intents);}const previous=intents.get(intent);if(previous?.before===document&&previous.assets===assets)return previous.after;
const after=assertStructuralChanges(intent.changes,document,registry,state,source,assets);intents.set(intent,{before:document,after,assets});return after;
+34
View File
@@ -1,3 +1,8 @@
/**
* 持久的部分读取证明与编辑/发送门禁。loadingRead 保留原捕获,acknowledged 可随已确认操作推进,
* working 还可含未确认编辑,三者不能混为一份旧源快照。local 可编辑不等于该范围可发送。
* v1/v2 保守暂停外部写;v3 支持已闭合 owner 字段写;v4 加完整ID命名空间,v5 加源树/导入,v6 加资源准入。
*/
import {loadingAssetURLs,validateLoadingAssetAdmissions,validateLoadingAssetUse} from './loading-assets.ts';
import {validateLoadingImportAdmissions} from './loading-import.ts';
import {validateStructuralIntent,structuralCommand,assertStructuralChanges,appliedSource,ownerReady} from './loading-structure.ts';
@@ -9,8 +14,11 @@ import type {CanonicalProposal,RuntimeState} from './contracts.ts';
/** Durable marker for a verified read prefix. Versions 1/2 hold all writes;
* version 3 admits certified scopes with a retained read baseline. No HTTP cursor. */
/** 当前捕获的稳定 token、声明 canvas 范围及接收数量;不保存或伪造可续传 HTTP 字节游标。 */
type LoadingBase={readonly sourceToken:string;readonly declaredCanvasIds:readonly Id[];readonly receivedElements:number;readonly totalElements:number};
/** 版本化范围证据:counts/owners/sourceCanvases 能力逐级增加,缺字段不能当有证明。 */
export type LoadingState=LoadingBase&({readonly version:1;readonly canvasElementCounts?:never;readonly elementOwners?:never;readonly sourceCanvases?:never}|{readonly version:2|3;readonly canvasElementCounts:CanvasElementCounts;readonly elementOwners?:never;readonly sourceCanvases?:never}|{readonly version:4;readonly canvasElementCounts:CanvasElementCounts;readonly elementOwners:ElementOwners;readonly sourceCanvases?:never}|{readonly version:5|6;readonly canvasElementCounts:CanvasElementCounts;readonly elementOwners:ElementOwners;readonly sourceCanvases:readonly CanvasRecord[]});
/** 验证前缀与捕获头一致:版本、树/ID命名空间、总数、直属计数和源版本向量;不是只比较已加载数量。 */
export function validateLoadingState(value:LoadingState,document:Document,versions:ReadonlyMap<Id,number>):void{
if(!value||![1,2,3,4,5,6].includes(value.version))throw fault('loading_interpreter_unavailable');
for(const [id,version]of versions)if(!Number.isSafeInteger(version)||version<1)throw fault('invalid_canonical_version',[id]);
@@ -22,15 +30,19 @@ export function validateLoadingState(value:LoadingState,document:Document,versio
if(value.version===5||value.version===6){validateCanvasTopology(value.sourceCanvases,value.elementOwners,document.projectId);if(!equal(value.sourceCanvases.map(canvas=>canvas.id).sort(),[...value.declaredCanvasIds].sort())||[...document.canvases].some(([id,record])=>!equal(record,value.sourceCanvases.find(canvas=>canvas.id===id))))throw fault('invalid_model_topology');}
if(value.version!==1){validateCanvasElementCounts(value.canvasElementCounts,value.declaredCanvasIds,value.totalElements);const counts=new Map<Id,number>();for(const element of document.elements.values()){const count=(counts.get(element.ownerCanvasId)??0)+1;if(count>value.canvasElementCounts[element.ownerCanvasId])throw fault('invalid_model_coverage');counts.set(element.ownerCanvasId,count);}}
}
/** 统一的部分加载拒绝,可用于解释按钮/命令暂不可用,而不是静默丢操作。 */
const denied:Availability=Object.freeze({enabled:false,code:'model_loading',targets:[]});
/** 普通内容字段可早编辑,引用、资源、表现/端口与任务字段受保护,防止缺依赖时改写结构。 */
function dataKeyAllowed(componentId:Id,key:string,registry:Registry):boolean{
const definition=registry.components.get(componentId);
return !!definition?.copy&&!['src','assetId','presentation','ports','ai','taskId','runId'].includes(key)&&!definition.resourceFields?.includes(key)&&!definition.copy.references.some(ref=>ref.path[0]===key||ref.path[0]==='*');
}
/** 连线路径需已有合法边且其 owner 的源成员全部到齐;不能以当前乐观 working 数量凑完整。 */
function pathAllowed(id:Id,document:Document,registry:Registry,loading?:LoadingState,base=document):boolean{
const element=document.elements.get(id);if((!loading||loading.version===1)||element?.structure!=='connection'||registry.resolve(element)!=='known')return false;
return (directCounts(base).get(element.ownerCanvasId)??0)===loading.canvasElementCounts[element.ownerCanvasId];
}
/** partial 模式的命令预检;新结构证明可处理特定命令,否则仅已有节点几何/安全内容等受限编辑。 */
export function loadingCommand(command:Command,document:Document,registry:Registry,loading?:LoadingState,base=document,source?:CanonicalProposal,assets?:ReadonlyMap<Id,string>):Availability{
if(loading&&(loading.version===4||loading.version===5||loading.version===6)&&source){const scoped=structuralCommand(command,document,registry,loading,source,assets);if(scoped)return scoped;}
if(command.type==='connection.path')return pathAllowed(command.target.id,document,registry,loading,base)?{enabled:true}:denied;
@@ -41,6 +53,7 @@ export function loadingCommand(command:Command,document:Document,registry:Regist
}
/** Check actual changes too: a feature handler cannot evade command admission by
* modifying a protected reference field as a side effect of an allowed edit. */
/** 再核对 handler 实际输出,防止表面是编辑 title,却顺带修改受保护引用字段;v4+走结构闭合校验。 */
export function assertLoadingChanges(changes:readonly Change[],document:Document,registry:Registry,loading?:LoadingState,source?:CanonicalProposal,assets?:ReadonlyMap<Id,string>):void{
if(loading&&(loading.version===4||loading.version===5||loading.version===6)&&source){assertStructuralChanges(changes,document,registry,loading,source,assets);return;}
for(const change of changes){
@@ -49,8 +62,10 @@ export function assertLoadingChanges(changes:readonly Change[],document:Document
if(!element||element.incarnation!==change.target.incarnation||registry.resolve(element)!=='known'||!(change.field==='path'&&pathAllowed(change.target.id,document,registry,loading)||change.field==='geometry'&&element.structure==='node'||change.field.startsWith('data.')&&dataKeyAllowed(element.componentId,change.field.slice(5),registry)))throw fault('model_loading',[change.target.id]);
}
}
/** 恢复/发布时联合校验 loading/source/ACK/意图/资源/导入,不让单表合法但整体证据矛盾的 Root 通过。 */
export function validateLoadingRuntime(runtime:RuntimeState,registry:Registry):void{
if(!runtime.loading){if(runtime.loadingRead)throw fault('invalid_loading_state');return;}
// v3+必须保留独立 loadingRead,旧版本不可偷带该字段;源校验不使用已经推进的 ACK 代替。
const source=runtime.loading.version>=3?runtime.loadingRead:undefined;
if(runtime.loading.version>=3&&!source||runtime.loading.version<3&&runtime.loadingRead)throw fault('invalid_loading_state');
validateLoadingState(runtime.loading,source?.document??runtime.acknowledged,source?.versions??runtime.versions);
@@ -59,6 +74,7 @@ export function validateLoadingRuntime(runtime:RuntimeState,registry:Registry):v
validateLoadingImportAdmissions(runtime);validateLoadingAssetAdmissions(runtime,registry);
const assets=loadingAssetURLs(runtime),assetUses=new Set<Id>();
let logical=source?.document??runtime.acknowledged;
// 按逻辑序列重演允许的变更与资源使用;不能从队列阶段猜已读源的结构。
for(const record of [...runtime.ledger.operations.values()].sort((a,b)=>a.intent.sequence-b.intent.sequence)){
if(record.intent.operationType==='asset.upload'&&runtime.loading.version===6){if(!assets.has(record.intent.resourceId))throw fault('invalid_loading_asset');continue;}
if(record.intent.operationType!=='document.change'||runtime.loading.version<3&&(record.requestId||record.resolution||!['pending','blocked'].includes(record.phase)))throw fault('invalid_loading_state');
@@ -66,12 +82,15 @@ export function validateLoadingRuntime(runtime:RuntimeState,registry:Registry):v
validateLoadingAssetUse(record.intent,logical,runtime,registry,assetUses);
}
}
/** 同一读取的前缀扩展只能追加:token/版本/范围证据/墓碑固定,旧记录不得在后块被修改或移除。 */
export function validatePrefixExtension(previous:Document,previousVersions:ReadonlyMap<Id,number>,state:LoadingState,next:Document,versions:ReadonlyMap<Id,number>,nextState:LoadingState,registry:Registry):void{
validateDocument(next,registry);validateLoadingState(nextState,next,versions);
if(next.projectId!==previous.projectId||state.sourceToken!==nextState.sourceToken||state.version!==nextState.version||!equal(state.canvasElementCounts,nextState.canvasElementCounts)||!equal(state.elementOwners,nextState.elementOwners)||!equal(state.sourceCanvases,nextState.sourceCanvases)||state.totalElements!==nextState.totalElements||nextState.receivedElements<state.receivedElements||!equal(state.declaredCanvasIds,nextState.declaredCanvasIds)||!equal(Object.fromEntries(previousVersions),Object.fromEntries(versions))||!equal(Object.fromEntries(previous.tombstones),Object.fromEntries(next.tombstones))||[...previous.canvases].some(([id,record])=>!equal(record,next.canvases.get(id)))||[...previous.elements].some(([id,record])=>!equal(record,next.elements.get(id))))throw fault('loading_source_mismatch');
}
/** Full fresh reads may include new records in an unchanged-version scope. Known
* records still cannot change without an owner version advancing. */
/** 校验终结或 fresh read:已知版本不倒退,同版本已知记录不变,墓碑不倒退。
* 原捕获补完要求相同 token/向量/总量;fresh read 可新增此前未收到项,v4+按新完整捕获协调。 */
export function validateLoadingCompletion(previous:Document,previousVersions:ReadonlyMap<Id,number>,state:LoadingState,next:Document,versions:ReadonlyMap<Id,number>,registry:Registry,fresh=false,sourceToken?:string):void{
validateDocument(next,registry);if(previous.projectId!==next.projectId)throw fault('project_mismatch');
for(const [id,version]of versions)if(!Number.isSafeInteger(version)||version<1)throw fault('invalid_canonical_version',[id]);
@@ -86,12 +105,19 @@ export function validateLoadingCompletion(previous:Document,previousVersions:Rea
}
/** Counts refer to the read baseline, never to optimistic working membership. */
// 按不可变源文档身份缓存直属成员数,不用 working 中本地新增项冒充源记录已齐。
const membershipCounts=new WeakMap<Document,ReadonlyMap<Id,number>>();
// 已收到/形成合法源前缀的 ID 集与完整 elementOwners 命名空间分开缓存。
const loadedNamespaces=new WeakMap<Document,import('../canvas-core/types.ts').ReadonlyValueSet<Id>>();
/** 当前源前缀的已加载 ID,供查询区分 unloaded 与本地删除后的 absent。 */
function loadedNamespace(document:Document){let ids=loadedNamespaces.get(document);if(!ids){ids=readSet(document.elements.keys());loadedNamespaces.set(document,ids);}return ids;}
// 完整捕获头的 ID→owner 以对象身份绑定,只派生只读成员集合,不另存业务记录。
const namespaceSets=new WeakMap<ElementOwners,import('../canvas-core/types.ts').ReadonlyValueSet<Id>>();
/** 从完整 owners 头取得全部源 ID,包含尚未收到 payload 的元素。 */
function namespace(owners:ElementOwners){let value=namespaceSets.get(owners);if(!value){value=readSet(Object.keys(owners));namespaceSets.set(owners,value);}return value;}
/** 计算并缓存单文档 owner 的直属数量,节点/边都计入,不递归累计子树。 */
function directCounts(document:Document):ReadonlyMap<Id,number>{let counts=membershipCounts.get(document);if(!counts){const next=new Map<Id,number>();for(const element of document.elements.values())next.set(element.ownerCanvasId,(next.get(element.ownerCanvasId)??0)+1);counts=next;membershipCounts.set(document,counts);}return counts;}
/** 给 query 同次发布的完整性:整个项目未完仍可有已认证完整 owner;born canvas 另按结构证明处理。 */
export function loadingQueryCoverage(snapshot:import('./contracts.ts').EngineSnapshot):import('../canvas-core/queries.ts').QueryCoverage {
const loading=snapshot.runtime.loading,base=snapshot.runtime.loadingRead?.document??snapshot.runtime.acknowledged;
if(!loading)return{projectComplete:true,topologyComplete:true,completeCanvasIds:[...snapshot.document.canvases.keys()]};
@@ -101,11 +127,13 @@ export function loadingQueryCoverage(snapshot:import('./contracts.ts').EngineSna
/** Validate the retained source separately from the moving ACK baseline. Only
* field edits in certified scopes may account for differences during loading. */
/** 核对原读取与移动中的 ACK;v3仅允许认证owner字段差异,v4+以已确认逻辑操作投影严格互证。 */
function validateLoadingRead(runtime:RuntimeState,registry:Registry):void {
const source=runtime.loadingRead!,loading=runtime.loading!,base=source.document,ack=runtime.acknowledged;
validateDocument(base,registry);
if(loading.version>=4){validateStructuralRead(runtime,registry);return;}
if(base.projectId!==ack.projectId||!equal(Object.fromEntries(base.canvases),Object.fromEntries(ack.canvases))||!equal(Object.fromEntries(base.tombstones),Object.fromEntries(ack.tombstones))||base.elements.size!==ack.elements.size||source.versions.size!==runtime.versions.size)throw fault('invalid_loading_read');
// 版本推进只能发生在源已闭合的 owner;信封/受保护字段仍不能被普通早编辑改变。
const ready=new Set(loadingQueryCoverage({runtime,document:ack} as import('./contracts.ts').EngineSnapshot).completeCanvasIds);
for(const [id,version]of source.versions){const current=runtime.versions.get(id);if(current===undefined||!Number.isSafeInteger(current)||current<version||current!==version&&!ready.has(id))throw fault('invalid_loading_read',[id]);}
for(const [id,before]of base.elements){const after=ack.elements.get(id);if(!after)throw fault('invalid_loading_read',[id]);if(equal(before,after))continue;
@@ -118,21 +146,26 @@ function validateLoadingRead(runtime:RuntimeState,registry:Registry):void {
for(const request of runtime.ledger.requests.values())if(request.kind!==undefined&&request.kind!=='document.change'||!request.canvasIds.length||new Set(request.canvasIds).size!==request.canvasIds.length||Object.keys(request.expectedVersions).length!==request.canvasIds.length||request.canvasIds.some(id=>!ready.has(id)||!Number.isSafeInteger(request.expectedVersions[id])||request.expectedVersions[id]<(source.versions.get(id)??Infinity)||request.expectedVersions[id]>(runtime.versions.get(id)??-1)))throw fault('invalid_loading_request');
}
/** Overlay only scopes whose ACK version advanced since this immutable capture. */
/** 在新源前缀上只叠已推进 ACK 的效果,不叠未确认 pending;pending 由后续条件重放重建。 */
export function mergeLoadingRead(runtime:RuntimeState,read:CanonicalProposal,registry:Registry):CanonicalProposal {
const source=runtime.loadingRead;if((!runtime.loading||runtime.loading.version<3)||!source)return read;
if(runtime.loading.version>=4){const projected=appliedSource(runtime,read);validateDocument(projected.document,registry);return projected;}
// v3按owner替换已确认范围,保留其它owner新收到的尾块;v4+用 appliedSource 的结构投影。
const advanced=new Set([...runtime.versions].filter(([id,version])=>version>(source.versions.get(id)??Infinity)).map(([id])=>id));
const elements=[...read.document.elements.values()].filter(record=>!advanced.has(record.ownerCanvasId));
for(const element of runtime.acknowledged.elements.values())if(advanced.has(element.ownerCanvasId))elements.push(element);
const document=importDocument(read.document,elements),versions=new Map(read.versions);for(const id of advanced)versions.set(id,runtime.versions.get(id)!);
validateDocument(document,registry);return{document,versions};
}
/** 以指定元素列表重建源文档,保留同一项目/canvas/墓碑,创建标准只读表而非返回可变拼接对象。 */
function importDocument(document:Document,elements:readonly import('../canvas-core/types.ts').ElementRecord[]):Document{return createDocument(document.projectId,[...document.canvases.values()],elements,document.tombstones);}
/** 发送门:v1/v2不发,v3要求全部参与owner闭合,v4+要求每项符合结构范围;无loading正常放行。 */
export function loadingCanSend(runtime:RuntimeState,canvasIds:readonly Id[],document=runtime.acknowledged):boolean {
if(!runtime.loading)return true;if(runtime.loading.version<3)return false;if(runtime.loading.version>=4)return canvasIds.length>0&&canvasIds.every(id=>structuralRequestScope(runtime,runtime.loadingRead!,id));
const ready=new Set(loadingQueryCoverage({runtime,document} as import('./contracts.ts').EngineSnapshot).completeCanvasIds);return canvasIds.length>0&&canvasIds.every(id=>ready.has(id));
}
/** source+已applied changes 必须精确等于 ACK 和版本向量,拒绝旧tail抹去确认后的创建/删除/重挂。 */
function validateStructuralRead(runtime:RuntimeState,registry:Registry){
const source=runtime.loadingRead!,expected=appliedSource(runtime,source),ack=runtime.acknowledged;
if(expected.document.projectId!==ack.projectId||!equal(Object.fromEntries(expected.document.canvases),Object.fromEntries(ack.canvases))||!equal(Object.fromEntries(expected.document.elements),Object.fromEntries(ack.elements))||!equal(Object.fromEntries(expected.document.tombstones),Object.fromEntries(ack.tombstones))||!equal(Object.fromEntries(expected.versions),Object.fromEntries(runtime.versions)))throw fault('invalid_loading_read');
@@ -140,6 +173,7 @@ function validateStructuralRead(runtime:RuntimeState,registry:Registry){
for(const request of runtime.ledger.requests.values())if(!(runtime.loading?.version===6&&request.kind==='asset.upload')&&(request.protocolVersion!==2||!request.canvasIds.length||Object.keys(request.expectedVersions).length!==request.canvasIds.length||request.canvasIds.some(id=>!structuralRequestScope(runtime,source,id)||!Number.isSafeInteger(request.expectedVersions[id])||request.expectedVersions[id]<(source.versions.get(id)??0)||request.expectedVersions[id]>(runtime.versions.get(id)??0))))throw fault('invalid_loading_request');
}
/** 原源owner须完整;不在源版本表的新canvas须有明确创建意图,不能凭任意ID绕过范围门。 */
export function structuralRequestScope(runtime:RuntimeState,source:CanonicalProposal,id:Id):boolean {
if(source.versions.has(id))return ownerReady(runtime.loading!,source,source.document,id);
return [...runtime.ledger.operations.values()].some(record=>record.intent.changes.some(change=>change.kind==='canvas'&&change.id===id&&!change.before&&!!change.after));
+19
View File
@@ -1,3 +1,8 @@
/**
* canonical 流读取的临时只读投影,依据依赖闭合生成可展示的合法文档前缀。
* 它没有 history/ledger/文件/持久恢复权,host/engine 另将前缀接入真正项目状态。
* 终结数据必须与同一捕获的前缀一致;已收到数量不等于已能展示数量,缺依赖的记录继续等待。
*/
import {validateCanvasTopology,validateElementOwners,type ElementOwners,validateCanvasElementCounts,type CanvasElementCounts} from './scope-counts.ts';
import type { CanvasRecord, Document, ElementRecord, Id, Registry } from '../canvas-core/types.ts';
import { ancestors, createDocument, validateDocument, validateElementRecord } from '../canvas-core/document.ts';
@@ -5,23 +10,31 @@ import { copyReferenceIds } from '../canvas-core/copy-paths.ts';
import { clone, equal, fault, freeze, readMap } from '../canvas-core/values.ts';
import type { CanonicalReadUpdate } from './contracts.ts';
/** host 追加版本向量与 loading 证明后的预览,不能把它等同于完整 ACK 基线。 */
export type LiveModelPreview=ModelPreview&{readonly loading:import('./loading.ts').LoadingState;readonly versions:ReadonlyMap<Id,number>};
/** complete 固定false;receivedElements 计已收到,waitingElements 计还因依赖不能发布的记录。 */
export type ModelPreview = { readonly canvasElementCounts?:CanvasElementCounts;readonly elementOwners?:ElementOwners;readonly sourceCanvases?:readonly CanvasRecord[]; readonly document:Document;readonly receivedElements:number;readonly totalElements:number;readonly waitingElements:number;readonly complete:false };
/** An ephemeral read projection. It owns neither a working document nor ACKs,
* history, operations, file bytes or a durable recovery cut. */
/** 建立单捕获投影器;一旦失败/终结就关闭,避免后续块继续污染合法旧预览。 */
export function createModelPreview(registry:Registry){
let elementOwners:ElementOwners|undefined;let canvasElementCounts:CanvasElementCounts|undefined;const ownerCounts=new Map<Id,number>();
let header=false,finished=false,projectId='',expected=0,received=0;
let tombstones:ReadonlyMap<Id,number>=new Map(),published:Document|undefined,snapshot:ModelPreview|undefined;
const canvases=new Map<Id,CanvasRecord>(),records=new Map<Id,ElementRecord>();
/** kind+ID 分命名空间标识依赖,canvas 与 element 即使字符串相同也不混淆。 */
const key=(kind:'canvas'|'element',id:Id)=>JSON.stringify([kind,id]);
// known 只存已经进入合法 published 前缀的键;records 暂存已收到但还未闭合的元素。
const known=new Set<string>();
/** 接收 header/增量/terminal;每批先校验所有记录,再计算新的最大闭合候选集合。 */
function update(value:CanonicalReadUpdate):ModelPreview|undefined{
if(finished)throw fault('model_preview_closed');
// terminal 必须含相同项目/树/墓碑和全部已收到记录,不能用另一份 fresh read 假装补完当前源。
if(value.status!=='receiving'){
if(header&&(value.status!=='ready'||value.document.projectId!==projectId||received!==expected||value.document.canvases.size!==canvases.size||[...canvases].some(([id,canvas])=>!equal(value.document.canvases.get(id),canvas))||!equal(Object.fromEntries(value.document.tombstones),Object.fromEntries(tombstones))||value.document.elements.size!==expected||[...(published?.elements??[]),...records].some(([id,record])=>!equal(value.document.elements.get(id),record))))throw fault('model_preview_changed');
finished=true;return snapshot;
}
// 头部一次性声明 topology/数量/可选完整 ID→owner;后续块不允许重定义这些捕获事实。
if(!header){
if(value.receivedElements!==0||value.elements.length||!value.canvases.length||!Number.isSafeInteger(value.totalElements)||value.totalElements<0)throw fault('invalid_model_preview');
projectId=value.projectId;expected=value.totalElements;tombstones=readMap(new Map(value.tombstones));
@@ -38,12 +51,14 @@ export function createModelPreview(registry:Registry){
}else if(value.projectId!==projectId||value.totalElements!==expected||value.canvases.length||value.tombstones.size||value.canvasElementCounts!==undefined||value.elementOwners!==undefined)throw fault('model_preview_changed');
if(value.receivedElements!==received+value.elements.length||value.receivedElements>expected)throw fault('invalid_model_preview');
// Validate the entire incoming batch before publishing any prefix from it.
// 整批预检ID去重、owner、基础记录和可选命名空间;不先发布批前半段再发现后半段非法。
const incoming=new Set<Id>();
for(const record of value.elements){if(records.has(record.id)||known.has(key('element',record.id))||incoming.has(record.id)||!canvases.has(record.ownerCanvasId))throw fault('invalid_model_preview');validateElementRecord(record,registry,tombstones);if(elementOwners&&(!Object.hasOwn(elementOwners,record.id)||elementOwners[record.id]!==record.ownerCanvasId))throw fault('invalid_model_identity');incoming.add(record.id);}
for(const record of value.elements){const count=(ownerCounts.get(record.ownerCanvasId)??0)+1;if(canvasElementCounts&&count>canvasElementCounts[record.ownerCanvasId])throw fault('invalid_model_coverage');ownerCounts.set(record.ownerCanvasId,count);}
for(const record of value.elements)records.set(record.id,freeze(clone(record)));received=value.receivedElements;
// Greatest closed subset: removing an unavailable dependency also removes its
// dependants. Fully received cycles survive and publish in one legal prefix.
// 候选图包含尚未发布的 canvas/记录,依赖为父/入口/owner/端点及组件明确声明引用。
const candidates=new Map<string,{canvas?:CanvasRecord;element?:ElementRecord;deps:Set<string>}>();
for(const canvas of canvases.values())if(!known.has(key('canvas',canvas.id)))candidates.set(key('canvas',canvas.id),{canvas,deps:new Set(canvas.parent?[key('canvas',canvas.parent.canvasId),key('element',canvas.parent.entryElementId)]:[])});
for(const element of records.values())if(!known.has(key('element',element.id))){
@@ -53,13 +68,17 @@ export function createModelPreview(registry:Registry){
for(const ref of definition?.copy?.references??[])for(const id of copyReferenceIds(element.data,ref))deps.add(key(ref.target,id));
candidates.set(key('element',element.id),{element,deps});
}
// 从“缺少且未收到的依赖”向反向依赖传播排除;完全收到的环可保留并在同一合法前缀发布。
const dependants=new Map<string,Set<string>>(),blocked=new Set<string>(),queue:string[]=[];
for(const [id,item]of candidates)for(const dependency of item.deps){let users=dependants.get(dependency);if(!users){users=new Set();dependants.set(dependency,users);}users.add(id);if(!known.has(dependency)&&!candidates.has(dependency)&&!blocked.has(id)){blocked.add(id);queue.push(id);}}
for(let index=0;index<queue.length;index++)for(const id of dependants.get(queue[index])??[])if(!blocked.has(id)){blocked.add(id);queue.push(id);}
// 在已发布前缀上加全部未阻塞候选,整份文档校验后才更新 published/known。
const nextCanvases=[...(published?.canvases.values()??[])],nextElements=[...(published?.elements.values()??[])],accepted:string[]=[];
for(const [id,item]of candidates)if(!blocked.has(id)){if(item.canvas)nextCanvases.push(item.canvas);if(item.element)nextElements.push(item.element);accepted.push(id);}
if(!published||accepted.length){const next=createDocument(projectId,nextCanvases,nextElements,tombstones);validateDocument(next,registry);published=next;for(const id of accepted){known.add(id);const item=candidates.get(id);if(item?.element)records.delete(item.element.id);}}
// 预览计数始终来自同捕获;v5/v6使用的 sourceCanvases 保留完整源拓扑,不由当前可见树猜。
snapshot=Object.freeze({...(elementOwners?{elementOwners,sourceCanvases:freeze([...canvases.values()])}:{}),...(canvasElementCounts?{canvasElementCounts}:{}),document:published,receivedElements:received,totalElements:expected,waitingElements:received-published.elements.size,complete:false});return snapshot;
}
/** 对外 update 失败即封闭解释器;getSnapshot 只读旧合法前缀,close 清临时表而不删除业务文档。 */
return Object.freeze({update(value:CanonicalReadUpdate){try{return update(value);}catch(error){finished=true;throw error;}},getSnapshot:()=>snapshot,close(){finished=true;records.clear();canvases.clear();known.clear();published=undefined;snapshot=undefined;}});
}