fix(vnext): close failed catalog capture lifecycle

Author: Light2Dark

docs/adr/0005-parser-independent-relation-completion.md

@@ -561,7 +561,9 @@ callback can always create a poisoned or independent unhandled rejection that
 no JavaScript library can retroactively contain. Legitimate asynchronous
 teardown must consume or report its own failure before the cleanup closure
 returns `undefined`. The valid `undefined` path returns without Promise
-allocation or a microtask. A closure avoids a structural TypeScript
+allocation or a microtask. Detached assimilation uses module-captured Promise
+intrinsics, so replacing the global Promise constructor cannot disable the
+drain. A closure avoids a structural TypeScript
 contract that would accept class instances or inherited methods which the
 hostile runtime boundary could not safely validate. Dropping the last owner
 removes the complete scope incarnation before cleanup. A later join creates a
@@ -660,6 +662,8 @@ outcomes settle through a discriminated request result.
 Completion is latest-wins per session:
 
 - a new completion request supersedes the previous request;
+- a new request that cannot capture an active epoch still supersedes and
+  detaches the previous request before reporting its unavailable outcome;
 - same-key supersession atomically attaches the new request consumer, or
   retags the existing observer as that consumer, before removing old request
   ownership; different-key supersession revokes the old observer first;

src/vnext/__tests__/relation-catalog-epoch-coordinator.test.ts

@@ -288,6 +288,56 @@ describe("catalog epoch coordinator construction and membership", () => {
     await Promise.resolve();
   });
 
