From d7c57716841cd65fa475e9c2fd776e88add09a17 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Wed, 26 Aug 2026 07:19:16 +0800 Subject: [PATCH] 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 --- .../matrix/client-resolver.test-helpers.ts | 2 +- .../matrix/src/matrix/client/shared.test.ts | 270 ++++++++++++++++-- extensions/matrix/src/matrix/client/shared.ts | 181 ++++++------ extensions/matrix/src/matrix/sdk.test.ts | 152 ++++++++-- .../matrix/src/matrix/sdk/client-base.ts | 68 +++-- .../src/matrix/sdk/idb-persistence.test.ts | 35 +++ .../matrix/src/matrix/sdk/idb-persistence.ts | 3 +- .../matrix/substrate/e2ee-client-internals.ts | 16 +- .../matrix/substrate/e2ee-client.test.ts | 21 +- 9 files changed, 582 insertions(+), 166 deletions(-) diff --git a/extensions/matrix/src/matrix/client-resolver.test-helpers.ts b/extensions/matrix/src/matrix/client-resolver.test-helpers.ts index 69ebbc72f190..7209bb6d518c 100644 --- a/extensions/matrix/src/matrix/client-resolver.test-helpers.ts +++ b/extensions/matrix/src/matrix/client-resolver.test-helpers.ts @@ -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; } diff --git a/extensions/matrix/src/matrix/client/shared.test.ts b/extensions/matrix/src/matrix/client/shared.test.ts index ee0b251b17eb..898ab4969f46 100644 --- a/extensions/matrix/src/matrix/client/shared.test.ts +++ b/extensions/matrix/src/matrix/client/shared.test.ts @@ -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): Promise): Promise { + 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(); 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(); + 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, failure: Error) => { + client.stopWithoutPersist.mockRejectedValue(failure); + }, + }, + { + name: "strict persistence fallback", + mode: "persist" as const, + configure: (client: ReturnType, 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, 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(); + const discard = createDeferred(); 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(); 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(); + const discard = createDeferred(); + 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> | 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(); + const monitorTasks = createDeferred(); + 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(); diff --git a/extensions/matrix/src/matrix/client/shared.ts b/extensions/matrix/src/matrix/client/shared.ts index bf72cdd749e7..32671e92614f 100644 --- a/extensions/matrix/src/matrix/client/shared.ts +++ b/extensions/matrix/src/matrix/client/shared.ts @@ -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; }; -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 { - if (state.leases.size === 0) { +async function waitForRetirementDrain( + state: SharedMatrixClientState, + task: Promise, + isPending: () => boolean, + timeoutMessage: string, +): Promise { + if (!isPending()) { return; } let deadline: NodeJS.Timeout | undefined; try { await Promise.race([ - state.noLeases.promise, + task, new Promise((_, 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(); + 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 | 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 { + 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); } } diff --git a/extensions/matrix/src/matrix/sdk.test.ts b/extensions/matrix/src/matrix/sdk.test.ts index 1f305a62259b..cba04859f79e 100644 --- a/extensions/matrix/src/matrix/sdk.test.ts +++ b/extensions/matrix/src/matrix/sdk.test.ts @@ -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(); + 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(); + 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 | 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 () => { diff --git a/extensions/matrix/src/matrix/sdk/client-base.ts b/extensions/matrix/src/matrix/sdk/client-base.ts index 72f082b943e0..2ed5746f9950 100644 --- a/extensions/matrix/src/matrix/sdk/client-base.ts +++ b/extensions/matrix/src/matrix/sdk/client-base.ts @@ -148,6 +148,9 @@ export abstract class MatrixClientBase { protected transactionScopePromise: Promise | null = null; private readonly messageWireDispatchGuards = new Map(); private sdkStopped = false; + private stopDiscardPromise: Promise | null = null; + private idbPersistPromise: Promise | null = null; + private idbPersistAbortController: AbortController | null = null; readonly dms = { update: async (): Promise => { @@ -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 { - if (this.stopPersistPromise) { - await this.stopPersistPromise; - return; - } - this.stopPersistPromise = (async () => { + private async stopClientGeneration(persist: boolean): Promise { + 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 { + this.stopPersistPromise ??= this.stopClientGeneration(true); await this.stopPersistPromise; } - stopWithoutPersist(): void { - this.syncStore?.discardPendingSyncCursorPersistence(); - this.stopSdkClient(); - this.decryptBridge?.stop(); - this.stopPersistPromise = Promise.resolve(); + stopWithoutPersist(): Promise { + // 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 { @@ -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); } } diff --git a/extensions/matrix/src/matrix/sdk/idb-persistence.test.ts b/extensions/matrix/src/matrix/sdk/idb-persistence.test.ts index 9a80a180c2cb..4158dfd9b8de 100644 --- a/extensions/matrix/src/matrix/sdk/idb-persistence.test.ts +++ b/extensions/matrix/src/matrix/sdk/idb-persistence.test.ts @@ -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(); + 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(); + } + }); }); diff --git a/extensions/matrix/src/matrix/sdk/idb-persistence.ts b/extensions/matrix/src/matrix/sdk/idb-persistence.ts index d6b6e3fa1564..cb0a145862e7 100644 --- a/extensions/matrix/src/matrix/sdk/idb-persistence.ts +++ b/extensions/matrix/src/matrix/sdk/idb-persistence.ts @@ -307,6 +307,7 @@ export async function persistIdbToDisk(params?: { snapshotPath?: string; databasePrefix?: string; strict?: boolean; + abortSignal?: AbortSignal; }): Promise { 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({ diff --git a/extensions/qa-lab/src/live-transports/matrix/substrate/e2ee-client-internals.ts b/extensions/qa-lab/src/live-transports/matrix/substrate/e2ee-client-internals.ts index 901d15e02640..667fc3b91f8b 100644 --- a/extensions/qa-lab/src/live-transports/matrix/substrate/e2ee-client-internals.ts +++ b/extensions/qa-lab/src/live-transports/matrix/substrate/e2ee-client-internals.ts @@ -39,15 +39,15 @@ export function createMatrixQaE2eeClientLifecycle(params: { drainPendingDecryptions: () => Promise; shutdownTimeoutMs: number; stopAndPersist: () => Promise; - stopWithoutPersist: () => void; + stopWithoutPersist: () => Promise; }) { const activeOperations = new Set>(); let shutdownStarted = false; let stopPromise: Promise | undefined; - const failShutdown = (phase: string, cause: unknown): never => { + const failShutdown = async (phase: string, cause: unknown): Promise => { 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; diff --git a/extensions/qa-lab/src/live-transports/matrix/substrate/e2ee-client.test.ts b/extensions/qa-lab/src/live-transports/matrix/substrate/e2ee-client.test.ts index 48e8e07e3c30..9d750056f769 100644 --- a/extensions/qa-lab/src/live-transports/matrix/substrate/e2ee-client.test.ts +++ b/extensions/qa-lab/src/live-transports/matrix/substrate/e2ee-client.test.ts @@ -21,6 +21,7 @@ const testing = { describe("matrix qa e2ee client storage", () => { function createLifecycleFixture(options?: { + discard?: () => Promise; drain?: () => Promise; 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((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(); }