Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
120 changes: 120 additions & 0 deletions src/daemon/__tests__/durable-capture-recovery-authority.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -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>;
}) => 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');
Expand Down
57 changes: 55 additions & 2 deletions src/daemon/durable-capture-recovery-authority.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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<K extends string, H extends LiveResourceHandle<C>, C>(
params: DurableCaptureRecoveryAuthorityParams<K, H, C>,
authority: DurableCaptureRecoveryAuthority<K, H, C>,
primaryError: unknown,
): Promise<void> {
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<K extends string, H extends LiveResourceHandle<C>, C>(
Expand Down