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

.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 });
+ });
+});