From 49c5aa88892a36c614215738ad33b0b23b8df7bf Mon Sep 17 00:00:00 2001 From: King Star Date: Fri, 28 Aug 2026 00:51:58 +0800 Subject: [PATCH] fix(daemon): dispose late durable capture authorities --- ...durable-capture-recovery-authority.test.ts | 120 ++++++++++++++++++ .../durable-capture-recovery-authority.ts | 57 ++++++++- 2 files changed, 175 insertions(+), 2 deletions(-) diff --git a/src/daemon/__tests__/durable-capture-recovery-authority.test.ts b/src/daemon/__tests__/durable-capture-recovery-authority.test.ts index 8e6b017e40..2402a0765d 100644 --- a/src/daemon/__tests__/durable-capture-recovery-authority.test.ts +++ b/src/daemon/__tests__/durable-capture-recovery-authority.test.ts @@ -18,8 +18,34 @@ const envelope = createDurableResourceEnvelope({ descriptor: { version: 1, body: {} }, }); +test('acquisition winner leaves authority disposal to the caller', async () => { + const disposeControl = vi.fn(async () => {}); + const authority = await acquireDurableCaptureRecoveryAuthorityBeforeDeadline({ + displayName: 'app-log', + envelope, + scope: { + signal: new AbortController().signal, + diagnostics: { emit: () => {} }, + progress: { report: () => {} }, + }, + deadlineMs: 10_000, + acquireControl: async () => ({ + reattach: async () => ({ status: 'missing' as const }), + cleanup: async () => ({ status: 'already-missing' as const }), + [Symbol.asyncDispose]: disposeControl, + }), + onLateCleanupFailure: () => {}, + }); + + expect(authority.reattached).toEqual({ status: 'missing' }); + expect(disposeControl).not.toHaveBeenCalled(); + await authority.control[Symbol.asyncDispose](); + expect(disposeControl).toHaveBeenCalledOnce(); +}); + test('deadline abort disposes authority that becomes active after the caller has timed out', async () => { vi.useFakeTimers(); + const combinedSignal = vi.spyOn(AbortSignal, 'any').mockReturnValue(new AbortController().signal); try { let resolveReattach!: (outcome: { status: 'active'; handle: AppLogLiveHandle }) => void; const forceCleanup = vi.fn(async () => ({ status: 'cleaned' }) as const); @@ -69,10 +95,104 @@ test('deadline abort disposes authority that becomes active after the caller has expect(disposeControl).toHaveBeenCalledOnce(); expect(cleanupFailures).not.toHaveBeenCalled(); } finally { + combinedSignal.mockRestore(); vi.useRealTimers(); } }); +test('deadline abort disposes control when late reattach reports a non-active outcome', async () => { + vi.useFakeTimers(); + const combinedSignal = vi.spyOn(AbortSignal, 'any').mockReturnValue(new AbortController().signal); + try { + let resolveReattach!: (outcome: { status: 'missing' }) => void; + const disposeControl = vi.fn(async () => {}); + const acquisition = acquireDurableCaptureRecoveryAuthorityBeforeDeadline({ + displayName: 'app-log', + envelope, + scope: { + signal: new AbortController().signal, + diagnostics: { emit: () => {} }, + progress: { report: () => {} }, + }, + deadlineMs: 25, + acquireControl: async () => ({ + reattach: async () => + await new Promise<{ status: 'missing' }>((resolve) => { + resolveReattach = resolve; + }), + cleanup: async () => ({ status: 'already-missing' }), + [Symbol.asyncDispose]: disposeControl, + }), + onLateCleanupFailure: () => {}, + }); + + const timedOut = expect(acquisition).rejects.toBeInstanceOf( + DurableCaptureRecoveryDeadlineError, + ); + await vi.advanceTimersByTimeAsync(25); + await timedOut; + + resolveReattach({ status: 'missing' }); + await vi.advanceTimersByTimeAsync(0); + + expect(disposeControl).toHaveBeenCalledOnce(); + } finally { + combinedSignal.mockRestore(); + vi.useRealTimers(); + } +}); + +test('cancellation keeps its primary error when late control cleanup fails', async () => { + const controller = new AbortController(); + const cancellation = new Error('request canceled'); + const cleanupError = new Error('late cleanup failed'); + const combinedSignal = vi.spyOn(AbortSignal, 'any').mockReturnValue(new AbortController().signal); + let publishControl!: (control: { + reattach: () => Promise<{ status: 'missing' }>; + cleanup: () => Promise<{ status: 'already-missing' }>; + [Symbol.asyncDispose]: () => Promise; + }) => void; + const cleanupFailures = vi.fn(); + try { + const acquisition = acquireDurableCaptureRecoveryAuthorityBeforeDeadline({ + displayName: 'app-log', + envelope, + scope: { + signal: controller.signal, + diagnostics: { emit: () => {} }, + progress: { report: () => {} }, + }, + deadlineMs: 10_000, + acquireControl: async () => + await new Promise((resolve) => { + publishControl = resolve; + }), + onLateCleanupFailure: cleanupFailures, + }); + + controller.abort(cancellation); + await expect(acquisition).rejects.toBe(cancellation); + + publishControl({ + reattach: async () => ({ status: 'missing' }), + cleanup: async () => ({ status: 'already-missing' }), + [Symbol.asyncDispose]: async () => { + throw cleanupError; + }, + }); + + await vi.waitFor(() => + expect(cleanupFailures).toHaveBeenCalledWith( + 'late_control_cleanup_failed', + cleanupError, + cancellation, + ), + ); + } finally { + combinedSignal.mockRestore(); + } +}); + test('request cancellation wins before the deadline and disposes late exact-owner control', async () => { const controller = new AbortController(); const cancellation = new Error('request canceled'); diff --git a/src/daemon/durable-capture-recovery-authority.ts b/src/daemon/durable-capture-recovery-authority.ts index e77d3083ca..c4fc7c7b2f 100644 --- a/src/daemon/durable-capture-recovery-authority.ts +++ b/src/daemon/durable-capture-recovery-authority.ts @@ -83,13 +83,66 @@ export async function acquireDurableCaptureRecoveryAuthorityBeforeDeadline< }, params.deadlineMs); timer.unref?.(); const acquisition = acquireRecoveryAuthority(params, scope); + type RaceWinner = + | Readonly<{ kind: 'acquisition' }> + | Readonly<{ kind: 'acquisition-error' }> + | Readonly<{ kind: 'deadline'; error: DurableCaptureRecoveryDeadlineError }> + | Readonly<{ kind: 'cancellation'; error: unknown }>; + let raceWinner: RaceWinner | undefined; + const acquisitionForRace = acquisition.then( + (authority) => { + raceWinner ??= { kind: 'acquisition' }; + return authority; + }, + (error: unknown) => { + raceWinner ??= { kind: 'acquisition-error' }; + throw error; + }, + ); + const deadlineForRace = deadline.catch((error: DurableCaptureRecoveryDeadlineError) => { + raceWinner ??= { kind: 'deadline', error }; + throw error; + }); + const cancellationForRace = cancellation.catch((error: unknown) => { + raceWinner ??= { kind: 'cancellation', error }; + throw error; + }); try { - return await Promise.race([acquisition, deadline, cancellation]); + return await Promise.race([acquisitionForRace, deadlineForRace, cancellationForRace]); } finally { clearTimeout(timer); stopListeningForCancellation(); - void acquisition.catch(() => {}); + if (raceWinner?.kind === 'deadline' || raceWinner?.kind === 'cancellation') { + const primaryError = raceWinner.error; + void acquisition.then( + (authority) => disposeLateRecoveryAuthority(params, authority, primaryError), + () => {}, + ); + } else { + void acquisition.catch(() => {}); + } + } +} + +async function disposeLateRecoveryAuthority, C>( + params: DurableCaptureRecoveryAuthorityParams, + authority: DurableCaptureRecoveryAuthority, + primaryError: unknown, +): Promise { + if (authority.reattached.status === 'active') { + await disposeLateAuthority( + params, + authority.reattached.handle, + 'late_handle_cleanup_failed', + primaryError, + ); } + await disposeLateAuthority( + params, + authority.control, + 'late_control_cleanup_failed', + primaryError, + ); } async function acquireRecoveryAuthority, C>(