diff --git a/.changeset/close-terminates-watch-iterators.md b/.changeset/close-terminates-watch-iterators.md new file mode 100644 index 0000000000..996db61f9c --- /dev/null +++ b/.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). diff --git a/packages/metadata-core/src/repository.ts b/packages/metadata-core/src/repository.ts index 2df69c50ea..8933715ce4 100644 --- a/packages/metadata-core/src/repository.ts +++ b/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.). */ diff --git a/packages/metadata-protocol/src/sys-metadata-repository.contract.test.ts b/packages/metadata-protocol/src/sys-metadata-repository.contract.test.ts index 269d474be0..99659fa1b6 100644 --- a/packages/metadata-protocol/src/sys-metadata-repository.contract.test.ts +++ b/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(p: Promise, ms: number): Promise { + return Promise.race([ + p, + new Promise((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 }); + }); +}); diff --git a/packages/metadata-protocol/src/sys-metadata-repository.ts b/packages/metadata-protocol/src/sys-metadata-repository.ts index f32891caca..4b176d8587 100644 --- a/packages/metadata-protocol/src/sys-metadata-repository.ts +++ b/packages/metadata-protocol/src/sys-metadata-repository.ts @@ -264,6 +264,26 @@ export function resetEnvWritableMetadataTypes(): void { _envWritableMetadataTypes = null; } +/** + * One live `watch()` subscription, as a record rather than a bare listener. + * + * Both halves live together because SHUTDOWN NEEDS THE SECOND ONE. A registry + * of event sinks can only express shutdown as "send an event", and an event is + * precisely what a filtered or numeric-`since` subscriber is entitled to drop + * (#11021). + */ +interface WatchSubscription { + /** Receives every broadcast event; applies this subscriber's own filters. */ + dispatch(evt: MetadataEvent): void; + /** + * Ends the iterator: settles any parked `next()` with `{ done: true }` and + * unregisters. The SAME routine `iterator.return()` runs, so a consumer that + * breaks its loop and a consumer whose repository shut down under it observe + * the same thing. + */ + terminate(): void; +} + export class SysMetadataRepository implements MetadataRepository { private readonly engine: SysMetadataEngine; private readonly organizationId: string | null; @@ -276,7 +296,11 @@ export class SysMetadataRepository implements MetadataRepository { * so we never broadcast events that got rolled back. */ private seqCounter = 0; - private readonly watchers = new Set<(evt: MetadataEvent) => void>(); + /** + * Every live `watch()` subscription. See {@link WatchSubscription} for why + * the terminator is held here next to the event sink. + */ + private readonly watchers = new Set(); private closed = false; /** Table name for the durable event log. */ @@ -1118,6 +1142,17 @@ export class SysMetadataRepository implements MetadataRepository { * no duplicates" across the seam. Replay is always drained first, so seq * order holds across it. `seqCounter` mirrors the durable `event_seq` * (see `put()`), which is what lets one set of numbers dedup both sides. + * + * ## How the stream ENDS + * + * Two ways, and they are the same way: the consumer calls `return()` (what + * `break`ing out of a `for await` does), or the repository shuts down under + * it and {@link close} runs the identical routine. Either settles a parked + * `next()` with `{ done: true }` and no value. A consumer therefore never + * has to recognise a shutdown *event* — there is not one to recognise, which + * is the #11021 repair; see `close()` for what modelling it as an event cost. + * Anything still queued or unreplayed at that point is dropped, on both + * paths alike. */ watch(filter: WatchFilter, since?: number): AsyncIterable { const self = this; @@ -1153,8 +1188,24 @@ export class SysMetadataRepository implements MetadataRepository { queue.push(evt); } }; + /** + * End this iterator. Runs on the consumer's own `return()` AND on + * `repo.close()` — ONE routine, so shutdown is never a shape the + * consumer has to recognise. Idempotent: a second call finds nothing + * registered and no promise parked. + */ + const terminate = (): void => { + stopped = true; + self.watchers.delete(subscription); + if (pendingResolve) { + const r = pendingResolve; + pendingResolve = null; + r({ value: undefined as any, done: true }); + } + }; + const subscription: WatchSubscription = { dispatch, terminate }; // Registered BEFORE the durable read starts — see the handoff note. - self.watchers.add(dispatch); + self.watchers.add(subscription); const replayReady: Promise = since === undefined @@ -1201,13 +1252,7 @@ export class SysMetadataRepository implements MetadataRepository { }); }, return(): Promise> { - stopped = true; - self.watchers.delete(dispatch); - if (pendingResolve) { - const r = pendingResolve; - pendingResolve = null; - r({ value: undefined as any, done: true }); - } + terminate(); return Promise.resolve({ value: undefined as any, done: true }); }, }; @@ -1215,27 +1260,49 @@ export class SysMetadataRepository implements MetadataRepository { }; } - /** Shut down all watch iterators. */ + /** + * Shut down every live `watch()` iterator. + * + * **Shutdown is not a metadata event** — #11021, and the reason this method + * no longer broadcasts anything. It used to push a synthetic + * `{ seq: -1, ref: { org: '', type: 'view', name: '_close' } }` through the + * same `dispatch` closure real events pass, then clear the registry. 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. `MetadataCache.start()` with + * any non-empty `watchFilter` is exactly that shape; + * - the `since` drop — `-1 <= since` holds against every real seq, so + * every numeric-`since` subscription drops it too. + * + * Dropped, and then unsubscribed by the clear: nothing could settle the + * parked promise, and the consumer's `for await` never returned. + * + * The subscriptions that DID pass both guards were no better off in the way + * that counts. They received the synthetic event as a REAL one — a `view` + * named `_close`, deleted, at seq -1 — which `MetadataManager.applyRepoEvent` + * duly turned into a cache invalidation and re-emitted to Studio's HMR + * stream; and then they hung on the next pull anyway, because delivering an + * event has never ended an iterator. + * + * So termination is expressed as termination: each subscription's + * {@link WatchSubscription.terminate}, which is the same routine the + * consumer's own `iterator.return()` runs. Idempotent, and a no-op when + * nothing is watching. + */ close(): void { this.closed = true; - // Drain watchers — each one's `return()` removes itself. + // Snapshot and clear BEFORE terminating: `terminate()` unregisters itself, + // and mutating a set under its own iteration is how the second watcher + // gets skipped. const snapshot = Array.from(this.watchers); + this.watchers.clear(); for (const w of snapshot) { try { - w({ - seq: -1, - op: 'delete', - ref: { org: '', type: 'view', name: '_close' } as MetaRef, - hash: null, - parentHash: null, - // #4556 — a synthetic drain event has no actor at all. - actor: null, - ts: new Date().toISOString(), - source: 'sys-metadata-repo-close', - }); - } catch { /* noop */ } + w.terminate(); + } catch { /* one wedged consumer must not strand the rest */ } } - this.watchers.clear(); } // ── helpers ───────────────────────────────────────────────────────── @@ -1641,7 +1708,7 @@ export class SysMetadataRepository implements MetadataRepository { private broadcast(evt: MetadataEvent): void { for (const w of Array.from(this.watchers)) { - try { w(evt); } catch { /* listener errors don't break the repo */ } + try { w.dispatch(evt); } catch { /* listener errors don't break the repo */ } } }