diff --git a/packages/provider-limrun/src/request-cancellation.test.ts b/packages/provider-limrun/src/request-cancellation.test.ts new file mode 100644 index 0000000000..d48927cca1 --- /dev/null +++ b/packages/provider-limrun/src/request-cancellation.test.ts @@ -0,0 +1,41 @@ +import { expect, test } from 'vitest'; +import { createLimrunRequestOperationDrain } from './request-cancellation.ts'; + +test('releases a settled operation from the drain', async () => { + const drain = createLimrunRequestOperationDrain(); + await expect(drain.wait(Promise.resolve('provider line'))).resolves.toBe('provider line'); + await expect(drain[Symbol.asyncDispose]()).resolves.toBeUndefined(); +}); + +test('propagates a provider rejection and still releases the drain', async () => { + const drain = createLimrunRequestOperationDrain(); + await expect(drain.wait(Promise.reject(new Error('socket closed')))).rejects.toThrow( + 'socket closed', + ); + await expect(drain[Symbol.asyncDispose]()).resolves.toBeUndefined(); +}); + +test('keeps an aborted operation owned until its provider source settles', async () => { + const drain = createLimrunRequestOperationDrain(); + let settleSource!: (value: string) => void; + const source = new Promise((resolve) => { + settleSource = resolve; + }); + const controller = new AbortController(); + const waiting = drain.wait(source, controller.signal, 'limrun operation aborted'); + const reason = new Error('release the instance'); + + controller.abort(reason); + await expect(waiting).rejects.toBe(reason); + + let released = false; + const disposing = drain[Symbol.asyncDispose]().then(() => { + released = true; + }); + await new Promise((resolve) => setTimeout(resolve, 5)); + expect(released).toBe(false); + + settleSource('late provider line'); + await expect(disposing).resolves.toBeUndefined(); + expect(released).toBe(true); +}); diff --git a/packages/provider-limrun/src/request-cancellation.ts b/packages/provider-limrun/src/request-cancellation.ts index d2d8a17133..d377dcf3c9 100644 --- a/packages/provider-limrun/src/request-cancellation.ts +++ b/packages/provider-limrun/src/request-cancellation.ts @@ -62,10 +62,12 @@ export function createLimrunRequestOperationDrain(): LimrunRequestOperationDrain wait: async (source: Promise, signal?: AbortSignal, abortMessage?: string) => { let settled!: () => void; const completion = new Promise((resolve) => { - settled = resolve; + settled = () => { + pending.delete(completion); + resolve(); + }; }); pending.add(completion); - void completion.then(() => pending.delete(completion)); return await awaitLimrunOperation(source, signal, abortMessage, settled); }, [Symbol.asyncDispose]: async () => { diff --git a/src/daemon/managed-device-allocation/lease-admission.ts b/src/daemon/managed-device-allocation/lease-admission.ts index a56bd0daf1..7fad72f5df 100644 --- a/src/daemon/managed-device-allocation/lease-admission.ts +++ b/src/daemon/managed-device-allocation/lease-admission.ts @@ -152,6 +152,7 @@ export function createManagedLeaseAdmission(params: { pending, horizon.deadline, AbortSignal.any([signal, fenced.signal]), + (error) => stop('authority-unconfirmed', normalizeError(error)), ); if (stopped) return stopped; if (abandoned) return abandoned; @@ -199,6 +200,7 @@ function waitForRenewal( work: Promise, deadline: Deadline, signal: AbortSignal, + onRejection: (error: unknown) => void, ): Promise { return new Promise((resolve) => { const timer = setTimeout(() => finish({ status: 'deadline-exceeded' }), deadline.remainingMs()); @@ -208,6 +210,14 @@ function waitForRenewal( listener[Symbol.dispose](); resolve(result); } - void work.then(() => finish()); + // A settled renewal only means "re-check the horizon"; a rejected one is unconfirmed + // authority, so it must fence instead of re-arming renewal work in a tight loop. + void work.then( + () => finish(), + (error) => { + onRejection(error); + finish(); + }, + ); }); } diff --git a/src/platform-runtime-screen-recording-apple-simulator-host.test.ts b/src/platform-runtime-screen-recording-apple-simulator-host.test.ts index f7149b154b..5e82b868a1 100644 --- a/src/platform-runtime-screen-recording-apple-simulator-host.test.ts +++ b/src/platform-runtime-screen-recording-apple-simulator-host.test.ts @@ -1,6 +1,6 @@ import fs from 'node:fs'; import path from 'node:path'; -import { beforeEach, expect, test, vi } from 'vitest'; +import { beforeEach, expect, onTestFinished, test, vi } from 'vitest'; import { normalizeError } from '@agent-device/kernel/errors'; import { mkdtempForTestSync } from './__tests__/test-utils/tmp-dir.ts'; import { createAppleScreenRecordingHost } from './platform-runtime-screen-recording-apple-host.ts'; @@ -196,6 +196,43 @@ test('late provider acquisition after abort is rolled back exactly once', async expect(late.kill).toHaveBeenCalledWith('SIGINT'); }); +test.each([ + { label: 'before the acquisition is observed', abortedInStart: true }, + { label: 'while the transport imports settle', abortedInStart: false }, +])( + 'discards a transport rejection that arrives when the start was aborted %s', + async ({ abortedInStart }) => { + const root = mkdtempForTestSync('agent-device-recording-rejected-start-'); + const controller = new AbortController(); + const reason = new Error('cancel simulator transport start'); + let rejectStart: ((error: unknown) => void) | undefined; + const start = vi.fn(() => { + if (abortedInStart) controller.abort(reason); + return new Promise['process']>((_resolve, reject) => { + rejectStart = reject; + }); + }); + + const rejection = observeUnhandledRejections(); + const starting = withAppleSimulatorScreenRecordingTransport( + { available: true, mode: 'transport-composed', start }, + async () => + await startAppleSimulatorRecording( + simulator, + path.join(root, 'capture.mp4'), + controller.signal, + ), + ); + const rejected = expect(starting).rejects.toBe(reason); + await vi.waitFor(() => expect(rejectStart).toBeTypeOf('function')); + if (!abortedInStart) controller.abort(reason); + rejectStart?.(controller.signal.reason); + + await rejected; + await expect(rejection.settle()).resolves.toEqual([]); + }, +); + test('resolved provider acquisition aborted before publication removes partial output and settles', async () => { const root = mkdtempForTestSync('agent-device-recording-acquired-abort-'); const outputPath = path.join(root, 'capture.mp4'); @@ -319,6 +356,7 @@ test.each([ }); test('pidless provider process is killed and settled before start fails', async () => { + const rejection = observeUnhandledRejections(); const running = background(undefined); await expect( withTransport( @@ -326,7 +364,8 @@ test('pidless provider process is killed and settled before start fails', async async () => await startAppleSimulatorRecording(simulator, '/tmp/pidless.mp4'), ), ).rejects.toThrow('complete process identity'); - expect(running.kill).toHaveBeenCalledWith('SIGINT'); + expect(running.kill.mock.calls).toEqual([['SIGINT']]); + await expect(rejection.settle()).resolves.toEqual([]); }); test('unpublished recorder cleanup gives SIGINT a grace window before forcing exit', async () => { @@ -352,6 +391,25 @@ test('unpublished recorder cleanup gives SIGINT a grace window before forcing ex } }); +function observeUnhandledRejections() { + const messages: string[] = []; + const listener = (reason: unknown) => { + messages.push(reason instanceof Error ? reason.message : String(reason)); + }; + process.on('unhandledRejection', listener); + onTestFinished(() => { + process.off('unhandledRejection', listener); + }); + return { + settle: async () => { + for (let turn = 0; turn < 2; turn += 1) { + await new Promise((resolve) => setImmediate(resolve)); + } + return messages; + }, + }; +} + function background(pid: number | undefined, command?: string) { let settle: ((result: { stdout: string; stderr: string; exitCode: number }) => void) | undefined; const wait = new Promise<{ stdout: string; stderr: string; exitCode: number }>((resolve) => { diff --git a/src/platform-runtime-screen-recording-apple-simulator-host.ts b/src/platform-runtime-screen-recording-apple-simulator-host.ts index b24c25c384..8620762c65 100644 --- a/src/platform-runtime-screen-recording-apple-simulator-host.ts +++ b/src/platform-runtime-screen-recording-apple-simulator-host.ts @@ -128,7 +128,7 @@ async function acquireSimulatorProcess( if ('child' in acquisition) { await rollbackAcquiredSimulatorProcess(acquisition); } else { - void started.then(rollbackAcquiredSimulatorProcess); + void started.then(rollbackAcquiredSimulatorProcess).catch(() => undefined); } throw signal.reason; } @@ -143,7 +143,7 @@ async function acquireSimulatorProcess( return await Promise.race([started, aborted]); } catch (error) { if (!signal.aborted) throw error; - void started.then(rollbackAcquiredSimulatorProcess); + void started.then(rollbackAcquiredSimulatorProcess).catch(() => undefined); throw signal.reason; } finally { removeAbort();