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

* test(metadata-fs): pin the invariant-8 shutdown contract (red baseline for #11127)

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

* chore(changeset): FileSystemRepository.close() terminates watch iterators

---------

Co-authored-by: Claude <pm@objectstack.ai>

Author: os-zhuang

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