fix(metadata): 集群对端的元数据写入现在会失效本节点的 listCache / registry (#5109)

.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`)仍然先于失效判断,本节点自己的广播不会让自己白白重建缓存。

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`

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<void> { 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<string, Map<string, unknown>>();
+
+    async load(type: string, name: string): Promise<MetadataLoadResult> {
+        const data = this.storage.get(type)?.get(name);
+        return data ? { data, source: 'shared-store', format: 'json', loadTime: 0 } : { data: null };
+    }
+    async loadMany<T = unknown>(type: string): Promise<T[]> {
+        return Array.from((this.storage.get(type) ?? new Map()).values()) as T[];
+    }
+    async exists(type: string, name: string): Promise<boolean> {
+        return this.storage.get(type)?.has(name) ?? false;
+    }
+    async stat(type: string, name: string): Promise<MetadataStats | null> {
+        return (await this.exists(type, name))
+            ? { size: 0, mtime: new Date().toISOString(), format: 'json' }
+            : null;
+    }
+    async list(type: string): Promise<string[]> {
+        return Array.from((this.storage.get(type) ?? new Map()).keys());
+    }
+    async save(type: string, name: string, data: unknown): Promise<MetadataSaveResult> {
+        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<void> {
+        this.storage.get(type)?.delete(name);
+    }
+}
+
 const flush = () => new Promise<void>((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<string, unknown> }).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<void>((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']);
+    });
+});