diff --git a/.changeset/cluster-peer-write-invalidates-caches.md b/.changeset/cluster-peer-write-invalidates-caches.md new file mode 100644 index 0000000000..59a8f2c7ed --- /dev/null +++ b/.changeset/cluster-peer-write-invalidates-caches.md @@ -0,0 +1,39 @@ +--- +"@objectstack/metadata": patch +--- + +fix(metadata): 集群对端的元数据写入现在会失效本节点的 `listCache` / registry (#5109) + +多节点部署下,节点 A 改一条 `view` / `permission` / `flow`,节点 B 收到 +`metadata.changed` 广播后**只叫醒了 watcher,却没有失效自己的缓存**。 +`attachClusterPubSub()` 的订阅回调此前只做一件事 —— `notifyWatchersLocal()`, +既不碰 `this.registry` 也不碰 `this.listCache`。后果是 B 上任何走 `list(type)` +的读在 `LIST_CACHE_TTL_MS`(30 秒)窗口内继续返回改动前的清单;更糟的是,被叫醒的 +watcher(ObjectQL SchemaRegistry 桥、Studio HMR SSE)如果回头调 `list()` 重新拉取, +拉到的还是旧的 —— 一份「失效通知」附带着失效数据。单机部署完全无感,只有多节点才暴露。 + +这与该通道自己声明的用途相反(`ClusterMetadataChangedPayload`:"consumed by peers +to **invalidate their local caches**",另见 `content/docs/kernel/cluster.mdx` §6.2 +与 `metadata-lifecycle.mdx`);现在实现与声明一致。 + +修法沿用同文件里 `applyRepoEvent()` 自 ADR-0008 PR-6 起就用对的那条路径,并把两条 +「外部写入」缝(仓库 watch 循环、集群对端回放)收敛到同一个私有方法 +`invalidateForForeignWrite(type, name)`: + +- **删除而不预填。** 即便事件带着 body,也只删除 registry 条目而不写入 —— + 那份 body 是别人那次写入的快照,可能已被后续写入取代,预填会与真实 head 竞态, + 并要求我们去规范化一份自己没有加载过的定义。删除后 `get()` 自然穿透到 loader / + repository,也就是真相所在。 +- **同步失效,先失效再通知。** 失效发生在收到消息的当拍(不在 `setImmediate` 内), + 通知仍然延迟派发。`setImmediate` 的存在理由是不让**消费方的 watcher 回调**背压 + pubsub 派发循环;而失效只是两次 `Map.delete`,不执行任何消费方代码,没有需要延迟的 + 东西——把它一起延迟只会留下「已收到广播、尚未失效」的读窗口,请求处理器里任何一个 + `await` 都足以撞进去。先失效后通知也与本文件其他写入路径 + (`register` / `unregister` / `applyRepoEvent`)一致,于是回头 `list()` 的 watcher + 拿到的是写后清单。 +- **无名事件只失效清单缓存。** `MetadataWatchEvent.name` 在 spec 里是可选的,无名事件 + 无法定位 registry 条目;此时不会把整个 type 的 registry 一并清掉 —— 那会驱逐 + `registerInMemory()` 注册的、任何 loader 都无法恢复的代码态构件(如 `origin:'code'` + 的 datasource)。 + +回环抑制(`originNode`)仍然先于失效判断,本节点自己的广播不会让自己白白重建缓存。 diff --git a/content/docs/kernel/cluster.mdx b/content/docs/kernel/cluster.mdx index b23e17009e..1c7649e185 100644 --- a/content/docs/kernel/cluster.mdx +++ b/content/docs/kernel/cluster.mdx @@ -346,9 +346,20 @@ The transport is handed to the manager by `MetadataClusterBridgePlugin` `kernel:ready`. `Runtime` registers that bridge automatically alongside the cluster service. -Peers suppress their own messages by `originNode` and replay the watch -event locally — there is currently **no** `version` / `name` / `tenantId` / -`operation` field and **no** version comparison. +On receipt a peer suppresses its own messages by `originNode`, then — **first, +synchronously** — invalidates its local caches for that type: it drops the +`registry` entry named by the nested watch event and the `list(type)` cache +(`invalidateForForeignWrite`). Only then does it replay the watch event into +its local watch hub, deferred by one tick so a slow consumer callback cannot +back-pressure the pubsub dispatch loop. That order is what lets a woken +consumer answer by re-reading through `list()` and see the write rather than +the pre-write set (#5109). The registry entry is **deleted, never pre-filled** +from the payload — the peer re-reads the shared store, which is the source of +truth. + +At the payload's top level there is still **no** `version` / `name` / +`tenantId` / `operation` field and **no** version comparison; the item's name +is carried only inside the replayed `event`. **Target spec (planned).** The richer, version-stamped payload below is defined as `MetadataChangedEventPayloadSchema` in `kernel/cluster.zod.ts` diff --git a/packages/metadata/src/metadata-manager-cluster.test.ts b/packages/metadata/src/metadata-manager-cluster.test.ts index e00cdd1425..2c2b3701f3 100644 --- a/packages/metadata/src/metadata-manager-cluster.test.ts +++ b/packages/metadata/src/metadata-manager-cluster.test.ts @@ -3,6 +3,8 @@ import { describe, it, expect, vi } from 'vitest'; import { MetadataManager } from './metadata-manager'; import { MemoryLoader } from './loaders/memory-loader'; +import type { MetadataLoader } from './loaders/loader-interface.js'; +import type { MetadataLoaderContract, MetadataLoadResult, MetadataSaveResult, MetadataStats } from '@objectstack/spec/system'; import type { IPubSub, PubSubMessage } from '@objectstack/spec/contracts'; vi.mock('@objectstack/core', () => ({ @@ -39,8 +41,79 @@ class TestPubSub implements IPubSub { async close(): Promise { this.subs.clear(); } } +/** + * A writable, `datasource:`-protocol loader whose storage can be SHARED by two + * managers — the in-test stand-in for the one `sys_metadata` table two cluster + * replicas both read and write. + * + * `MemoryLoader` cannot play this role: its protocol is `memory:`, and + * `MetadataManager.register()` persists only to `datasource:` loaders that + * declare `capabilities.write`. Without a shared writable store, node A's + * write is invisible to node B no matter how node B's caches behave, and the + * #5109 regression cannot be observed at all. + */ +class SharedStoreLoader implements MetadataLoader { + readonly contract: MetadataLoaderContract = { + name: 'shared-store', + protocol: 'datasource:', + capabilities: { read: true, write: true, watch: false, list: true }, + }; + + // type -> name -> data. Deliberately public: tests assert on it. + readonly storage = new Map>(); + + async load(type: string, name: string): Promise { + const data = this.storage.get(type)?.get(name); + return data ? { data, source: 'shared-store', format: 'json', loadTime: 0 } : { data: null }; + } + async loadMany(type: string): Promise { + return Array.from((this.storage.get(type) ?? new Map()).values()) as T[]; + } + async exists(type: string, name: string): Promise { + return this.storage.get(type)?.has(name) ?? false; + } + async stat(type: string, name: string): Promise { + return (await this.exists(type, name)) + ? { size: 0, mtime: new Date().toISOString(), format: 'json' } + : null; + } + async list(type: string): Promise { + return Array.from((this.storage.get(type) ?? new Map()).keys()); + } + async save(type: string, name: string, data: unknown): Promise { + let typeStore = this.storage.get(type); + if (!typeStore) { typeStore = new Map(); this.storage.set(type, typeStore); } + typeStore.set(name, data); + return { success: true, path: `${type}/${name}` }; + } + async delete(type: string, name: string): Promise { + this.storage.get(type)?.delete(name); + } +} + const flush = () => new Promise((r) => setImmediate(r)); +/** Read a manager's private list cache without waiting on any async seam. */ +const cachedTypes = (mgr: MetadataManager): string[] => + Array.from((mgr as unknown as { listCache: Map }).listCache.keys()); + +/** + * Two managers wired to one bus and one shared store — the cluster shape: + * two replicas of the same app in front of the same `sys_metadata`. + */ +function makeCluster() { + const bus = new TestPubSub(); + const store = new SharedStoreLoader(); + const a = new MetadataManager({ formats: ['json'], loaders: [store] }); + const b = new MetadataManager({ formats: ['json'], loaders: [store] }); + a.attachClusterPubSub(bus, 'node-A'); + b.attachClusterPubSub(bus, 'node-B'); + return { bus, store, a, b }; +} + +const viewNames = (items: unknown[]): string[] => + (items as { name: string }[]).map((v) => v.name).sort(); + function makeManager(): MetadataManager { return new MetadataManager({ formats: ['json'], @@ -171,3 +244,154 @@ describe('MetadataManager — cluster pub/sub bridge', () => { expect(received).toHaveLength(0); }); }); + +/** + * #5109 — a peer's write must invalidate THIS node's caches, not just wake its + * watchers. + * + * Before this, `attachClusterPubSub`'s subscriber did exactly one thing on an + * incoming `metadata.changed`: `notifyWatchersLocal()`. It never touched + * `registry` or `listCache`. So node B's watchers were woken while every + * `list(type)` on B kept answering the pre-write set for up to + * `LIST_CACHE_TTL_MS` (30s) — and a watcher that responded to the wake-up by + * re-reading through `list()` was handed the stale set back. An invalidation + * notice carrying invalidated data. + * + * The fix reuses `applyRepoEvent`'s long-standing shape (delete the registry + * entry, drop the list cache, THEN announce) via `invalidateForForeignWrite`. + */ +describe('MetadataManager — a cluster peer write invalidates local caches (#5109)', () => { + const view = (name: string, title: string) => ({ name, title, object: 'account' }); + + it('B\'s pre-warmed list() sees A\'s register() without waiting out the 30s TTL', async () => { + const { store, a, b } = makeCluster(); + await store.save('view', 'v_existing', view('v_existing', 'existing')); + + // B pre-warms its list cache — the issue's repro, verbatim. + expect(viewNames(await b.list('view'))).toEqual(['v_existing']); + + // A writes. Same tick, no timers: the 30s TTL has not lapsed. + await a.register('view', 'v_new', view('v_new', 'new')); + + expect(viewNames(await b.list('view'))).toEqual(['v_existing', 'v_new']); + }); + + it('propagates A\'s unregister() to B\'s list()', async () => { + const { store, a, b } = makeCluster(); + await store.save('view', 'v_doomed', view('v_doomed', 'doomed')); + expect(viewNames(await b.list('view'))).toEqual(['v_doomed']); + + await a.unregister('view', 'v_doomed'); + + expect(await b.list('view')).toEqual([]); + }); + + it('invalidates SYNCHRONOUSLY on receipt — not inside the deferred replay', async () => { + const { store, a, b } = makeCluster(); + await store.save('view', 'v_existing', view('v_existing', 'existing')); + await b.list('view'); + expect(cachedTypes(b)).toEqual(['view']); + + const woken: unknown[] = []; + b.subscribe('view', (e: unknown) => { woken.push(e); }); + + await a.register('view', 'v_new', view('v_new', 'new')); + + // Resuming from `await` is a microtask, so no `setImmediate` callback + // can have run yet: the watcher replay is still pending... + expect(woken).toHaveLength(0); + // ...and the cache is ALREADY gone. This is the seam the fix pins: a + // read landing between receipt and the deferred tick — every `await` + // inside a request handler is such a window — must not be served + // stale. Deferring the invalidation alongside the notify would leave + // this expectation red. + expect(cachedTypes(b)).toEqual([]); + + await flush(); + expect(woken).toHaveLength(1); + }); + + it('a watcher that re-reads via list() on the wake-up gets the post-write set', async () => { + const { store, a, b } = makeCluster(); + await store.save('view', 'v_existing', view('v_existing', 'existing')); + await b.list('view'); + + // The ObjectQL SchemaRegistry bridge and the HMR SSE stream both react + // to the event by re-reading. That re-read is what used to contradict + // the notification it was answering. + let seenByWatcher: string[] = []; + const observed = new Promise((resolve) => { + b.subscribe('view', () => { + void b.list('view').then((items: unknown[]) => { + seenByWatcher = viewNames(items); + resolve(); + }); + }); + }); + + await a.register('view', 'v_new', view('v_new', 'new')); + await observed; + + expect(seenByWatcher).toEqual(['v_existing', 'v_new']); + }); + + it('drops B\'s stale registry entry so get() falls through to the shared store', async () => { + const { store, a, b } = makeCluster(); + // B holds its own copy in the in-memory registry, which shadows the + // store in both get() and list(). Dropping the list cache alone would + // leave B serving this copy forever. + await b.register('view', 'v_shared', view('v_shared', 'old')); + expect((await b.get('view', 'v_shared') as { title: string }).title).toBe('old'); + + await a.register('view', 'v_shared', view('v_shared', 'new')); + + expect((await b.get('view', 'v_shared') as { title: string }).title).toBe('new'); + expect((await b.list('view') as { title: string }[])[0].title).toBe('new'); + // Deleted, never pre-filled from the payload — the answer came from + // the store, which is the source of truth (see + // `invalidateForForeignWrite`). + expect(store.storage.get('view')?.get('v_shared')).toMatchObject({ title: 'new' }); + }); + + it('a nameless remote event invalidates the list cache but keeps in-memory-only entries', async () => { + const bus = new TestPubSub(); + const mgr = makeManager(); + mgr.attachClusterPubSub(bus, 'node-B'); + + // `registerInMemory` artefacts (code-owned datasources, ADR-0015 + // Addendum) exist in NO loader — evicting one on a nameless event + // would be an unrecoverable loss in exchange for a guess. + mgr.registerInMemory('datasource', 'crm_db', { name: 'crm_db', origin: 'code' }); + await mgr.list('datasource'); + expect(cachedTypes(mgr)).toEqual(['datasource']); + + // `MetadataWatchEvent.name` is optional in the spec, so this payload + // is legal on the wire. + await bus.publish('metadata.changed', { + originNode: 'node-A', + type: 'datasource', + event: { type: 'changed', path: '/some/file.json', timestamp: Date.now() }, + }); + await flush(); + + expect(cachedTypes(mgr)).toEqual([]); + expect(await mgr.get('datasource', 'crm_db')).toMatchObject({ name: 'crm_db' }); + }); + + it('leaves the local node\'s own caches alone on a loopback event', async () => { + const { store, a } = makeCluster(); + await store.save('view', 'v_existing', view('v_existing', 'existing')); + await a.list('view'); + expect(cachedTypes(a)).toEqual(['view']); + + // A hears its own broadcast back on a driver without dedup — the + // loopback guard must still short-circuit before any invalidation, or + // every local write would pay a needless cache rebuild. + await (a as unknown as { clusterPubSub: IPubSub }).clusterPubSub.publish( + 'metadata.changed', + { originNode: 'node-A', type: 'view', event: { type: 'changed', name: 'v_existing', path: '' } }, + ); + + expect(cachedTypes(a)).toEqual(['view']); + }); +}); diff --git a/packages/metadata/src/metadata-manager.ts b/packages/metadata/src/metadata-manager.ts index b47922c89f..c69369c4bb 100644 --- a/packages/metadata/src/metadata-manager.ts +++ b/packages/metadata/src/metadata-manager.ts @@ -185,9 +185,12 @@ export class MetadataManager implements IMetadataService { // become cluster-wide: // • Local notifyWatchers() publishes on `metadata.changed` so peers // can invalidate their caches. - // • Subscribed remote events are replayed into the local watch hub - // so existing consumers (ObjectQLPlugin, Studio HMR, …) see - // uniform behavior regardless of which node initiated the change. + // • A subscribed remote event first invalidates THIS node's caches + // (registry entry + listCache, via `invalidateForForeignWrite` — + // #5109) and is then replayed into the local watch hub, so existing + // consumers (ObjectQLPlugin, Studio HMR, …) see uniform behavior + // regardless of which node initiated the change — including when + // they answer the event by re-reading through `list()`. // `originNode` on the payload prevents loopback; `partitionKey` keeps // per-object ordering on partitioned drivers. private clusterPubSub?: IPubSub; @@ -207,9 +210,9 @@ export class MetadataManager implements IMetadataService { // ── #5089 (#5040 E2): declared-endpoint index ──────────────────────── // Backs `matchEndpoint`. Lazily built from `api` items on the first call // and invalidated by every path that can change them — see - // `invalidateListCache` (local writes, repo events, HMR/artifact ingest, - // which registers with `notify:false`) and the `subscribe('api', …)` - // registration below (cluster peer replay, which reaches watchers only). + // `invalidateListCache` (local writes, repo events, HMR/artifact ingest + // which registers with `notify:false`, and — since #5109 — cluster peer + // replay) and the `subscribe('api', …)` registration below. private static readonly ENDPOINT_METADATA_TYPE = 'api'; private readonly endpointMatcher: EndpointMatcher; @@ -218,17 +221,18 @@ export class MetadataManager implements IMetadataService { this.logger = createLogger({ level: 'info', format: 'pretty' }); // [#5089] Endpoint index (see `matchEndpoint`). Two invalidation seams, - // covering disjoint event sets — both are needed, neither is redundant: - // 1. `invalidateListCache('api')` — every LOCAL mutation of the stored - // set, including the `{ notify: false }` writes the artifact ingest - // and the HMR reload use, which by construction never reach a - // watcher. It is the same invariant the list cache carries: if the - // cached list of a type is stale, so is the index built from it. - // 2. `subscribe('api', …)` — a CLUSTER peer's write, which - // `attachClusterPubSub` replays through `notifyWatchersLocal` only - // and therefore does not pass through (1). (That the peer replay - // leaves the manager's OWN caches stale is #5109; the index does not - // inherit the bug because it listens on the watcher too.) + // covering overlapping but non-identical event sets — both are kept: + // 1. `invalidateListCache('api')` — every mutation of the stored set + // this manager learns about, including the `{ notify: false }` + // writes the artifact ingest and the HMR reload use, which by + // construction never reach a watcher. It is the same invariant the + // list cache carries: if the cached list of a type is stale, so is + // the index built from it. Since #5109 a CLUSTER peer's write also + // passes through here, via `invalidateForForeignWrite`. + // 2. `subscribe('api', …)` — the watcher seam, which additionally + // covers events raised by subclasses / test doubles that call + // `notifyWatchers` without a cache mutation of their own. + // Double invalidation is idempotent, so the overlap is free. this.endpointMatcher = new EndpointMatcher({ listApiItems: () => this.listForIndex(MetadataManager.ENDPOINT_METADATA_TYPE), logger: this.logger, @@ -2027,22 +2031,57 @@ export class MetadataManager implements IMetadataService { } } + /** + * Drop every local cache of `type` (and of `name` within it) that a change + * we did not perform ourselves has just invalidated, so the next read falls + * through to the source of truth. + * + * The two callers are the manager's two *foreign-write* seams — the + * repository watch loop ({@link applyRepoEvent}) and the cluster peer replay + * in {@link attachClusterPubSub}. Both learn about a write that landed + * somewhere else (the repo head; another node's `sys_metadata`) and hold + * caches that the write silently aged out. Local writes do not come through + * here: `register()` / `unregister()` / `registerInMemory()` update the + * registry to the value they just wrote and call `invalidateListCache()` + * themselves. + * + * **Delete, do not pre-fill.** Even when the event carries a body we drop the + * registry entry rather than writing the body into it: the body reaching us + * is a snapshot of *someone else's* write, already possibly superseded, and + * pre-filling would race with the true head and require us to re-canonicalise + * a definition we did not load. Lazy invalidation is the safer default — + * `get()` then falls through to the loaders / repository, which is where the + * truth is. (This paragraph is the rationale `applyRepoEvent` carried since + * ADR-0008 PR-6; #5109 extended the same choice to the cluster path.) + * + * `name` is optional because `MetadataWatchEvent.name` is: a nameless event + * cannot address a registry entry, so it invalidates the list cache only. + * Dropping the whole type store instead would evict `registerInMemory()` + * artefacts (code-owned datasources, ADR-0015 Addendum) that no loader can + * restore — an unrecoverable loss in exchange for a guess. + */ + private invalidateForForeignWrite(type: string, name?: string): void { + if (name) { + const typeStore = this.registry.get(type); + if (typeStore) { + typeStore.delete(name); + if (typeStore.size === 0) this.registry.delete(type); + } + } + this.invalidateListCache(type); + } + /** Translate a repo event to the legacy MetadataWatchEvent + invalidate caches. */ private applyRepoEvent(evt: MetadataEvent): void { const ref: MetaRef = evt.ref; const type = ref.type; const name = ref.name; - // Invalidate in-memory registry so manager.get() falls through to - // loaders / repository on next read. We do NOT pre-fill the registry - // here — that would race with the repo head and require us to - // re-canonicalise. Lazy invalidation is the safer default. - const typeStore = this.registry.get(type); - if (typeStore) { - typeStore.delete(name); - if (typeStore.size === 0) this.registry.delete(type); - } - this.listCache.delete(type); + // Invalidate before announcing, so a watcher that re-reads on the event + // observes the write rather than the pre-write cache. See + // {@link invalidateForForeignWrite} for why the registry entry is deleted + // rather than pre-filled. + this.invalidateForForeignWrite(type, name); const legacyType: 'added' | 'changed' | 'deleted' = evt.op === 'create' ? 'added' @@ -2137,6 +2176,40 @@ export class MetadataManager implements IMetadataService { // Loopback guard — never replay events we just emitted. if (p?.originNode && p.originNode === this.clusterNodeId) return; if (!p?.type || !p.event) return; + + // [#5109] Invalidate FIRST, and SYNCHRONOUSLY on receipt — this is + // what the channel is for ("consumed by peers to invalidate their + // local caches", see ClusterMetadataChangedPayload). Until this + // landed, a peer's write only woke this node's watchers: the registry + // entry and the `listCache` were left untouched, so every `list(type)` + // kept serving the pre-write set for up to LIST_CACHE_TTL_MS (30s) — + // and a watcher that re-read via `list()` in response to the wake-up + // got the stale set back, an invalidation notice carrying invalidated + // data. + // + // Two deliberate choices, both pinned by tests in + // `metadata-manager-cluster.test.ts`: + // + // • BEFORE the notify, matching every other write path in this file + // (`register` / `unregister` / `applyRepoEvent` all invalidate, + // then announce): a watcher must never be able to observe the + // event and the pre-event cache at the same time. + // • OUTSIDE the `setImmediate`, unlike the notify. The deferral + // exists so a slow *watcher callback* — arbitrary consumer code — + // cannot back-pressure the pubsub dispatch loop. Invalidation is + // two `Map.delete`s and runs no consumer code, so it has nothing + // to defer for, while deferring it would leave a window between + // receipt and the next tick in which reads still answer stale. + // Any `await` in a request handler is enough to lose that race. + try { + this.invalidateForForeignWrite(p.type, p.event.name); + } catch (err) { + this.logger.error('Cluster remote invalidation failed', undefined, { + type: p.type, + error: err instanceof Error ? err.message : String(err), + }); + } + // Defer to setImmediate so a slow local handler can't back-pressure // the pubsub dispatch loop on memory drivers. setImmediate(() => {