fix(providers): retain candidate cleanup until close succeeds (#10020)

src/application/binary/AnalysisProviderRegistry.test.ts

@@ -1529,7 +1529,7 @@ it("joins cancelled readiness finalization and shutdown before admitting another
 });
 
 it.each(["alpha", "beta"])(
-  "retains unknown owned-resource cleanup when %s has no completion signal",
+  "releases unflagged and flagged candidate receipts only after their owners confirm close (%s unflagged)",
   async (unknownOwner) => {
     const unknown = new ProviderAdapterError(unknownOwner, "close_binary", {
       userMessage: "cleanup completion unknown",
@@ -1552,7 +1552,75 @@ it.each(["alpha", "beta"])(
     released = true;
     expect(await registry.closeOwnedResources()).toEqual(ok(null));
     expect(tracked.cleanupIncomplete).toBe(false);
-    expect(closed.error.cleanupIncomplete).toBe(true);
+    expect(closed.error.cleanupIncomplete).toBe(false);
+  },
+);
+
+it.each([
+  ["returned", "physical"],
+  ["returned", "cancelled"],
+  ["thrown", "physical"],
+  ["thrown", "cancelled"],
+] as const)(
+  "retains an unflagged %s %s close failure until that candidate confirms completion",
+  async (delivery, kind) => {
+    const reason = { reason: "controlled caller cancellation" };
+    const physical = { reason: "controlled descriptor close failure" };
+    const original =
+      kind === "cancelled"
+        ? new AnalysisCancelledError("close_binary", { cause: reason })
+        : new ProviderAdapterError("alpha", "close_binary", {
+            cause: physical,
+            diagnostics: { reason: "descriptor close failed" },
+          });
+    const alpha = candidate("alpha").provider;
+    let released = false;
+    let inspections = 0;
+    let peerCloses = 0;
+    const registry = new AnalysisProviderRegistry([
+      {
+        ...alpha,
+        inspectAvailability: async (options) => {
+          inspections++;
+          return alpha.inspectAvailability(options);
+        },
+        closeOwnedResources: async () => {
+          if (released) return ok(null);
+          if (delivery === "thrown") throw original;
+          return err(original);
+        },
+      },
+      {
+        ...candidate("beta").provider,
+        closeOwnedResources: async () => {
+          peerCloses++;
+          return ok(null);
+        },
+      },
+    ]);
+    await withVerifierCleanup(async () => {
+      const closed = await registry.closeOwnedResources();
+      if (closed.ok) throw new Error("Expected incomplete candidate close");
+      expect(closed.error).toBeInstanceOf(ProviderCleanupError);
+      expect(closed.error.cause).toBe(original);
+      expect(closed.error.cleanupPrimary).toBe(original);
+      expect(original.cause).toBe(kind === "cancelled" ? reason : physical);
+      expect(registry.hasPendingOwnedResources()).toBe(true);
+      expect((await registry.select(DATABASE_TARGET, "alpha")).ok).toBe(false);
+      expect(inspections).toBe(0);
+      expect(peerCloses).toBe(2);
+      released = true;
+      expect(await registry.closeOwnedResources()).toEqual(ok(null));
+      expect(closed.error.cleanupIncomplete).toBe(false);
+      expect(registry.hasPendingOwnedResources()).toBe(false);
+      expect((await registry.select(DATABASE_TARGET, "alpha")).ok).toBe(true);
+      expect(inspections).toBe(1);
+    }, [
+      async () => {
+        released = true;
+        expect(await registry.closeOwnedResources()).toEqual(ok(null));
+      },
+    ]);
   },
 );
 
@@ -1686,3 +1754,185 @@ it("registers selection before async readiness can race shutdown", async () => {
   expect(calls).toBe(2);
   expect(registry.hasPendingOwnedResources()).toBe(false);
 });
+
+it.each(["typed", "untyped", "returned"] as const)(
+  "joins peer cleanup when a candidate close callback fails (%s)",
+  async (kind) => {
+    const failure = new ProviderCleanupError(
+      "alpha",
+      ["alpha profile reader"],
+      {
+        reason: "controlled close rejection",
+      },
+    );
+    let denied = true;
+    const rejection =
+      kind === "untyped" ? { reason: "opaque rejection" } : failure;
+    const closed: string[] = [];
+    const registry = new AnalysisProviderRegistry([
+      {
+        ...candidate("alpha").provider,
+        closeOwnedResources: async () => {
+          closed.push("alpha");
+          if (denied) {
+            if (kind === "returned") return err(failure);
+            throw rejection;
+          }
+          return ok(null);
+        },
+      },
+      {
+        ...candidate("beta").provider,
+        closeOwnedResources: async () => {
+          closed.push("beta");
+          return ok(null);
+        },
+      },
+    ]);
+    try {
+      const result = await registry.closeOwnedResources();
+      if (result.ok) throw new Error("Expected retained candidate cleanup");
+      if (kind !== "untyped") expect(result.error).toBe(failure);
+      else {
+        expect(result.error).toBeInstanceOf(ProviderCleanupError);
+        expect(result.error.cause).toBe(rejection);
+      }
+      expect(closed).toEqual(["alpha", "beta"]);
+      expect(registry.hasPendingOwnedResources()).toBe(true);
+      denied = false;
+      expect(await registry.closeOwnedResources()).toEqual(ok(null));
+      expect(result.error.cleanupIncomplete).toBe(false);
+      expect(registry.hasPendingOwnedResources()).toBe(false);
+    } finally {
+      denied = false;
+      await registry.closeOwnedResources();
+    }
+  },
+);
+
+it.each(["alpha", "beta"] as const)(
+  "retains each candidate receipt until its own close succeeds (%s first)",
+  async (first) => {
+    const failures = {
+      alpha: new ProviderCleanupError("alpha", ["alpha-reader"], {}),
+      beta: new ProviderCleanupError("beta", ["beta-reader"], {}),
+    };
+    const released = new Set<string>();
+    const calls: string[] = [];
+    const registry = new AnalysisProviderRegistry([
+      ...(["alpha", "beta"] as const).map((id) => ({
+        ...candidate(id).provider,
+        closeOwnedResources: async () => {
+          calls.push(id);
+          if (released.has(id)) return ok(null);
+          if (id === "alpha") throw failures.alpha;
+          return err(failures.beta);
+        },
+      })),
+      {
+        ...candidate("gamma").provider,
+        closeOwnedResources: async () => {
+          calls.push("gamma");
+          return ok(null);
+        },
+      },
+    ]);
+    try {
+      const initial = await registry.closeOwnedResources();
+      if (initial.ok) throw new Error("Expected both retained candidates");
+      expect(calls).toEqual(["alpha", "beta", "gamma"]);
+      expect(initial.error.cleanupResources).toEqual([
+        "alpha-reader",
+        "beta-reader",
+      ]);
+      expect(initial.error.cause).toBeInstanceOf(AggregateError);
+      if (!(initial.error.cause instanceof AggregateError))
+        throw new Error("Expected attributed failures");
+      expect(initial.error.cause.errors).toEqual([
+        failures.alpha,
+        failures.beta,
+      ]);
+      released.add(first);
+      const partial = await registry.closeOwnedResources();
+      expect(partial.ok).toBe(false);
+      const remaining = first === "alpha" ? "beta" : "alpha";
+      expect(failures[first].cleanupIncomplete).toBe(false);
+      expect(failures[remaining].cleanupIncomplete).toBe(true);
+      expect(initial.error.cleanupIncomplete).toBe(true);
+      expect(initial.error.cleanupResources).toEqual([`${remaining}-reader`]);
+      expect(registry.hasPendingOwnedResources()).toBe(true);
+      released.add(remaining);
+      expect(await registry.closeOwnedResources()).toEqual(ok(null));
+      expect(initial.error.cleanupIncomplete).toBe(false);
+      expect(registry.hasPendingOwnedResources()).toBe(false);
+    } finally {
+      released.add("alpha");
+      released.add("beta");
+      expect(await registry.closeOwnedResources()).toEqual(ok(null));
+    }
+  },
+);
+
+it("does not inspect a new selection after cancellation while its retained candidate cleanup settles", async () => {
+  const base = candidate("alpha").provider;
+  const failure = new ProviderCleanupError("alpha", ["profile-reader"], {});
+  let allowClose = false;
+  let inspections = 0;
+  let enter!: () => void;
+  let finish!: () => void;
+  const entered = new Promise<void>((resolve) => {
+    enter = resolve;
+  });
+  const gate = new Promise<void>((resolve) => {
+    finish = resolve;
+  });
+  const registry = new AnalysisProviderRegistry([
+    {
+      ...base,
+      inspectAvailability: () => {
+        inspections++;
+        return base.inspectAvailability();
+      },
+      closeOwnedResources: async () => {
+        if (!allowClose) throw failure;
+        enter();
+        await gate;
+        return ok(null);
+      },
+    },
+  ]);
+  const controller = new AbortController();
+  const reason = { request: "cancel while cleanup completes" };
+  let selecting: ReturnType<typeof registry.select> | undefined;
+  await withVerifierCleanup(async () => {
+    expect(await registry.closeOwnedResources()).toEqual(err(failure));
+    allowClose = true;
+    selecting = registry.select(DATABASE_TARGET, "alpha", {
+      signal: controller.signal,
+    });
+    await awaitAcknowledgement(entered, selecting, "candidate cleanup retry");
+    controller.abort(reason);
+    finish();
+    const result = await selecting;
+    expect(result).toMatchObject({
+      ok: false,
+      error: { _tag: "AnalysisCancelledError", cause: reason },
+    });
+    expect(inspections).toBe(0);
+    expect(failure.cleanupIncomplete).toBe(false);
+    expect(registry.hasPendingOwnedResources()).toBe(false);
+    expect((await registry.select(DATABASE_TARGET, "alpha")).ok).toBe(true);
+    expect(inspections).toBe(1);
+  }, [
+    () => {
+      allowClose = true;
+      finish();
+    },
+    async () => {
+      await selecting;
+    },
+    async () => {
+      expect(await registry.closeOwnedResources()).toEqual(ok(null));
+    },
+  ]);
+});

src/application/binary/AnalysisProviderRegistry.ts

@@ -132,6 +132,10 @@ export class AnalysisProviderRegistry {
     >
   >();
   readonly #selectionCleanup = new Set<AnalysisError>();
+  readonly #candidateCleanup = new Map<
+    AnalysisProviderCandidate,
+    Set<AnalysisError>
+  >();
   readonly #profileResources = new ArtifactResourceScope();
   readonly #profileOwner: RetainedResourceScope = {
     retain: (owner, resource) =>
@@ -256,8 +260,41 @@ export class AnalysisProviderRegistry {
     }
     for (const provider of this.#providers) {
       if (provider.closeOwnedResources === undefined) continue;
-      const result = await provider.closeOwnedResources();
-      if (!result.ok) failures.push(result.error);
+      let result: Result<null, unknown>;
+      try {
+        result = await provider.closeOwnedResources();
+      } catch (cause: unknown) {
+        result = err(cause);
+      }
+      if (result.ok) {
+        for (const failure of this.#candidateCleanup.get(provider) ?? [])
+          failure.resolveCleanup();
+        this.#candidateCleanup.delete(provider);
+      } else {
+        const failure =
+          result.error instanceof AnalysisError &&
+          result.error.cleanupIncomplete
+            ? result.error
+            : new ProviderCleanupError(
+                provider.identity().id,
+                ["provider-owned resources"],
+                {
+                  reason: "provider resource close did not confirm completion",
+                },
+                {
+                  cause: result.error,
+                  cleanupPrimary:
+                    result.error instanceof AnalysisError
+                      ? result.error
+                      : undefined,
+                },
+              );
+        failures.push(failure);
+        const retained = this.#candidateCleanup.get(provider) ?? new Set();
+        retained.add(failure);
+        this.#candidateCleanup.set(provider, retained);
+        this.#selectionCleanup.add(failure);
+      }
     }
     for (const failure of this.#selectionCleanup) {
       if (failure.cleanupIncomplete) failures.push(failure);