fix(metadata-fs): FileSystemRepository.close() terminates every live watch() iterator (#11127)

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

@@ -0,0 +1,44 @@
+---
+"@objectstack/metadata-fs": patch
+---
+
+`FileSystemRepository.close()` now terminates every live `watch()` iterator
+instead of leaving it parked (#11127). A consumer holding a `for await` over
+`watch()` at shutdown never saw its loop end — on a repository that was already
+gone.
+
+`close()` retired the chokidar watcher and the resync sweep and stopped there.
+It never reached the event broker, and the broker had no teardown of its own:
+`subscribe`/`unsubscribe` add to and delete from a plain `Set`, and nothing else
+emptied it. Each iterator parks its pending `next()` on a `waiter` that only two
+things can settle — a broker `push`, or the iterator's own terminator, which ran
+from `iterator.return()`/`throw()` and from nowhere else. After `close()` the
+chokidar source was gone so no `push` could arrive, and the subscriber was still
+registered with nothing left to run its terminator.
+
+Unlike the sibling defect in `SysMetadataRepository` (#11021) this was not
+filter-dependent: there was no drain attempt at all, so every subscription shape
+hung, `watch({})` included. Measured before the fix: nine cases —
+`watch({org}, seq)`, `watch({org})`, `watch({})`, a ref-exact filter, a watcher
+over the real chokidar watcher, a watcher with no pull outstanding, four
+concurrent watchers, and the `return()`-symmetry comparison — were all still
+unsettled 2s after `close()`. `MetadataManager.startRepositoryWatch()`, which
+awaits `iter.next()` in a loop, is exactly the shape that hung.
+
+The broker now holds each subscription's terminator next to its event sink, and
+`close()` runs every 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. Shutdown is deliberately **not** delivered
+as an event: a synthetic drain event is subject to the very filters `watch()`
+applies to real ones, and delivering an event has never ended an iterator
+(invariant 8, `@objectstack/metadata-core`'s `repository.ts`).
+
+One narrower path is closed with it. `watch()` returns a deferred iterable whose
+subscriber registers only once the eager log read resolves, so a `close()`
+landing inside that window swept a broker the subscription had not yet joined —
+the same forever-parked shape by a different route. `watch()` now carries the
+close generation it was opened under, and a subscription that arrives after a
+shutdown terminates on arrival.
+
+Invariant 8 named `FileSystemRepository` as its one known non-conformance. With
+this change the invariant has no declared exceptions, and its text says so.

packages/metadata-core/src/repository.ts

@@ -74,13 +74,18 @@
  *
  * 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.
+ * offered, this is what it owes. Measured across today's three, and there
+ * are **no declared exceptions**: `SysMetadataRepository` conforms (#11021);
+ * `FileSystemRepository` conforms (#11127 — its `close()` used to retire
+ * the filesystem watcher and the resync sweep without ever reaching its
+ * event broker, leaving a parked iterator parked for every subscription
+ * shape, `watch({})` included; it now runs each subscription's terminator);
+ * `InMemoryRepository` offers no repository-level shutdown at all, so its
+ * iterators end only through `return()`.
+ *
+ * A new implementation that offers `close()` joins that list or it does not
+ * conform — this row carries the measurement, so an implementation added
+ * without one is the omission, not an exception.
  */
 
 import type {

packages/metadata-fs/src/repository.ts

@@ -19,6 +19,9 @@
  * - The root directory is created **on the first write, not on attach**
  * (#7000). Attaching and reading a repository whose root does not exist
  * is legal and answers "empty"; see `start()` / `ensureRoot()`.
+ * - `close()` ENDS every live `watch()` iterator — the same observation the
+ * consumer's own `iterator.return()` produces, never a synthetic event
+ * standing in for shutdown (#11127; invariant 8 in `metadata-core`).
  */
 
 import fs from 'node:fs/promises';
@@ -132,6 +135,15 @@ export class FileSystemRepository implements MetadataRepository {
  * the first degradation). An entry is cleared when that path reads again.
  */
  private readonly resyncFaults = new Set<string>();
+ /**
+ * Bumped by every `close()`. `watch()` reads it before its deferred log
+ * replay starts and hands the comparison to `createWatchIterable`, so a
+ * subscription that registers AFTER the shutdown sweep terminates on
+ * arrival instead of parking forever (#11127). A counter rather than a
+ * boolean because `start()` may follow `close()`: a repository restart must
+ * not poison the watchers opened after it.
+ */
+ private closeGeneration = 0;
 
  constructor(opts: FileSystemRepositoryOptions) {
  this.org = opts.org;
@@ -195,10 +207,42 @@ export class FileSystemRepository implements MetadataRepository {
  if (this.started && !this.disableWatch && !this.watcher) this.startWatcher();
  }
 
+ /**
+ * Shut the repository down, ending every live `watch()` iterator.
+ *
+ * **Shutdown terminates; it does not emit** — invariant 8 in
+ * `@objectstack/metadata-core`'s `repository.ts`, and the reason this method
+ * reaches the broker at all. It used to retire the chokidar watcher and the
+ * resync sweep and stop there. The broker has no teardown of its own
+ * (`subscribe`/`unsubscribe` add to and delete from a plain `Set`), and each
+ * iterator parks its pending `next()` on a `waiter` that only a broker
+ * `push` or the iterator's own terminator can settle. After `close()` the
+ * chokidar source was gone, so no `push` could arrive; the subscriber was
+ * still registered, and nothing ran its terminator. A consumer holding a
+ * `for await` at shutdown — `MetadataManager.startRepositoryWatch()` is
+ * exactly that shape — therefore never saw its loop end, for EVERY
+ * subscription shape including `watch({})`.
+ *
+ * Termination is expressed as termination: each subscription's
+ * `terminate()`, which is the same routine the consumer's own
+ * `iterator.return()` runs, so no consumer has to tell "the repository shut
+ * down under me" apart from "I broke my own loop". A synthetic drain event
+ * would be the wrong shape and was measured to be so (#11021): the
+ * subscriptions most in need of draining are exactly the ones whose filter
+ * or numeric `since` drops it, and delivering an event has never ended an
+ * iterator.
+ */
  async close(): Promise<void> {
  // Retire the sweep BEFORE awaiting the watcher, so a sweep that lands
  // during `watcher.close()` cannot reschedule itself behind our back.
  this.stopResync();
+ // Terminate BEFORE the await for the same reason: a `watcher.close()` that
+ // rejects must not leave a consumer's `for await` parked forever, and a
+ // straggler event from the dying watcher has no one left to reach. Events
+ // still queued or unreplayed at this moment MAY be dropped (invariant 8),
+ // on this path and on `return()` alike.
+ this.closeGeneration++;
+ this.broker.terminateAll();
  if (this.watcher) {
  await this.watcher.close();
  this.watcher = null;
@@ -284,6 +328,11 @@ export class FileSystemRepository implements MetadataRepository {
  if (matchEvent(evt, filter)) replay.push(evt);
  }
  })();
+ // Read BEFORE the read above can complete: the subscriber below is
+ // registered only when it does, which is a window `close()`'s sweep cannot
+ // see (#11127). Compared on arrival, a shutdown inside that window ends
+ // this iterator instead of parking it.
+ const generation = this.closeGeneration;
  // We must await replay before returning, but the public API is
  // sync-returning AsyncIterable. Wrap in a deferred iterable.
  return deferredIterable(promise.then(() =>
@@ -294,6 +343,7 @@ export class FileSystemRepository implements MetadataRepository {
  broker: this.broker,
  matches: matchEvent,
  branchKeyOf: (e) => e.ref.org,
+ arrivesClosed: () => this.closeGeneration !== generation,
  }),
  ));
  }

packages/metadata-fs/src/sync.ts

@@ -31,16 +31,42 @@ export class KeyedMutex {
  }
 }
 
+/**
+ * One live `watch()` subscription, as a record rather than a bare event sink.
+ *
+ * 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
+ * — and delivering one has never ended an iterator anyway. See invariant 8 in
+ * `@objectstack/metadata-core`'s `repository.ts` (#11021, #11127).
+ */
 export interface BrokerSubscriber {
  filter: WatchFilter;
  closed: boolean;
  push(evt: MetadataEvent): void;
+ /**
+ * Ends this subscription's iterator: settles a parked `next()` with
+ * `{ done: true }` and no value, 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 interface EventBroker {
  subscribe(sub: BrokerSubscriber): void;
  unsubscribe(sub: BrokerSubscriber): void;
  publish(evt: MetadataEvent): void;
+ /**
+ * Terminate every live subscription. This is what `FileSystemRepository`'s
+ * repository-level `close()` owes a pending iterator (#11127): before it
+ * existed, `close()` retired the chokidar watcher and the resync sweep and
+ * stopped there, so the source that could settle a parked `next()` was gone
+ * while the subscriber stayed registered with nothing left to settle it.
+ *
+ * Idempotent, and a no-op when nothing is watching.
+ */
+ terminateAll(): void;
 }
 
 export function createBroker(matches: (evt: MetadataEvent, filter: WatchFilter) => boolean): EventBroker {
@@ -55,5 +81,19 @@ export function createBroker(matches: (evt: MetadataEvent, filter: WatchFilter)
  s.push(evt);
  }
  },
+ terminateAll: () => {
+ // Snapshot and clear BEFORE terminating: `terminate()` unregisters
+ // itself, and mutating a Set under its own iteration is how the second
+ // subscriber gets skipped.
+ const snapshot = Array.from(subs);
+ subs.clear();
+ for (const s of snapshot) {
+ try {
+ s.terminate();
+ } catch {
+ /* one wedged consumer must not strand the rest */
+ }
+ }
+ },
  };
 }

packages/metadata-fs/src/watch-iterable.ts

@@ -18,6 +18,14 @@ export interface CreateWatchIteratorArgs {
  /** Returns true if `evt.ref` matches `filter`. */
  matches: (evt: MetadataEvent, filter: WatchFilter) => boolean;
  branchKeyOf: (evt: MetadataEvent) => string;
+ /**
+ * Checked ONCE, immediately after this subscription is registered. True
+ * means the repository shut down while `watch()`'s deferred log replay was
+ * still in flight, so this subscription arrived after `close()` had already
+ * swept the broker. It is terminated on arrival rather than left parked on a
+ * broker nobody will publish to or drain again (#11127).
+ */
+ arrivesClosed?: () => boolean;
 }
 
 export function createWatchIterable(
@@ -32,6 +40,10 @@ export function createWatchIterable(
  const subscriber: BrokerSubscriber = {
  filter: args.filter,
  closed: false,
+ // Assigned below, once `close` exists. Termination and the consumer's own
+ // `return()` are ONE routine, deliberately: invariant 8 requires shutdown
+ // to be indistinguishable from `iterator.return()`.
+ terminate: () => undefined,
  push: (evt) => {
  if (subscriber.closed) return;
  const k = evtKey(evt);
@@ -84,6 +96,13 @@ export function createWatchIterable(
  return { value: undefined, done: true };
  };
 
+ // The terminator `repo.close()` runs. Same routine as `return()` below.
+ subscriber.terminate = close;
+
+ // Shutdown that landed while the deferred log replay was in flight — this
+ // subscription missed the sweep, so it terminates on arrival (#11127).
+ if (args.arrivesClosed?.()) close();
+
  const iterator: AsyncIterator<MetadataEvent> = {
  next: () => {
  if (closed) return Promise.resolve({ value: undefined, done: true });