+  it("drains a plain thenable with captured Promise intrinsics after the global constructor is replaced", async () => {
+    const intrinsicPromise = Promise;
+    let thenCalls = 0;
+    const thenable = new Proxy(
+      {},
+      {
+        get(target, property, receiver): unknown {
+          if (property === "then") {
+            return (
+              _resolve: (value: unknown) => void,
+              reject: (reason?: unknown) => void,
+            ): void => {
+              thenCalls += 1;
+              reject(new Error("detached plain thenable"));
+            };
+          }
+          return Reflect.get(target, property, receiver);
+        },
+      },
+    );
+    function HostilePromise(): never {
+      throw new Error("mutable global Promise was used");
+    }
+    expect(
+      Reflect.set(globalThis, "Promise", HostilePromise),
+    ).toBe(true);
+    try {
+      const created = Reflect.apply(
+        createSqlCatalogEpochCoordinator,
+        undefined,
+        [
+          capturedProvider(),
+          undefined,
+          () => thenable,
+        ],
+      );
+      expect(created.status).toBe("created");
+      if (created.status !== "created") {
+        throw new Error("Expected a coordinator fixture");
+      }
+      const createdCoordinator = created.coordinator;
+      expect(() => createdCoordinator.dispose()).not.toThrow();
+    } finally {
+      Reflect.set(globalThis, "Promise", intrinsicPromise);
+    }
+    await intrinsicPromise.resolve();
+    await intrinsicPromise.resolve();
+    expect(thenCalls).toBe(1);
+  });
+
   it("validates exact bounded well-formed scopes without raw errors", () => {
     const owner = coordinator();
     expect(

src/vnext/__tests__/relation-catalog-search-work.test.ts

@@ -1330,6 +1330,78 @@ describe("catalog search epoch authority and isolation", () => {
     });
   });
 
+  it("supersedes reentrant work when its retiring membership makes the next capture fail", async () => {
+    const listener: {
+      current: ((event: unknown) => void) | null;
+    } = { current: null };
+    const calls: ProviderCall[] = [];
+    const captured = accepted(
+      captureSqlRelationCatalogProvider({
+        id: "catalog",
+        search(
+          request: SqlCatalogSearchRequest,
+          signal: AbortSignal,
+        ) {
+          const settlement = deferred<unknown>();
+          calls.push({ request, settlement, signal });
+          return settlement.promise;
+        },
+        subscribe(
+          _scope: string,
+          onInvalidation: (event: unknown) => void,
+        ) {
+          listener.current = onInvalidation;
+          return () => undefined;
+        },
+      }),
+    );
+    const service = coordinator(captured);
+    let session: SqlCatalogSearchWorkOwner | null = null;
+    const reentrant: {
+      current: SqlCatalogSearchWorkTicket | null;
+    } = { current: null };
+    const prepared = service.prepareOwner(
+      "scope",
+      POSTGRESQL_SQL_RELATION_DIALECT,
+      {
+        prepareCatalogChange: () => {
+          reentrant.current =
+            session?.request(input("reentrant")) ?? null;
+          return null;
+        },
+      },
+    );
+    expect(prepared.status).toBe("prepared");
+    if (prepared.status !== "prepared") {
+      throw new Error("Expected prepared owner");
+    }
+    session = prepared.owner;
+    expect(session.activate()).toEqual({ status: "active" });
+    const initial = session.request(input("initial"));
+
+    listener.current?.({ epoch: epoch(1) });
+    await expect(initial.result).resolves.toEqual({
+      status: "superseded",
+    });
+    expect(reentrant.current).not.toBeNull();
+    expect(calls).toHaveLength(2);
+    expect(calls[0]?.signal.aborted).toBe(true);
+    expect(calls[1]?.signal.aborted).toBe(false);
+
+    const afterRetirement = session.request(input("after"));
+    await expect(afterRetirement.result).resolves.toEqual({
+      reason: "disposed",
+      status: "unavailable",
+    });
+    if (!reentrant.current) {
+      throw new Error("Expected reentrant request");
+    }
+    await expect(reentrant.current.result).resolves.toEqual({
+      status: "superseded",
+    });
+    expect(calls[1]?.signal.aborted).toBe(true);
+  });
+
   it("makes a higher response self-supersede and retire other same-scope work only", async () => {
     const provider = providerHarness();
     const service = coordinator(provider.captured);

src/vnext/relation-catalog-epoch-coordinator.ts

@@ -240,6 +240,8 @@ const SUBMITTED_RESULT: SqlCatalogResponseEpochSubmissionResult =
   Object.freeze({ status: "submitted" });
 const NO_PREPARE_CATALOG_CHANGE = (): null => null;
 const IGNORE_DETACHED_REJECTION = (): void => {};
+const INTRINSIC_PROMISE = Promise;
+const INTRINSIC_PROMISE_RESOLVE = Promise.resolve;
 const INTRINSIC_PROMISE_THEN = Promise.prototype.then;
 const FAILED_EPOCH_TRANSITION: unique symbol = Symbol(
   "FailedSqlCatalogEpochTransition",
@@ -392,21 +394,18 @@ function drainDetachedSettlement(result: unknown): void {
     return;
   }
   try {
-    Reflect.apply(INTRINSIC_PROMISE_THEN, result, [
+    const settlement = Reflect.apply(
+      INTRINSIC_PROMISE_RESOLVE,
+      INTRINSIC_PROMISE,
+      [result],
+    );
+    Reflect.apply(INTRINSIC_PROMISE_THEN, settlement, [
       undefined,
       IGNORE_DETACHED_REJECTION,
     ]);
-    return;
   } catch {
-    // Non-native thenables are assimilated through a fresh wrapper.
+    // The detached value is hostile and cannot be observed safely.
   }
-  const settlement = new Promise<unknown>((resolve) => {
-    resolve(result);
-  });
-  Reflect.apply(INTRINSIC_PROMISE_THEN, settlement, [
-    undefined,
-    IGNORE_DETACHED_REJECTION,
-  ]);
 }
 
 function cleanupSubscription(

src/vnext/relation-catalog-search-work.ts

@@ -1463,6 +1463,17 @@ function requestWork(
     return makeImmediateTicket(SUPERSEDED_OUTCOME);
   }
   if (captured.status !== "captured") {
+    const pending = effects();
+    const previous = owner.current;
+    if (previous) {
+      detachConsumerInto(
+        state,
+        previous,
+        SUPERSEDED_OUTCOME,
+        pending,
+      );
+    }
+    runEffects(state, pending);
     return makeImmediateTicket(
       unavailableOutcome(
         captured.reason === "inactive"