fix(matrix): await persistence shutdown (#129605)

* fix(matrix): await persistence shutdown * fix(matrix): join startup during shutdown * fix(matrix): bound startup retirement * fix(matrix): fence late retirement ownership * fix(matrix): centralize generation retirement
2026-08-25 20:05:46 -06:00 · 2026-08-26 07:19:16 +08:00
parent c2e1aa2047
commit d7c5771684
9 changed files with 582 additions and 166 deletions
@@ -46,7 +46,7 @@ export function createMockMatrixClient(): MatrixClient {
    start: vi.fn(async () => undefined),
    stop: vi.fn(() => undefined),
    stopAndPersist: vi.fn(async () => undefined),
-    stopWithoutPersist: vi.fn(() => undefined),
+    stopWithoutPersist: vi.fn(async () => undefined),
  } as unknown as MatrixClient;
 }

@@ -50,7 +50,7 @@ function createMockClient(name: string, callOrder: string[] = []) {
    stopAndPersist: vi.fn(async () => {
      callOrder.push("persist");
    }),
-    stopWithoutPersist: vi.fn(() => {
+    stopWithoutPersist: vi.fn(async () => {
      callOrder.push("discard");
    }),
    drainPendingDecryptions: vi.fn(async (reason: string) => {
@@ -87,6 +87,13 @@ async function expectMatrixStartupAbort(promise: Promise<unknown>): Promise<void
  });
 }

+async function expectPending(promise: Promise<unknown>): Promise<void> {
+  const settled = vi.fn();
+  void promise.then(settled, settled);
+  await Promise.resolve();
+  expect(settled).not.toHaveBeenCalled();
+}
+
 describe("shared Matrix client generations", () => {
  beforeAll(async () => {
    ({ acquireSharedMatrixClient, stopSharedClientForAccount } = await import("./shared.js"));
@@ -231,12 +238,7 @@ describe("shared Matrix client generations", () => {
    });
    const lateRelease = monitor.release({ mode: "persist" });
    expect(monitor.release({ mode: "discard" })).toBe(lateRelease);
-    let lateReleaseSettled = false;
-    void lateRelease.then(() => {
-      lateReleaseSettled = true;
-    });
-    await Promise.resolve();
-    expect(lateReleaseSettled).toBe(false);
+    await expectPending(lateRelease);

    waitForTasks.resolve();
    await Promise.all([forcedRetirement, lateRelease]);
@@ -391,7 +393,12 @@ describe("shared Matrix client generations", () => {
  it("bounds non-cooperative transient drain and replaces after every late release", async () => {
    vi.useFakeTimers();
    const callOrder: string[] = [];
+    const discard = createDeferred<void>();
    const client = createMockClient("main", callOrder);
+    client.stopWithoutPersist.mockImplementation(async () => {
+      callOrder.push("discard");
+      await discard.promise;
+    });
    const replacementClient = createMockClient("replacement");
    createMatrixClientMock.mockResolvedValueOnce(client).mockResolvedValueOnce(replacementClient);
    const auth = authFor("main");
@@ -421,30 +428,41 @@ describe("shared Matrix client generations", () => {
    expect(client.stopWithoutPersist).not.toHaveBeenCalled();

    await vi.advanceTimersByTimeAsync(1);
+    await vi.waitFor(() => {
+      expect(client.stopWithoutPersist).toHaveBeenCalledTimes(1);
+    });
+    await expect(stopSharedClientForAccount(auth)).rejects.toMatchObject({
+      message: "Matrix transient leases did not drain within 5000ms",
+    });
+
+    discard.resolve();
    await expect(retirementError).resolves.toMatchObject({
      message: "Matrix transient leases did not drain within 5000ms",
    });
-    expect(firstTransient.abortSignal.aborted).toBe(true);
    expect(finalTransient.abortSignal.aborted).toBe(true);
    expect(client.stopWithoutPersist).toHaveBeenCalledTimes(1);
    expect(client.stopAndPersist).not.toHaveBeenCalled();
    await expect(acquireSharedMatrixClient({ auth })).rejects.toMatchObject({
      message: "Matrix transient leases did not drain within 5000ms",
    });
-    expect(createMatrixClientMock).toHaveBeenCalledTimes(1);

    const firstLateRelease = firstTransient.release({ mode: "persist" });
    const duplicateLateRelease = firstTransient.release({ mode: "persist" });
    expect(duplicateLateRelease).toBe(firstLateRelease);
    await firstLateRelease;
-    expect(client.stopWithoutPersist).toHaveBeenCalledTimes(1);
-    expect(client.stopAndPersist).not.toHaveBeenCalled();
    await expect(acquireSharedMatrixClient({ auth })).rejects.toMatchObject({
      message: "Matrix transient leases did not drain within 5000ms",
    });
-    expect(createMatrixClientMock).toHaveBeenCalledTimes(1);
+    await expect(stopSharedClientForAccount(auth)).rejects.toMatchObject({
+      message: "Matrix transient leases did not drain within 5000ms",
+    });

-    await finalTransient.release({ mode: "persist" });
+    const finalLateRelease = finalTransient.release({ mode: "persist" });
+    const finalRepeatedForce = stopSharedClientForAccount(auth);
+    await finalLateRelease;
+    await expect(finalRepeatedForce).rejects.toMatchObject({
+      message: "Matrix transient leases did not drain within 5000ms",
+    });

    const [firstReplacement, secondReplacement] = await Promise.all([
      acquireSharedMatrixClient({ auth, startClient: false }),
@@ -462,7 +480,7 @@ describe("shared Matrix client generations", () => {
    const cause = new Error("monitor cleanup failed");
    const client = createMockClient("main");
    createMatrixClientMock.mockResolvedValue(client);
-    const auth = authFor("main");
+    const auth = authFor("late-monitor-cleanup");
    const monitor = await acquireSharedMatrixClient({
      auth,
      role: "monitor",
@@ -499,7 +517,7 @@ describe("shared Matrix client generations", () => {
      }
    });
    createMatrixClientMock.mockResolvedValue(client);
-    const auth = authFor("main");
+    const auth = authFor("poisoned-decryption-drain");
    const monitor = await acquireSharedMatrixClient({
      auth,
      role: "monitor",
@@ -735,7 +753,7 @@ describe("shared Matrix client generations", () => {
    const cause = new Error("monitor cleanup failed");
    const client = createMockClient("main");
    createMatrixClientMock.mockResolvedValue(client);
-    const auth = authFor("main");
+    const auth = authFor("monitor-cleanup-failure");
    const monitor = await acquireSharedMatrixClient({
      auth,
      role: "monitor",
@@ -748,6 +766,8 @@ describe("shared Matrix client generations", () => {
    await expect(monitor.release({ mode: "persist" })).rejects.toBe(cause);
    expect(client.stopWithoutPersist).toHaveBeenCalledTimes(1);
    await expect(acquireSharedMatrixClient({ auth })).rejects.toBe(cause);
+    await expect(stopSharedClientForAccount(auth)).rejects.toBe(cause);
+    await expect(acquireSharedMatrixClient({ auth })).rejects.toBe(cause);
    expect(createMatrixClientMock).toHaveBeenCalledTimes(1);
  });

@@ -796,26 +816,101 @@ describe("shared Matrix client generations", () => {
    expect(client.stopWithoutPersist).toHaveBeenCalledTimes(1);
  });

-  it("preserves and propagates an earlier persist requirement when final lease discards", async () => {
-    const cause = new Error("crypto persist failed");
+  it("keeps a discarded generation unavailable until async cleanup settles", async () => {
+    const discard = createDeferred<void>();
+    const client = createMockClient("main");
+    const replacementClient = createMockClient("replacement");
+    client.stopWithoutPersist.mockReturnValue(discard.promise);
+    createMatrixClientMock.mockResolvedValueOnce(client).mockResolvedValueOnce(replacementClient);
+    const auth = authFor("main");
+    const lease = await acquireSharedMatrixClient({ auth, startClient: false });
+
+    const release = lease.release({ mode: "discard" });
+    await vi.waitFor(() => {
+      expect(client.stopWithoutPersist).toHaveBeenCalledTimes(1);
+    });
+    const replacementPromise = acquireSharedMatrixClient({ auth, startClient: false });
+    await Promise.resolve();
+    expect(createMatrixClientMock).toHaveBeenCalledTimes(1);
+
+    discard.resolve();
+    await release;
+    const replacement = await replacementPromise;
+    expect(replacement.client).toBe(replacementClient);
+    expect(createMatrixClientMock).toHaveBeenCalledTimes(2);
+    await replacement.release({ mode: "discard" });
+  });
+
+  it.each([
+    {
+      name: "normal discard",
+      mode: "discard" as const,
+      configure: (client: ReturnType<typeof createMockClient>, failure: Error) => {
+        client.stopWithoutPersist.mockRejectedValue(failure);
+      },
+    },
+    {
+      name: "strict persistence fallback",
+      mode: "persist" as const,
+      configure: (client: ReturnType<typeof createMockClient>, failure: Error) => {
+        client.stopAndPersist.mockRejectedValue(new Error("crypto persist failed"));
+        client.stopWithoutPersist.mockRejectedValue(failure);
+      },
+    },
+    {
+      name: "final-drain discard",
+      mode: "persist" as const,
+      configure: (client: ReturnType<typeof createMockClient>, failure: Error) => {
+        client.drainPendingDecryptions
+          .mockResolvedValueOnce(undefined)
+          .mockRejectedValueOnce(new Error("final decryption drain timed out"));
+        client.stopWithoutPersist.mockRejectedValue(failure);
+      },
+    },
+  ])("retains a generation when $name shutdown fails", async ({ name, mode, configure }) => {
+    const failure = new Error(`${name} shutdown failed`);
+    const client = createMockClient("main");
+    configure(client, failure);
+    createMatrixClientMock.mockResolvedValue(client);
+    const auth = authFor(`${name}-failure`);
+    const lease = await acquireSharedMatrixClient({ auth, startClient: false });
+
+    await expect(lease.release({ mode })).rejects.toBe(failure);
+    await expect(acquireSharedMatrixClient({ auth, startClient: false })).rejects.toBe(failure);
+    expect(createMatrixClientMock).toHaveBeenCalledTimes(1);
+  });
+
+  it("awaits discard fallback before surfacing strict persistence failure", async () => {
+    const persistFailure = new Error("crypto persist failed");
+    const persist = createDeferred<void>();
+    const discard = createDeferred<void>();
    const firstClient = createMockClient("first");
    const replacementClient = createMockClient("replacement");
-    firstClient.stopAndPersist.mockRejectedValue(cause);
+    firstClient.stopAndPersist.mockReturnValue(persist.promise);
+    firstClient.stopWithoutPersist.mockReturnValue(discard.promise);
    createMatrixClientMock
      .mockResolvedValueOnce(firstClient)
      .mockResolvedValueOnce(replacementClient);
-    const auth = authFor("main");
-    const first = await acquireSharedMatrixClient({ auth, startClient: false });
-    const final = await acquireSharedMatrixClient({ auth, startClient: false });
+    const auth = authFor("strict-persist-failure");
+    const lease = await acquireSharedMatrixClient({ auth, startClient: false });

-    await first.release({ mode: "persist" });
-    expect(firstClient.stopAndPersist).not.toHaveBeenCalled();
-    await expect(final.release({ mode: "discard" })).rejects.toBe(cause);
-    expect(firstClient.stopWithoutPersist).not.toHaveBeenCalled();
+    const releaseError = expect(lease.release({ mode: "persist" })).rejects.toBe(persistFailure);
+    persist.reject(persistFailure);
+    await vi.waitFor(() => {
+      expect(firstClient.stopWithoutPersist).toHaveBeenCalledTimes(1);
+    });
+    const blockedAcquire = acquireSharedMatrixClient({ auth, startClient: false });
+    await expect(blockedAcquire).rejects.toBe(persistFailure);
+    expect(createMatrixClientMock).toHaveBeenCalledTimes(1);
+    const forcedRetirement = stopSharedClientForAccount(auth);
+    await expectPending(forcedRetirement);

+    discard.resolve();
+    await releaseError;
+    await expect(forcedRetirement).resolves.toBeUndefined();
    const replacement = await acquireSharedMatrixClient({ auth, startClient: false });
    expect(replacement.client).toBe(replacementClient);
-    await replacement.release();
+    await replacement.release({ mode: "discard" });
  });

  it("discards and replaces a generation when the final decryption drain fails", async () => {
@@ -841,7 +936,7 @@ describe("shared Matrix client generations", () => {
    await replacement.release({ mode: "discard" });
  });

-  it("aborts a first starter and waiter during forced retirement without late reuse", async () => {
+  it("joins an in-flight startup during forced retirement", async () => {
    const start = createDeferred<void>();
    const firstClient = createMockClient("first");
    const replacementClient = createMockClient("replacement");
@@ -866,19 +961,134 @@ describe("shared Matrix client generations", () => {
    const ownerAbort = expectMatrixStartupAbort(ownerStart);
    const waiterAbort = expectMatrixStartupAbort(waiterStart);

-    await Promise.all([stopSharedClientForAccount(auth), ownerAbort, waiterAbort]);
+    const retirement = stopSharedClientForAccount(auth);
+    await Promise.all([ownerAbort, waiterAbort]);
    expect(callerAbort.signal.aborted).toBe(false);
    expect(startupSignal?.aborted).toBe(true);
    expect(owner.abortSignal.aborted).toBe(true);
    expect(waiter.abortSignal.aborted).toBe(true);
+    expect(firstClient.stopAndPersist).not.toHaveBeenCalled();
+
+    await expectPending(retirement);

    start.resolve();
-    await Promise.resolve();
+    await retirement;
+    expect(firstClient.stopAndPersist).toHaveBeenCalledTimes(1);
    const replacement = await acquireSharedMatrixClient({ auth, startClient: false });
    expect(replacement.client).toBe(replacementClient);
    await replacement.release({ mode: "discard" });
  });

+  it("bounds forced retirement while startup remains stuck and fences late cleanup", async () => {
+    vi.useFakeTimers();
+    const start = createDeferred<void>();
+    const discard = createDeferred<void>();
+    const firstClient = createMockClient("first");
+    const replacementClient = createMockClient("replacement");
+    firstClient.start.mockReturnValue(start.promise);
+    firstClient.stopWithoutPersist.mockReturnValue(discard.promise);
+    createMatrixClientMock
+      .mockResolvedValueOnce(firstClient)
+      .mockResolvedValueOnce(replacementClient);
+    const auth = authFor("main");
+    const lease = await acquireSharedMatrixClient({ auth, startClient: false });
+    const startup = lease.start();
+    await vi.waitFor(() => {
+      expect(firstClient.start).toHaveBeenCalledTimes(1);
+    });
+
+    const retirementError = stopSharedClientForAccount(auth).then(
+      () => null,
+      (error: unknown) => error,
+    );
+    await expectMatrixStartupAbort(startup);
+    await vi.advanceTimersByTimeAsync(4_999);
+    expect(firstClient.stopWithoutPersist).not.toHaveBeenCalled();
+
+    await vi.advanceTimersByTimeAsync(1);
+    await expect(retirementError).resolves.toMatchObject({
+      message: "Matrix client startup did not settle within 5000ms during retirement",
+    });
+    await expect(acquireSharedMatrixClient({ auth, startClient: false })).rejects.toMatchObject({
+      message: "Matrix client startup did not settle within 5000ms during retirement",
+    });
+    expect(createMatrixClientMock).toHaveBeenCalledTimes(1);
+
+    start.resolve();
+    await vi.waitFor(() => {
+      expect(firstClient.stopWithoutPersist).toHaveBeenCalledTimes(1);
+    });
+    expect(firstClient.stopAndPersist).not.toHaveBeenCalled();
+    await expect(acquireSharedMatrixClient({ auth, startClient: false })).rejects.toMatchObject({
+      message: "Matrix client startup did not settle within 5000ms during retirement",
+    });
+    expect(createMatrixClientMock).toHaveBeenCalledTimes(1);
+
+    discard.resolve();
+    await discard.promise;
+    let replacement: Awaited<ReturnType<typeof acquireSharedMatrixClient>> | undefined;
+    await vi.waitFor(async () => {
+      replacement = await acquireSharedMatrixClient({ auth, startClient: false });
+    });
+    if (!replacement) {
+      throw new Error("expected replacement Matrix client");
+    }
+    expect(replacement.client).toBe(replacementClient);
+    expect(createMatrixClientMock).toHaveBeenCalledTimes(2);
+    await replacement.release({ mode: "discard" });
+  });
+
+  it("retires monitor ownership when late startup discard fails", async () => {
+    vi.useFakeTimers();
+    const start = createDeferred<void>();
+    const monitorTasks = createDeferred<void>();
+    const discardFailure = new Error("late discard failed");
+    const client = createMockClient("first");
+    client.start.mockReturnValue(start.promise);
+    client.stopWithoutPersist.mockRejectedValue(discardFailure);
+    createMatrixClientMock.mockResolvedValue(client);
+    const auth = authFor("late-discard-failure");
+    const monitor = await acquireSharedMatrixClient({
+      auth,
+      role: "monitor",
+      startClient: false,
+    });
+    const monitorRetirement = createMonitorRetirement([]);
+    monitorRetirement.waitForTasks.mockReturnValue(monitorTasks.promise);
+    monitor.registerMonitorRetirement(monitorRetirement);
+    const startup = monitor.start();
+
+    await vi.waitFor(() => {
+      expect(client.start).toHaveBeenCalledTimes(1);
+    });
+    const retirementError = stopSharedClientForAccount(auth).then(
+      () => null,
+      (error: unknown) => error,
+    );
+    await expectMatrixStartupAbort(startup);
+    await vi.advanceTimersByTimeAsync(5_000);
+    await expect(retirementError).resolves.toMatchObject({
+      message: "Matrix client startup did not settle within 5000ms during retirement",
+    });
+    expect(monitorRetirement.closeTaskAdmission).toHaveBeenCalledTimes(1);
+    expect(monitorRetirement.detachListeners).toHaveBeenCalledTimes(1);
+    expect(monitorRetirement.waitForTasks).toHaveBeenCalledTimes(1);
+
+    start.resolve();
+    await vi.waitFor(() => {
+      expect(client.stopWithoutPersist).toHaveBeenCalledTimes(1);
+    });
+    monitorTasks.resolve();
+    await vi.waitFor(() => {
+      expect(monitorRetirement.cleanup).toHaveBeenCalledTimes(1);
+    });
+
+    await expect(acquireSharedMatrixClient({ auth, startClient: false })).rejects.toBe(
+      discardFailure,
+    );
+    expect(createMatrixClientMock).toHaveBeenCalledTimes(1);
+  });
+
  it("does not let one aborted startup waiter remove another lease", async () => {
    const client = createMockClient("main");
    const start = createDeferred<void>();
@@ -6,7 +6,7 @@ import { createLazyRuntimeModule } from "openclaw/plugin-sdk/lazy-runtime";
 import type { CoreConfig } from "../../types.js";
 import type { MatrixClient } from "../sdk.js";
 import { LogService } from "../sdk/logger.js";
-import { awaitMatrixStartupWithAbort } from "../startup-abort.js";
+import { awaitMatrixStartupWithAbort, throwIfMatrixStartupAborted } from "../startup-abort.js";
 import { resolveMatrixAuth, resolveMatrixAuthContext } from "./config.js";
 import type { MatrixAuth } from "./types.js";

@@ -15,7 +15,7 @@ const loadMatrixCreateClientDeps = createLazyRuntimeModule(() =>
    createMatrixClient: runtime.createMatrixClient,
  })),
 );
-const MATRIX_TRANSIENT_LEASE_DRAIN_TIMEOUT_MS = 5_000;
+const MATRIX_RETIREMENT_DRAIN_TIMEOUT_MS = 5_000;

 export type MatrixClientLeaseRole = "monitor" | "transient";
 export type MatrixClientReleaseMode = "stop" | "persist" | "discard";
@@ -36,14 +36,7 @@ export type SharedMatrixClientLease = {
  release: (params?: { mode?: MatrixClientReleaseMode }) => Promise<void>;
 };

-type SharedMatrixClientPhase =
-  | "open"
-  | "quiescing"
-  | "closing"
-  | "late-drain"
-  | "late-drain-stopped";
-
-type PoisonDisposition = "replace-after-stop" | "replace-after-late-drain" | "retain";
+type SharedMatrixClientPhase = "open" | "quiescing" | "closing" | "late-drain";

 type SharedMatrixClientLeaseState = {
  abortController: AbortController;
@@ -140,12 +133,6 @@ function deleteSharedClientState(state: SharedMatrixClientState): void {
  sharedClientPromises.delete(state.key);
 }

-function deleteSharedClientStateAfterLateDrain(state: SharedMatrixClientState): void {
-  if (state.phase === "late-drain-stopped" && state.leases.size === 0) {
-    deleteSharedClientState(state);
-  }
-}
-
 async function ensureSharedClientStarted(
  state: SharedMatrixClientState,
  abortSignal?: AbortSignal,
@@ -171,7 +158,8 @@ async function ensureSharedClientStarted(
      }
    }

-    await awaitMatrixStartupWithAbort(state.client.start({ abortSignal }), abortSignal);
+    await state.client.start({ abortSignal });
+    throwIfMatrixStartupAborted(abortSignal);
    state.started = true;
  })();
  const guardedStart = startPromise.finally(() => {
@@ -322,26 +310,27 @@ function forceReleaseLeases(
  state.noLeases.resolve();
 }

-async function waitForLeaseDrain(state: SharedMatrixClientState): Promise<void> {
-  if (state.leases.size === 0) {
+async function waitForRetirementDrain(
+  state: SharedMatrixClientState,
+  task: Promise<unknown>,
+  isPending: () => boolean,
+  timeoutMessage: string,
+): Promise<void> {
+  if (!isPending()) {
    return;
  }
  let deadline: NodeJS.Timeout | undefined;
  try {
    await Promise.race([
-      state.noLeases.promise,
+      task,
      new Promise<never>((_, reject) => {
        deadline = setTimeout(() => {
-          if (state.leases.size === 0) {
+          if (!isPending()) {
            return;
          }
          state.phase = "late-drain";
-          reject(
-            new Error(
-              `Matrix transient leases did not drain within ${MATRIX_TRANSIENT_LEASE_DRAIN_TIMEOUT_MS}ms`,
-            ),
-          );
-        }, MATRIX_TRANSIENT_LEASE_DRAIN_TIMEOUT_MS);
+          reject(new Error(timeoutMessage));
+        }, MATRIX_RETIREMENT_DRAIN_TIMEOUT_MS);
        deadline.unref?.();
      }),
    ]);
@@ -361,8 +350,40 @@ function beginGenerationRetirement(params: {
    return state.retirementPromise;
  }
  state.phase = "quiescing";
-  state.retirementPromise = Promise.resolve().then(async () => {
-    let poisonDisposition: PoisonDisposition = "replace-after-stop";
+  const result = createDeferred<void>();
+  state.retirementPromise = result.promise;
+  const owner = Promise.resolve().then(async () => {
+    const startup = state.startPromise;
+    if (startup) {
+      try {
+        await waitForRetirementDrain(
+          state,
+          startup.catch(() => undefined),
+          () => state.startPromise === startup,
+          `Matrix client startup did not settle within ${MATRIX_RETIREMENT_DRAIN_TIMEOUT_MS}ms during retirement`,
+        );
+      } catch (error) {
+        state.poisonError = toRetirementError(error);
+        result.reject(state.poisonError);
+        const outcomes = await Promise.allSettled([
+          startup
+            .catch(() => undefined)
+            .then(async () => {
+              state.started = false;
+              await state.client.stopWithoutPersist();
+            }),
+          retireMonitorLeases(state, params.monitorLeases ?? []),
+          state.noLeases.promise,
+        ]);
+        const failure = outcomes.find((outcome) => outcome.status === "rejected");
+        if (failure) {
+          state.poisonError = toRetirementError(failure.reason);
+        } else {
+          deleteSharedClientState(state);
+        }
+        throw state.poisonError;
+      }
+    }
    try {
      await state.client.quiesceSync();
      state.started = false;
@@ -371,67 +392,71 @@ function beginGenerationRetirement(params: {
      state.poisonError = toRetirementError(error);
    }

+    let monitorRetired = true;
    try {
      await retireMonitorLeases(state, params.monitorLeases ?? []);
    } catch (error) {
      state.poisonError ??= toRetirementError(error);
-      poisonDisposition = "retain";
+      monitorRetired = false;
    }

    state.phase = "closing";
+    let lateLeaseDrain: Promise<void> | null = null;
    try {
-      await waitForLeaseDrain(state);
+      await waitForRetirementDrain(
+        state,
+        state.noLeases.promise,
+        () => state.leases.size > 0,
+        `Matrix transient leases did not drain within ${MATRIX_RETIREMENT_DRAIN_TIMEOUT_MS}ms`,
+      );
    } catch (error) {
      state.poisonError ??= toRetirementError(error);
-      if (poisonDisposition !== "retain") {
-        poisonDisposition = "replace-after-late-drain";
-      }
+      result.reject(state.poisonError);
+      lateLeaseDrain = state.noLeases.promise;
    }

-    if (state.poisonError) {
-      const decryptionsDrained = await state.client
-        .drainPendingDecryptions("matrix poisoned client shutdown")
-        .then(
+    let failure = state.poisonError;
+    let canDelete = monitorRetired;
+    if (failure) {
+      canDelete =
+        (await state.client.drainPendingDecryptions("matrix poisoned client shutdown").then(
          () => true,
          () => false,
-        );
-      state.client.stopWithoutPersist();
-      if (decryptionsDrained) {
-        if (poisonDisposition === "replace-after-stop") {
-          deleteSharedClientState(state);
-        } else if (poisonDisposition === "replace-after-late-drain") {
-          // The timeout cannot revoke ownership. Keep the stopped generation keyed
-          // until every operation that crossed the deadline genuinely returns.
-          state.phase = "late-drain-stopped";
-          deleteSharedClientStateAfterLateDrain(state);
-        }
+        )) && canDelete;
+    } else {
+      try {
+        await state.client.drainPendingDecryptions("matrix shared client final shutdown");
+      } catch (error) {
+        failure = state.poisonError = toRetirementError(error);
      }
-      throw state.poisonError;
    }

-    try {
-      await state.client.drainPendingDecryptions("matrix shared client final shutdown");
-    } catch (error) {
-      state.poisonError = toRetirementError(error);
+    let discard = failure !== null || state.releaseMode === "discard";
+    if (!discard) {
      try {
-        state.client.stopWithoutPersist();
-      } finally {
-        deleteSharedClientState(state);
-      }
-      throw state.poisonError;
-    }
-    try {
-      if (state.releaseMode === "persist") {
        await state.client.stopAndPersist();
-      } else if (state.releaseMode === "discard") {
-        state.client.stopWithoutPersist();
-      } else {
-        await state.client.stopAndPersist().catch(() => state.client.stopWithoutPersist());
+      } catch (error) {
+        discard = true;
+        if (state.releaseMode === "persist") {
+          failure = state.poisonError = toRetirementError(error);
+        }
      }
-    } finally {
+    }
+    if (discard) {
+      await state.client.stopWithoutPersist().catch((error: unknown) => {
+        failure = state.poisonError = toRetirementError(error);
+        canDelete = false;
+      });
+    }
+    await lateLeaseDrain;
+    if (canDelete) {
      deleteSharedClientState(state);
    }
+    if (failure) {
+      throw failure;
+    }
  });
+  void owner.then(result.resolve, result.reject);
  abortTransientLeases(state);
  return state.retirementPromise;
 }
@@ -492,9 +517,8 @@ function createSharedMatrixClientLease(
        state.noLeases.resolve();
      }

-      if (state.phase === "late-drain" || state.phase === "late-drain-stopped") {
+      if (state.phase === "late-drain") {
        leaseState.releasePromise = Promise.resolve();
-        deleteSharedClientStateAfterLateDrain(state);
        return leaseState.releasePromise;
      }

@@ -540,24 +564,21 @@ export async function acquireSharedMatrixClient(
 }

 async function forceRetireState(state: SharedMatrixClientState): Promise<void> {
+  if (state.phase === "late-drain") {
+    throw state.poisonError ?? new Error("Matrix client generation is still retiring");
+  }
  state.releaseMode = mergeReleaseMode(state.releaseMode, "stop");
  const retirementPromise = beginGenerationRetirement({
    state,
    monitorLeases: Array.from(state.leases).filter((lease) => lease.role === "monitor"),
  });
  forceReleaseLeases(state, retirementPromise);
-  if (state.poisonError) {
-    await retirementPromise.catch(() => undefined);
-    deleteSharedClientState(state);
-    return;
-  }
-  await retirementPromise.catch((error: unknown) => {
-    if (!state.poisonError) {
-      throw error;
+  try {
+    await retirementPromise;
+  } catch (error) {
+    if (sharedClientStates.get(state.key) === state) {
+      throw state.poisonError ?? error;
    }
-  });
-  if (state.poisonError) {
-    deleteSharedClientState(state);
  }
 }

@@ -12,15 +12,22 @@ import { EventStatus } from "matrix-js-sdk/lib/models/event-status.js";
 import { SyncApi, SyncState } from "matrix-js-sdk/lib/sync.js";
 import { createDeferred } from "openclaw/plugin-sdk/extension-shared";
 import { resetPluginStateStoreForTests } from "openclaw/plugin-sdk/plugin-state-test-runtime";
+import { useAutoCleanupTempDirTracker } from "openclaw/plugin-sdk/test-env";
 // Matrix tests cover sdk plugin behavior.
 import { createRequireRecord } from "openclaw/plugin-sdk/test-fixtures";
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import { installMatrixTestRuntime } from "../test-runtime.js";
 import type { CoreConfig } from "../types.js";
-import { readMatrixRecoveryKeyStateForPath } from "./crypto-state-store.js";
+import {
+  readMatrixIdbSnapshotJson,
+  readMatrixRecoveryKeyStateForPath,
+} from "./crypto-state-store.js";
 import { MatrixDecryptBridge } from "./sdk/decrypt-bridge.js";
+import { clearAllIndexedDbState } from "./sdk/idb-persistence.test-helpers.js";
+import { LogService } from "./sdk/logger.js";

 const createSharedMatrixClientMock = vi.hoisted(() => vi.fn());
+const tempDirs = useAutoCleanupTempDirTracker(afterEach);

 vi.mock("./client/create-client.js", () => ({
  createMatrixClient: createSharedMatrixClientMock,
@@ -1525,6 +1532,34 @@ describe("MatrixClient request hardening", () => {
    });
  });

+  it("single-flights concurrent shutdown after discard starts", async () => {
+    const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-matrix-sdk-discard-"));
+    clearMatrixSyncApiForNeverStartedClient();
+
+    try {
+      const client = new MatrixClient("https://matrix.example.org", "token", {
+        storageRootDir: tempDir,
+      });
+      const store = lastCreateClientOpts?.store as
+        | { discardPendingSyncCursorPersistence: () => void }
+        | undefined;
+      if (!store) {
+        throw new Error("expected Matrix sync store");
+      }
+      const discardSpy = vi.spyOn(store, "discardPendingSyncCursorPersistence");
+
+      const first = client.stopWithoutPersist();
+      const second = client.stopWithoutPersist();
+      const persist = client.stopAndPersist();
+      await Promise.all([first, second, persist]);
+
+      expect(discardSpy).toHaveBeenCalledTimes(1);
+      expect(matrixJsClient.stopClient).toHaveBeenCalledTimes(1);
+    } finally {
+      fs.rmSync(tempDir, { recursive: true, force: true });
+    }
+  });
+
  it("arms and removes the STOPPED waiter around protected classic sync stop", async () => {
    const client = new MatrixClient("https://matrix.example.org", "token");
    await client.start();
@@ -2042,7 +2077,7 @@ describe("MatrixClient event bridge", () => {
    rejectSdkDecryption?.(new Error("late SDK decryption failure"));
    await Promise.resolve();
    expect(matrixJsClient.stopClient).not.toHaveBeenCalled();
-    client.stopWithoutPersist();
+    await client.stopWithoutPersist();
  });

  it("retries failed decryptions immediately on crypto key update signals", async () => {
@@ -2289,7 +2324,7 @@ describe("MatrixClient event bridge", () => {
      expect(encrypted.attemptDecryption).toHaveBeenCalledTimes(1);
      expect(delivered).toEqual(["m.room.message"]);
    } finally {
-      client.stopWithoutPersist();
+      await client.stopWithoutPersist();
    }
  });

@@ -2323,7 +2358,7 @@ describe("MatrixClient event bridge", () => {
    await vi.advanceTimersByTimeAsync(200_000);
    expect(encrypted.attemptDecryption).toHaveBeenCalledTimes(8);

-    client.stopWithoutPersist();
+    await client.stopWithoutPersist();
    encrypted.attemptDecryption.mockClear();
    encrypted.onAttemptDecryption(() => {
      encrypted.markDecrypted({
@@ -2397,7 +2432,7 @@ describe("MatrixClient event bridge", () => {
      await Promise.resolve();
      expect(delivered).toEqual(["m.room.message"]);
    } finally {
-      client.stopWithoutPersist();
+      await client.stopWithoutPersist();
    }
  });

@@ -2576,7 +2611,7 @@ describe("MatrixClient event bridge", () => {
    const client = new MatrixClient("https://matrix.example.org", "token");

    await client.start();
-    client.stopWithoutPersist();
+    await client.stopWithoutPersist();

    await expect(client.start()).rejects.toThrow(
      "Matrix client has been fully stopped and cannot be restarted; acquire a new shared client generation",
@@ -2635,6 +2670,44 @@ describe("MatrixClient crypto bootstrapping", () => {
    });
  });

+  it("does not persist or start sync when startup aborts during crypto initialization", async () => {
+    resetPluginStateStoreForTests();
+    installMatrixTestRuntime();
+    const tempDir = tempDirs.make("matrix-idb-startup-abort-");
+    const databasePrefix = "openclaw-matrix-startup-abort";
+    const initCrypto = createDeferred<void>();
+    matrixJsClient.initRustCrypto.mockReturnValue(initCrypto.promise);
+    const databasesSpy = vi.spyOn(indexedDB, "databases");
+    const abortController = new AbortController();
+
+    try {
+      const client = new MatrixClient("https://matrix.example.org", "token", {
+        encryption: true,
+        idbSnapshotPath: path.join(tempDir, "crypto-idb-snapshot.json"),
+        cryptoDatabasePrefix: databasePrefix,
+      });
+      const startup = client.start({ abortSignal: abortController.signal });
+
+      await vi.waitFor(() => {
+        expect(matrixJsClient.initRustCrypto).toHaveBeenCalledTimes(1);
+      });
+      abortController.abort();
+      expect(matrixJsClient.startClient).not.toHaveBeenCalled();
+      expect(databasesSpy).not.toHaveBeenCalled();
+
+      initCrypto.resolve();
+      await expectAbortError(startup);
+      expect(readMatrixIdbSnapshotJson(tempDir)).toBeNull();
+      expect(databasesSpy).not.toHaveBeenCalled();
+      expect(matrixJsClient.startClient).not.toHaveBeenCalled();
+    } finally {
+      initCrypto.resolve();
+      databasesSpy.mockRestore();
+      await clearAllIndexedDbState({ databasePrefix });
+      resetPluginStateStoreForTests();
+    }
+  });
+
  it("bootstraps cross-signing with setupNewCrossSigning enabled", async () => {
    const bootstrapCrossSigning = vi.fn(async () => {});
    matrixJsClient.getCrypto = vi.fn(() => ({
@@ -2978,28 +3051,61 @@ describe("MatrixClient crypto bootstrapping", () => {
    });
  });

-  it("schedules periodic crypto snapshot persistence", async () => {
-    const databasesSpy = vi.spyOn(indexedDB, "databases").mockResolvedValue([]);
+  it("awaits and cancels active periodic crypto persistence during discard shutdown", async () => {
+    resetPluginStateStoreForTests();
+    installMatrixTestRuntime();
+    const tempDir = tempDirs.make("matrix-idb-interval-");
+    const pendingDatabases = createDeferred<IDBDatabaseInfo[]>();
+    const databasesSpy = vi
+      .spyOn(indexedDB, "databases")
+      .mockResolvedValueOnce([])
+      .mockReturnValueOnce(pendingDatabases.promise);
    const setIntervalSpy = vi.spyOn(globalThis, "setInterval");
+    const warnSpy = vi.spyOn(LogService, "warn").mockImplementation(() => {});
+    let shutdown: Promise<void> | undefined;

-    const client = new MatrixClient("https://matrix.example.org", "token", {
-      encryption: true,
-      idbSnapshotPath: path.join(os.tmpdir(), "matrix-idb-interval.json"),
-      cryptoDatabasePrefix: "openclaw-matrix-interval",
-    });
+    try {
+      const client = new MatrixClient("https://matrix.example.org", "token", {
+        encryption: true,
+        idbSnapshotPath: path.join(tempDir, "crypto-idb-snapshot.json"),
+        cryptoDatabasePrefix: "openclaw-matrix-interval",
+      });

-    await client.start();
+      await client.start();

-    expect(databasesSpy).toHaveBeenCalled();
-    const intervalCall = setIntervalSpy.mock.calls.find((call) => call[1] === 60_000) as
-      | unknown[]
-      | undefined;
-    if (!intervalCall) {
-      throw new Error("expected Matrix IDB snapshot interval");
+      const intervalCall = setIntervalSpy.mock.calls.find((call) => call[1] === 60_000) as
+        | unknown[]
+        | undefined;
+      if (!intervalCall || typeof intervalCall[0] !== "function") {
+        throw new Error("expected Matrix IDB snapshot interval");
+      }
+      intervalCall[0]();
+      intervalCall[0]();
+      await vi.waitFor(() => {
+        expect(databasesSpy).toHaveBeenCalledTimes(2);
+      });
+
+      shutdown = Promise.resolve(client.stopWithoutPersist());
+      let shutdownSettled = false;
+      void shutdown.then(() => {
+        shutdownSettled = true;
+      });
+      await vi.waitFor(() => {
+        expect(matrixJsClient.stopClient).toHaveBeenCalledTimes(1);
+      });
+      expect(shutdownSettled).toBe(false);
+
+      pendingDatabases.resolve([]);
+      await shutdown;
+      expect(readMatrixIdbSnapshotJson(tempDir)).toBeNull();
+      expect(warnSpy).not.toHaveBeenCalled();
+    } finally {
+      pendingDatabases.resolve([]);
+      await shutdown?.catch(() => undefined);
+      warnSpy.mockRestore();
+      databasesSpy.mockRestore();
+      setIntervalSpy.mockRestore();
    }
-    expect(intervalCall[0]).toBeTypeOf("function");
-    expect(intervalCall[1]).toBe(60_000);
-    client.stop();
  });

  it("reports own verification status when crypto marks device as verified", async () => {
@@ -148,6 +148,9 @@ export abstract class MatrixClientBase {
  protected transactionScopePromise: Promise<string> | null = null;
  private readonly messageWireDispatchGuards = new Map<string, MatrixMessageWireDispatchGuard>();
  private sdkStopped = false;
+  private stopDiscardPromise: Promise<void> | null = null;
+  private idbPersistPromise: Promise<void> | null = null;
+  private idbPersistAbortController: AbortController | null = null;

  readonly dms = {
    update: async (): Promise<boolean> => {
@@ -494,6 +497,7 @@ export abstract class MatrixClientBase {
    throwIfMatrixStartupAborted(opts.abortSignal);
    this.registerBridge();
    await this.initializeCryptoIfNeeded(opts.abortSignal);
+    throwIfMatrixStartupAborted(opts.abortSignal);

    await this.client.startClient({
      initialSyncLimit: this.initialSyncLimit,
@@ -547,10 +551,6 @@ export abstract class MatrixClientBase {
    if (this.sdkStopped) {
      return;
    }
-    if (this.idbPersistTimer) {
-      clearInterval(this.idbPersistTimer);
-      this.idbPersistTimer = null;
-    }
    this.currentSyncState = null;
    this.currentSyncError = undefined;
    this.client.stopClient();
@@ -583,15 +583,23 @@ export abstract class MatrixClientBase {
      .catch(noop);
  }

-  async stopAndPersist(): Promise<void> {
-    if (this.stopPersistPromise) {
-      await this.stopPersistPromise;
-      return;
-    }
-    this.stopPersistPromise = (async () => {
+  private async stopClientGeneration(persist: boolean): Promise<void> {
+    if (persist) {
      await this.quiesceSync();
-      this.stopSdkClient();
-      this.decryptBridge?.stop();
+    } else {
+      await this.quiesceSync().catch(noop);
+      this.syncStore?.discardPendingSyncCursorPersistence();
+    }
+    if (this.idbPersistTimer) {
+      clearInterval(this.idbPersistTimer);
+      this.idbPersistTimer = null;
+    }
+    this.idbPersistAbortController?.abort();
+    const activePeriodicPersist = this.idbPersistPromise;
+    this.stopSdkClient();
+    this.decryptBridge?.stop();
+    await activePeriodicPersist;
+    if (persist) {
      const runtime = loadedMatrixCryptoRuntime ?? (await loadMatrixCryptoRuntime());
      await runtime.persistIdbToDisk({
        snapshotPath: this.idbSnapshotPath,
@@ -600,15 +608,22 @@ export abstract class MatrixClientBase {
      });
      this.syncStore?.markCleanShutdown();
      await this.syncStore?.flush();
-    })();
+    }
+  }
+
+  async stopAndPersist(): Promise<void> {
+    this.stopPersistPromise ??= this.stopClientGeneration(true);
    await this.stopPersistPromise;
  }

-  stopWithoutPersist(): void {
-    this.syncStore?.discardPendingSyncCursorPersistence();
-    this.stopSdkClient();
-    this.decryptBridge?.stop();
-    this.stopPersistPromise = Promise.resolve();
+  stopWithoutPersist(): Promise<void> {
+    // Memoization closes concurrent callers; durable failure still requires discard cleanup.
+    if (!this.stopPersistPromise) {
+      this.stopPersistPromise = this.stopDiscardPromise = this.stopClientGeneration(false);
+    }
+    return (this.stopDiscardPromise ??= this.stopPersistPromise.catch(() =>
+      this.stopClientGeneration(false),
+    ));
  }

  protected async bootstrapCryptoIfNeeded(abortSignal?: AbortSignal): Promise<void> {
@@ -686,18 +701,31 @@ export abstract class MatrixClientBase {
      await persistIdbToDisk({
        snapshotPath: this.idbSnapshotPath,
        databasePrefix: this.cryptoDatabasePrefix,
+        abortSignal,
      });
      throwIfMatrixStartupAborted(abortSignal);

      // Periodically persist to capture new Olm sessions and room keys.
      this.idbPersistTimer = setInterval(() => {
-        persistIdbToDisk({
+        if (this.idbPersistPromise) {
+          return;
+        }
+        const abortController = new AbortController();
+        this.idbPersistAbortController = abortController;
+        this.idbPersistPromise = persistIdbToDisk({
          snapshotPath: this.idbSnapshotPath,
          databasePrefix: this.cryptoDatabasePrefix,
-        }).catch(noop);
+          abortSignal: abortController.signal,
+        })
+          .catch(noop)
+          .finally(() => {
+            this.idbPersistPromise = null;
+            this.idbPersistAbortController = null;
+          });
      }, MATRIX_IDB_PERSIST_INTERVAL_MS);
      this.idbPersistTimer.unref?.();
    } catch (err) {
+      throwIfMatrixStartupAborted(abortSignal);
      LogService.warn("MatrixClientLite", "Failed to initialize rust crypto:", err);
    }
  }
@@ -3,6 +3,7 @@ import "fake-indexeddb/auto";
 import fs from "node:fs";
 import os from "node:os";
 import path from "node:path";
+import { createDeferred } from "openclaw/plugin-sdk/extension-shared";
 import { resetFileLockStateForTest } from "openclaw/plugin-sdk/file-lock";
 import { resetPluginStateStoreForTests } from "openclaw/plugin-sdk/plugin-state-test-runtime";
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
@@ -183,4 +184,38 @@ describe("Matrix IndexedDB persistence", () => {
      databasesSpy.mockRestore();
    }
  });
+
+  it("cancels an active snapshot before writing without warning", async () => {
+    const snapshotPath = path.join(tmpDir, "crypto-idb-snapshot.json");
+    await seedDatabase({
+      name: cryptoDatabaseName,
+      storeName: "sessions",
+      records: [{ key: "room-1", value: { session: "abc123" } }],
+    });
+    const databaseList = await indexedDB.databases();
+    const pendingDatabases = createDeferred<IDBDatabaseInfo[]>();
+    const databasesSpy = vi.spyOn(indexedDB, "databases").mockReturnValue(pendingDatabases.promise);
+    const abortController = new AbortController();
+
+    try {
+      const persistence = persistIdbToDisk({
+        snapshotPath,
+        databasePrefix: DATABASE_PREFIX,
+        abortSignal: abortController.signal,
+      });
+      await vi.waitFor(() => {
+        expect(databasesSpy).toHaveBeenCalledTimes(1);
+      });
+
+      abortController.abort();
+      pendingDatabases.resolve(databaseList);
+
+      await expect(persistence).resolves.toBeUndefined();
+      expect(readMatrixIdbSnapshotJson(tmpDir)).toBeNull();
+      expect(warnSpy).not.toHaveBeenCalled();
+    } finally {
+      pendingDatabases.resolve(databaseList);
+      databasesSpy.mockRestore();
+    }
+  });
 });
@@ -307,6 +307,7 @@ export async function persistIdbToDisk(params?: {
  snapshotPath?: string;
  databasePrefix?: string;
  strict?: boolean;
+  abortSignal?: AbortSignal;
 }): Promise<void> {
  const snapshotPath = params?.snapshotPath ?? resolveDefaultIdbSnapshotPath();
  let callbackStarted = false;
@@ -330,7 +331,7 @@ export async function persistIdbToDisk(params?: {
        }
        throwIfLegacySnapshotNeedsDoctor(snapshotPath, storedSnapshotJson);
        const snapshot = await dumpIndexedDatabases(params?.databasePrefix);
-        if (snapshot.length === 0) {
+        if (params?.abortSignal?.aborted || snapshot.length === 0) {
          return 0;
        }
        writeMatrixIdbSnapshotJson({
@@ -39,15 +39,15 @@ export function createMatrixQaE2eeClientLifecycle(params: {
  drainPendingDecryptions: () => Promise<void>;
  shutdownTimeoutMs: number;
  stopAndPersist: () => Promise<void>;
-  stopWithoutPersist: () => void;
+  stopWithoutPersist: () => Promise<void>;
 }) {
  const activeOperations = new Set<Promise<unknown>>();
  let shutdownStarted = false;
  let stopPromise: Promise<void> | undefined;

-  const failShutdown = (phase: string, cause: unknown): never => {
+  const failShutdown = async (phase: string, cause: unknown): Promise<never> => {
    try {
-      params.stopWithoutPersist();
+      await params.stopWithoutPersist();
    } catch {
      // Preserve the lifecycle failure that explains why persistence was skipped.
    }
@@ -70,17 +70,15 @@ export function createMatrixQaE2eeClientLifecycle(params: {
          Promise.allSettled(activeOperations),
          graceMs,
          "active Matrix SDK operations did not settle before shutdown",
-        ).catch((error: unknown) => {
-          failShutdown("waiting for active Matrix SDK operations", error);
-        });
+        ).catch((error: unknown) =>
+          failShutdown("waiting for active Matrix SDK operations", error),
+        );
      }
      await withMatrixQaE2eeTimeout(
        params.drainPendingDecryptions(),
        Math.max(0, deadline - Date.now()),
        "pending Matrix decryptions did not drain before shutdown",
-      ).catch((error: unknown) => {
-        failShutdown("draining pending Matrix decryptions", error);
-      });
+      ).catch((error: unknown) => failShutdown("draining pending Matrix decryptions", error));
      await params.stopAndPersist();
    })();
    return stopPromise;
@@ -21,6 +21,7 @@ const testing = {

 describe("matrix qa e2ee client storage", () => {
  function createLifecycleFixture(options?: {
+    discard?: () => Promise<void>;
    drain?: () => Promise<void>;
    shutdownTimeoutMs?: number;
  }) {
@@ -35,7 +36,10 @@ describe("matrix qa e2ee client storage", () => {
      stopAndPersist: vi.fn(async () => {
        calls.push("stop-and-persist");
      }),
-      stopWithoutPersist: vi.fn(() => calls.push("stop-and-discard")),
+      stopWithoutPersist: vi.fn(async () => {
+        calls.push("stop-and-discard");
+        await options?.discard?.();
+      }),
    });
    return { calls, lifecycle };
  }
@@ -100,7 +104,12 @@ describe("matrix qa e2ee client storage", () => {
  it("discards without persisting when active operation grace expires", async () => {
    vi.useFakeTimers();
    try {
+      let finishDiscard: (() => void) | undefined;
      const { calls, lifecycle } = createLifecycleFixture({
+        discard: () =>
+          new Promise<void>((resolve) => {
+            finishDiscard = resolve;
+          }),
        shutdownTimeoutMs: 100,
      });
      void lifecycle.runOperation({
@@ -118,8 +127,16 @@ describe("matrix qa e2ee client storage", () => {

      await vi.advanceTimersByTimeAsync(100);

-      await rejection;
+      let rejected = false;
+      void stop.catch(() => {
+        rejected = true;
+      });
+      await Promise.resolve();
+      expect(rejected).toBe(false);
      expect(calls).toEqual(["operation", "detach", "stop-and-discard"]);
+
+      finishDiscard?.();
+      await rejection;
    } finally {
      vi.useRealTimers();
    }