fix(metadata-protocol): close() terminates watch iterators instead of emitting a drain event (#11021) (#11136)

`SysMetadataRepository.close()` modelled shutdown as a metadata event —
`{ seq: -1, ref: { org: '', type: 'view', name: '_close' } }` broadcast
through the same dispatch closure real events pass, then `watchers.clear()`.
Both of that closure's guards reject it: `matchesFilter` drops it for any
subscription naming an org (the synthetic ref's org is the empty string), a
type other than `view`, or a name; and `evt.seq <= since` holds for -1 against
every real seq. Dropped and then unsubscribed, nothing could settle the parked
promise and the consumer's `for await` never returned.

The subscriptions that passed both guards were no better off: they received the
synthetic event as a real one — a `view` named `_close`, deleted, at seq -1,
which MetadataManager turns into a cache invalidation and re-emits to Studio's
HMR stream — and hung on the next pull anyway, because delivering an event does
not end an iterator.

The watcher registry now holds each subscription's terminator next to its event
sink, and `close()` runs the terminator — the same routine `iterator.return()`
runs. Invariant 8 in metadata-core's repository.ts states the contract that was
unstated, and records FileSystemRepository's non-conformance (#11127).


Claude-Session: https://claude.ai/code/session_01RfyXxZ2WPjcjhuXpiQQc3y

Co-authored-by: Claude <noreply@anthropic.com>

Author: os-zhuang

.changeset/close-terminates-watch-iterators.md

@@ -0,0 +1,41 @@
+---
+"@objectstack/metadata-protocol": patch
+---
+
+`SysMetadataRepository.close()` now terminates every live `watch()` iterator
+instead of broadcasting a synthetic drain event (#11021). A consumer holding a
+`for await` over `watch()` at shutdown could hang forever, and the hang was
+worst for the subscription shapes most likely to be in use.
+
+Shutdown was modelled as a metadata event — `{ seq: -1, ref: { org: '', type:
+'view', name: '_close' } }` — pushed through the same dispatch closure real
+events pass, followed by clearing the watcher registry. Both of that closure's
+guards reject it:
+
+- `matchesFilter` drops it for any subscription naming an `org` (the synthetic
+  ref's org is the empty string), a `type` other than `view`, or a `name` —
+  `MetadataCache.start()` with any non-empty `watchFilter` is exactly that
+  shape;
+- the `since` drop-filter drops it for every numeric-`since` subscription,
+  since `-1 <= since` holds against every real seq.
+
+Dropped and then unsubscribed, nothing could settle the parked promise. Measured
+before the fix: `watch({org:'system'}, seq)` and `watch({org:'system'})` were
+both still unsettled 500ms after `close()`. The empty-filter case looked drained
+and was not — it received the synthetic event as a *real* one (a `view` named
+`_close`, deleted, at seq -1, which `MetadataManager` turns into a cache
+invalidation and re-emits to Studio's HMR stream) and then hung on the next pull
+anyway, because delivering an event does not end an iterator.
+
+`close()` now runs each subscription's terminator — the same routine the
+consumer's own `iterator.return()` runs — so a parked `next()` settles with
+`{ done: true }` and no value, and so does every later one. Consumers no longer
+need to recognise a shutdown event, because there is no longer one to recognise;
+nothing in the repo ever named the `_close` sentinel.
+
+The contract this repairs was unstated, which is why the two defensible repair
+shapes were both arguable. It is stated now: invariant 8 in
+`packages/metadata-core/src/repository.ts` ("shutdown terminates; it does not
+emit") says what a repository-level `close()` owes a pending iterator, and
+records the one measured non-conformance among today's implementations
+(`FileSystemRepository`, filed as #11127).

packages/metadata-core/src/repository.ts

@@ -50,6 +50,37 @@
  *    implementation was read first.
  * 7. **Tombstones, not holes.** `delete` produces a `delete` event;
  *    `get` returns null but `history` still shows the lineage.
+ * 8. **Shutdown terminates; it does not emit.** An implementation that offers
+ *    a repository-level shutdown (`close()`) MUST end every live `watch()`
+ *    iterator: a `next()` parked at that moment settles with `done: true` and
+ *    no value, and every later `next()` does the same. That is the identical
+ *    observation the consumer's own `iterator.return()` produces, deliberately
+ *    — so no consumer has to tell "the repository shut down under me" apart
+ *    from "I broke my own loop". Events still queued or unreplayed at that
+ *    moment MAY be dropped, on both paths alike.
+ *
+ *    **Shutdown MUST NOT be delivered AS an event.** Written as a MUST NOT
+ *    because it was tried, and both of its halves were measured (#11021). A
+ *    synthetic "we are closing" event is subject to the very filters `watch()`
+ *    applies to real ones, so the subscriptions that most need draining are
+ *    exactly the ones that drop it: any non-empty `filter` rejects a ref
+ *    invented to belong to no org, and any numeric `since` rejects a seq
+ *    invented to precede every real one. Those consumers then wait forever,
+ *    because the same shutdown unsubscribes them. Meanwhile a consumer whose
+ *    filter happens to admit it is not rescued either — it reads a real
+ *    metadata change for a ref that never existed (invalidating caches and
+ *    re-emitting downstream), and its iterator hangs on the *next* pull
+ *    regardless, because delivering an event has never ended one.
+ *
+ *    Stated conditionally because `close()` is not on the interface below;
+ *    it is offered by some implementations and not others. Where it is
+ *    offered, this is what it owes. Measured across today's three:
+ *    `SysMetadataRepository` conforms; `InMemoryRepository` offers no
+ *    repository-level shutdown at all, so its iterators end only through
+ *    `return()`; `FileSystemRepository.close()` retires the filesystem watcher
+ *    and the resync sweep but never reaches its event broker, so a parked
+ *    iterator stays parked — the one non-conformance, filed as #11127 rather
+ *    than quietly omitted from this row.
  */
 
 import type {
@@ -112,7 +143,10 @@ export interface MetadataRepository {
    *     already committed MAY also be delivered, but callers MUST NOT rely
    *     on it; a caller that needs them passes a numeric `since` or reads
    *     `history()`. See invariant 6.
-   *   - Stay open until the consumer breaks the loop.
+   *   - Stay open until the consumer breaks the loop — or until the
+   *     repository shuts down under it, where an implementation offers a
+   *     `close()`. Both end the stream the same way: `done: true`, no value,
+   *     never a synthetic event standing in for shutdown. See invariant 8.
    *   - Survive transient backend disconnects (implementation's choice
    *     how to resume — Postgres LISTEN reconnect, JSONL tail, etc.).
    */

packages/metadata-protocol/src/sys-metadata-repository.contract.test.ts

@@ -58,7 +58,7 @@ import {
   hashSpec,
 } from '@objectstack/metadata-core';
 import { runRepositoryContractTests } from '@objectstack/metadata-core/testing';
-import type { MetadataEvent } from '@objectstack/metadata-core';
+import type { MetadataEvent, WatchFilter } from '@objectstack/metadata-core';
 import { SysMetadataRepository } from './sys-metadata-repository.js';
 
 interface Row {
@@ -448,3 +448,134 @@ describe('SysMetadataRepository — invariant 6, both halves (#10842)', () => {
     await iter.return?.(undefined);
   });
 });
+
+/**
+ * #11021 — what `close()` owes a pending iterator.
+ *
+ * `close()` used to model shutdown as a metadata EVENT: it broadcast a
+ * synthetic `{ seq: -1, ref: { org: '', type: 'view', name: '_close' } }`
+ * through the same `dispatch` closure every real event passes, and then
+ * cleared the watcher set. Both of that closure's guards reject it:
+ *
+ *   - `matchesFilter` — the synthetic ref's org is the EMPTY STRING and its
+ *     type is always `view`, so any subscription naming an `org`, a `type`
+ *     other than `view`, or a `name` drops it;
+ *   - the `since` drop — `-1 <= since` holds against every real seq, so every
+ *     numeric-`since` subscription drops it too.
+ *
+ * Dropped, and then unsubscribed by `watchers.clear()`: nothing could ever
+ * settle the promise, and the consumer's `for await` never returned. The
+ * matrix below is the one the card was filed on, plus the row that is easy to
+ * misread — an EMPTY filter with no `since` passed both guards, so the pending
+ * pull settled, but it settled with `done: false` carrying the synthetic event
+ * as though a view named `_close` had been deleted at seq -1. The iterator
+ * then hung on the NEXT pull just like the other two.
+ *
+ * The repair is that shutdown is not an event. `close()` runs the same
+ * termination routine `iterator.return()` runs, on every live watcher — which
+ * is what these cases assert, and it is why the assertion is on `done: true`
+ * with NO value rather than on "something arrived".
+ */
+describe('SysMetadataRepository — close() terminates every live watcher (#11021)', () => {
+  const ref = { org: 'system', type: 'view' as const, name: 'sample_view' };
+
+  const PENDING = Symbol('still-pending');
+
+  /**
+   * Settle-or-report-pending. Every case here has to tell "settled with
+   * `done: true`" apart from "still unsettled", and a bare `await` on the
+   * unsettled shape hangs the RUN rather than failing the case.
+   */
+  function within<T>(p: Promise<T>, ms: number): Promise<T | typeof PENDING> {
+    return Promise.race([
+      p,
+      new Promise<typeof PENDING>((resolve) => setTimeout(() => resolve(PENDING), ms)),
+    ]);
+  }
+
+  /**
+   * Let the durable-replay promise settle, so the pull under test is genuinely
+   * PARKED on the live listener rather than still inside `await replayReady`.
+   * Without this the numeric-`since` row would prove less than it claims.
+   */
+  const parked = () => new Promise((resolve) => setTimeout(resolve, 50));
+
+  it.each([
+    ['filtered + numeric `since`', { org: 'system' } as WatchFilter, true],
+    // ⭐ The row that proves the org-filter half bites ON ITS OWN. A fix
+    // tested only against the `since` half looks complete and leaves this —
+    // `MetadataCache.start()` with any non-empty `watchFilter` — hanging.
+    ['filtered, no `since` at all', { org: 'system' } as WatchFilter, false],
+    // The row that looked drained and was not: it received the synthetic
+    // event, then hung on the next pull.
+    ['empty filter, no `since`', {} as WatchFilter, false],
+  ])('close() settles the pending next() with done:true — %s', async (_label, filter, withSince) => {
+    const repo = makeRepo();
+    const a = await repo.put(ref, { label: '1' }, { parentVersion: null, actor: 't' });
+
+    const iter = (withSince ? repo.watch(filter, a.seq) : repo.watch(filter))[
+      Symbol.asyncIterator
+    ]();
+    const pending = iter.next();
+    await parked();
+
+    repo.close();
+
+    // Termination — not a synthetic event wearing `done: false`.
+    expect(await within(pending, 500)).toEqual({ value: undefined, done: true });
+    // …and the iterator is FINISHED, not merely unblocked once. This is the
+    // half the old empty-filter row hid: one pull settled, the next hung.
+    expect(await within(iter.next(), 500)).toEqual({ value: undefined, done: true });
+  });
+
+  it('finishes a watcher that has no pull outstanding at close() time', async () => {
+    const repo = makeRepo();
+    const iter = repo.watch({ org: 'system' })[Symbol.asyncIterator]();
+    await parked();
+
+    repo.close();
+
+    expect(await within(iter.next(), 500)).toEqual({ value: undefined, done: true });
+  });
+
+  it('terminates EVERY live watcher, and is idempotent', async () => {
+    const repo = makeRepo();
+    const iters = [
+      repo.watch({ org: 'system' })[Symbol.asyncIterator](),
+      repo.watch({ type: 'view' })[Symbol.asyncIterator](),
+      repo.watch({ org: 'system', type: 'view', name: 'sample_view' })[Symbol.asyncIterator](),
+      repo.watch({})[Symbol.asyncIterator](),
+    ];
+    const pendings = iters.map((it) => it.next());
+    await parked();
+
+    repo.close();
+    repo.close();
+
+    for (const p of pendings) {
+      expect(await within(p, 500)).toEqual({ value: undefined, done: true });
+    }
+  });
+
+  it('ends the stream exactly the way the consumer’s own `return()` does', async () => {
+    // The contract sentence, as a comparison rather than a claim: a consumer
+    // that breaks its loop and a consumer whose repository shut down under it
+    // observe the SAME thing, so neither has to special-case the other.
+    const byReturn = makeRepo();
+    const a = byReturn.watch({ org: 'system' })[Symbol.asyncIterator]();
+    const aPending = a.next();
+    await parked();
+    void a.return?.(undefined);
+
+    const byClose = makeRepo();
+    const b = byClose.watch({ org: 'system' })[Symbol.asyncIterator]();
+    const bPending = b.next();
+    await parked();
+    byClose.close();
+
+    const viaReturn = await within(aPending, 500);
+    const viaClose = await within(bPending, 500);
+    expect(viaClose).toEqual(viaReturn);
+    expect(viaClose).toEqual({ value: undefined, done: true });
+  });
+});