diff --git a/packages/cli/src/rest/__tests__/test-sessions.spec.ts b/packages/cli/src/rest/__tests__/test-sessions.spec.ts index fe65baf66..ef00e7daf 100644 --- a/packages/cli/src/rest/__tests__/test-sessions.spec.ts +++ b/packages/cli/src/rest/__tests__/test-sessions.spec.ts @@ -41,7 +41,7 @@ describe('TestSessions REST client', () => { shouldRecord: true, } as any) - expect(api.post).toHaveBeenCalledWith('/next/test-sessions/run', expect.objectContaining({ + expect(api.post).toHaveBeenCalledWith('/v1/test-sessions/run', expect.objectContaining({ repoInfo: githubRepoInfo, }), expect.any(Object)) }) @@ -126,6 +126,71 @@ describe('TestSessions REST client', () => { }) }) + it('runs a test session via the async v1 endpoint', async () => { + const api = { + post: vi.fn().mockResolvedValue({ data: { schedulingId: 'op-id', sequenceIds: {} } }), + } + const testSessions = new TestSessions(api as any) + + const payload = { name: 'project', checkRunJobs: [], project: { logicalId: 'p' }, runLocation: 'eu-central-1', shouldRecord: false } + await testSessions.run(payload as any) + + expect(api.post).toHaveBeenCalledWith('/v1/test-sessions/run', payload, expect.objectContaining({ + transformRequest: expect.any(Function), + })) + }) + + it('gets scheduling completion by ID', async () => { + const api = { + get: vi.fn().mockResolvedValue({ data: { schedulingId: 'op-id', status: 'SUCCEEDED', error: null } }), + } + const testSessions = new TestSessions(api as any) + const signal = new AbortController().signal + + await testSessions.getSchedulingCompletion('op-id', { signal }) + + expect(api.get).toHaveBeenCalledWith('/v1/test-sessions/scheduling/op-id/completion', { + params: { maxWaitSeconds: 30 }, + signal, + }) + }) + + it('polls scheduling completion timeouts until complete', async () => { + const completed = { schedulingId: 'op-id', status: 'FAILED', error: { code: 'SCHEDULING_ERROR', message: 'nope' } } + const api = { + get: vi.fn() + .mockRejectedValueOnce(new RequestTimeoutError({ + statusCode: 408, + error: 'Request Timeout', + message: 'Scheduling is still in progress, but maximum per-request wait time was exceeded. Please retry.', + })) + .mockResolvedValueOnce({ data: completed }), + } + const testSessions = new TestSessions(api as any) + + const result = await testSessions.pollSchedulingUntilComplete('op-id') + + expect(result).toEqual(completed) + expect(api.get).toHaveBeenCalledTimes(2) + }) + + it('stops polling scheduling when the signal is aborted', async () => { + const api = { + get: vi.fn().mockRejectedValue(new RequestTimeoutError({ + statusCode: 408, + error: 'Request Timeout', + message: 'Still in progress.', + })), + } + const testSessions = new TestSessions(api as any) + const controller = new AbortController() + controller.abort() + + await expect(testSessions.pollSchedulingUntilComplete('op-id', { signal: controller.signal })) + .rejects.toBeInstanceOf(RequestTimeoutError) + expect(api.get).toHaveBeenCalledTimes(1) + }) + it('lists public test sessions with filters', async () => { const api = { get: vi.fn().mockResolvedValue({ data: { length: 0, entries: [], nextId: null } }), diff --git a/packages/cli/src/rest/test-sessions.ts b/packages/cli/src/rest/test-sessions.ts index f5c5a8650..e3ce7a22b 100644 --- a/packages/cli/src/rest/test-sessions.ts +++ b/packages/cli/src/rest/test-sessions.ts @@ -49,6 +49,45 @@ export type TriggerTestSessionResponse = { sequenceIds: Record } +export type RunTestSessionResponse = { + /** The scheduling operation dispatching this run's check runs in the background. */ + schedulingId: string + /** Only present for recorded runs (shouldRecord: true). */ + testSessionId?: string + testResultIds?: Record + sequenceIds: Record +} + +export type TestSessionSchedulingStatus = 'PENDING' | 'RUNNING' | 'SUCCEEDED' | 'FAILED' + +export type TestSessionSchedulingOperation = { + schedulingId: string + testSessionId: string | null + status: TestSessionSchedulingStatus + checksTotal: number + error: { code: string, message: string } | null + createdAt: string + startedAt: string | null + endedAt: string | null +} + +/** + * The background dispatch of the test session's check runs failed (or the + * worker died — errorCode ABANDONED): no results will arrive for the + * undispatched checks, so the run should fail immediately instead of waiting + * out the per-check result timeouts. + */ +export class TestSessionSchedulingFailedError extends Error { + /** The operation's error code (e.g. SCHEDULING_ERROR, ABANDONED), when known. */ + code?: string + + constructor (message: string, options?: ErrorOptions & { code?: string }) { + super(message, options) + this.name = 'TestSessionSchedulingFailedError' + this.code = options?.code + } +} + export type TestSessionMetadata = { environment?: string repoUrl?: string @@ -157,11 +196,42 @@ class TestSessions { } async run (payload: RunTestSessionRequest) { - return await this.api.post('/next/test-sessions/run', payload, { + return await this.api.post('/v1/test-sessions/run', payload, { transformRequest: compressJSONPayload, }) } + /** + * Long-poll the dispatch status of an asynchronously started test-session + * run. The server blocks for up to `maxWaitSeconds` and returns the + * operation once it reaches a terminal state, or responds 408 + * ({@link RequestTimeoutError}) while dispatch is still in progress — the + * retry cadence lives in the caller. + */ + getSchedulingCompletion (schedulingId: string, options?: { maxWaitSeconds?: number, signal?: AbortSignal }) { + return this.api.get(`/v1/test-sessions/scheduling/${schedulingId}/completion`, { + params: { maxWaitSeconds: options?.maxWaitSeconds ?? COMPLETION_MAX_WAIT_SECONDS }, + signal: options?.signal, + }) + } + + async pollSchedulingUntilComplete ( + schedulingId: string, + options?: { signal?: AbortSignal }, + ): Promise { + while (true) { + try { + const { data } = await this.getSchedulingCompletion(schedulingId, options) + return data + } catch (err) { + if (err instanceof RequestTimeoutError && !options?.signal?.aborted) { + continue + } + throw err + } + } + } + /** * @throws {NoMatchingChecksError} If no checks matched the request. */ diff --git a/packages/cli/src/services/__tests__/abstract-check-runner.spec.ts b/packages/cli/src/services/__tests__/abstract-check-runner.spec.ts index f36e7bbde..3dc24d7ad 100644 --- a/packages/cli/src/services/__tests__/abstract-check-runner.spec.ts +++ b/packages/cli/src/services/__tests__/abstract-check-runner.spec.ts @@ -9,6 +9,7 @@ vi.mock('../../rest/api.js', () => ({ testSessions: { run: vi.fn().mockResolvedValue({ data: { testSessionId: 'ts-123', sequenceIds: {} } }), getResultShortLinks: vi.fn().mockResolvedValue({ data: {} }), + pollSchedulingUntilComplete: vi.fn(), }, assets: { getLogs: vi.fn().mockResolvedValue([]), @@ -32,6 +33,8 @@ vi.mock('../socket-client.js', () => ({ // --------------------------------------------------------------------------- import { SocketClient } from '../socket-client.js' +import { testSessions } from '../../rest/api.js' +import { TestSessionSchedulingFailedError } from '../../rest/test-sessions.js' /** Minimal concrete subclass — scheduleChecks immediately returns with zero checks so the runner exits cleanly. */ class StubCheckRunner extends AbstractCheckRunner { @@ -240,3 +243,135 @@ describe('AbstractCheckRunner — SocketClient lifecycle', () => { expect(mockClient.endAsync).toHaveBeenCalledTimes(1) }) }) + +// --------------------------------------------------------------------------- +// Scheduling watch +// --------------------------------------------------------------------------- + +/** Returns a schedulingId and one pending check, so the run only settles via a + * check result or a scheduling failure. */ +class SchedulingStubRunner extends AbstractCheckRunner { + checksToSchedule: Array<{ check: any, sequenceId: SequenceId }> + + constructor (checks: Array<{ check: any, sequenceId: SequenceId }>) { + super('acc-1', 60, false, false) + this.checksToSchedule = checks + } + + // eslint-disable-next-line @typescript-eslint/no-unused-vars + scheduleChecks (_checkRunSuiteId: string): Promise<{ + testSessionId?: string + schedulingId?: string + checks: Array<{ check: any, sequenceId: SequenceId }> + }> { + return Promise.resolve({ testSessionId: 'ts-stub', schedulingId: 'op-1', checks: this.checksToSchedule }) + } +} + +describe('AbstractCheckRunner — scheduling watch', () => { + beforeEach(() => { + vi.clearAllMocks() + vi.mocked(SocketClient.connect).mockResolvedValue({ + on: vi.fn(), + subscribeAsync: vi.fn().mockResolvedValue(undefined), + endAsync: vi.fn().mockResolvedValue(undefined), + } as any) + vi.spyOn(process, 'rawListeners').mockReturnValue([]) + vi.spyOn(process, 'removeAllListeners').mockReturnValue(process) + vi.spyOn(process, 'on').mockReturnValue(process) + vi.spyOn(process, 'off').mockReturnValue(process) + }) + + afterEach(() => { + vi.restoreAllMocks() + }) + + it('fails the run immediately when the scheduling operation reports FAILED', async () => { + vi.mocked(testSessions.pollSchedulingUntilComplete).mockResolvedValue({ + schedulingId: 'op-1', + testSessionId: 'ts-stub', + status: 'FAILED', + checksTotal: 1, + error: { code: 'SCHEDULING_ERROR', message: 'Unable to find private location' }, + createdAt: '2026-01-01T00:00:00.000Z', + startedAt: null, + endedAt: null, + }) + + const runner = new SchedulingStubRunner([{ check: {}, sequenceId: 'seq-1' }]) + const errors: Error[] = [] + runner.on(Events.ERROR, err => errors.push(err)) + + await runner.run() + + expect(errors).toHaveLength(1) + expect(errors[0]).toBeInstanceOf(TestSessionSchedulingFailedError) + expect(errors[0].message).toEqual('Unable to find private location') + expect(testSessions.pollSchedulingUntilComplete).toHaveBeenCalledWith('op-1', expect.objectContaining({ + signal: expect.any(AbortSignal), + })) + }) + + it('does not settle the run when the scheduling operation succeeds', async () => { + vi.mocked(testSessions.pollSchedulingUntilComplete).mockResolvedValue({ + schedulingId: 'op-1', + testSessionId: 'ts-stub', + status: 'SUCCEEDED', + checksTotal: 1, + error: null, + createdAt: '2026-01-01T00:00:00.000Z', + startedAt: null, + endedAt: null, + }) + + // One pending check: the run may only settle once that check finishes — + // the SUCCEEDED watch must neither error nor resolve the race early. + const runner = new SchedulingStubRunner([{ check: {}, sequenceId: 'seq-1' }]) + const errors: Error[] = [] + let finished = false + runner.on(Events.ERROR, err => errors.push(err)) + runner.on(Events.RUN_FINISHED, () => { + finished = true + }) + + const runPromise = runner.run() + // Let the (mock-resolved) scheduling watch settle before the check does. + await new Promise(resolve => setTimeout(resolve, 20)) + expect(finished).toBe(false) + runner.emit(Events.CHECK_FINISHED) + await runPromise + ;(runner as any).disableAllTimeouts() + + expect(errors).toHaveLength(0) + expect(finished).toBe(true) + }) + + it('fails a detached run when dispatch fails', async () => { + vi.mocked(testSessions.pollSchedulingUntilComplete).mockResolvedValue({ + schedulingId: 'op-1', + testSessionId: 'ts-stub', + status: 'FAILED', + checksTotal: 1, + error: { code: 'ABANDONED', message: 'the worker stopped reporting progress' }, + createdAt: '2026-01-01T00:00:00.000Z', + startedAt: null, + endedAt: null, + }) + + const runner = new SchedulingStubRunner([{ check: {}, sequenceId: 'seq-1' }]) + ;(runner as any).detach = true + const errors: Array = [] + let detached = false + runner.on(Events.ERROR, err => errors.push(err)) + runner.on(Events.DETACH, () => { + detached = true + }) + + await runner.run() + + expect(detached).toBe(false) + expect(errors).toHaveLength(1) + expect(errors[0]).toBeInstanceOf(TestSessionSchedulingFailedError) + expect(errors[0].code).toEqual('ABANDONED') + }) +}) diff --git a/packages/cli/src/services/abstract-check-runner.ts b/packages/cli/src/services/abstract-check-runner.ts index 30a777581..0fd091aff 100644 --- a/packages/cli/src/services/abstract-check-runner.ts +++ b/packages/cli/src/services/abstract-check-runner.ts @@ -5,7 +5,11 @@ import * as uuid from 'uuid' import { EventEmitter } from 'node:events' import type { MqttClient } from 'mqtt' import type { Region } from '../index.js' -import { TestResultsShortLinks } from '../rest/test-sessions.js' +import { + TestResultsShortLinks, + TestSessionSchedulingFailedError, + TestSessionSchedulingOperation, +} from '../rest/test-sessions.js' import { PlaywrightCheck } from '../constructs/index.js' // eslint-disable-next-line no-restricted-syntax @@ -44,6 +48,13 @@ export const DEFAULT_PLAYWRIGHT_CHECK_RUN_TIMEOUT_SECONDS = 1200 const DEFAULT_SCHEDULING_DELAY_EXCEEDED_MS = 20000 +function schedulingFailedError (operation: TestSessionSchedulingOperation): TestSessionSchedulingFailedError { + return new TestSessionSchedulingFailedError( + operation.error?.message ?? 'The test session could not be scheduled.', + { code: operation.error?.code }, + ) +} + export default abstract class AbstractCheckRunner extends EventEmitter { checks: Map testSessionId?: string @@ -75,6 +86,8 @@ export default abstract class AbstractCheckRunner extends EventEmitter { abstract scheduleChecks (checkRunSuiteId: string): Promise<{ testSessionId?: string + /** Present when the run was dispatched asynchronously (POST /v1/test-sessions/run). */ + schedulingId?: string checks: Array<{ check: any, sequenceId: SequenceId }> }> @@ -82,15 +95,20 @@ export default abstract class AbstractCheckRunner extends EventEmitter { let socketClient = null let sigintHandler: (() => void) | null = null let previousSigintListeners: Array<(...args: any[]) => void> = [] + const schedulingWatchAbort = new AbortController() try { const checkRunSuiteId = uuid.v4() if (this.detach) { - const { testSessionId, checks } = await this.scheduleChecks(checkRunSuiteId) + const { testSessionId, schedulingId, checks } = await this.scheduleChecks(checkRunSuiteId) this.testSessionId = testSessionId this.checks = new Map( checks.map(({ check, sequenceId }) => [sequenceId, { check }]), ) + // Dispatch happens in the background after the 202: confirm it + // succeeded (a matter of seconds) before detaching, so a dispatch + // failure still exits non-zero like the old synchronous endpoint did. + await this.assertSchedulingSucceeded(schedulingId) this.emit(Events.RUN_STARTED, checks, testSessionId) this.emit(Events.DETACH) return @@ -101,7 +119,7 @@ export default abstract class AbstractCheckRunner extends EventEmitter { // Configure the socket listener and allChecksFinished listener before starting checks to avoid race conditions await this.configureResultListener(checkRunSuiteId, socketClient) - const { testSessionId, checks } = await this.scheduleChecks(checkRunSuiteId) + const { testSessionId, schedulingId, checks } = await this.scheduleChecks(checkRunSuiteId) this.testSessionId = testSessionId this.checks = new Map( checks.map(({ check, sequenceId }) => [sequenceId, { check }]), @@ -146,17 +164,31 @@ export default abstract class AbstractCheckRunner extends EventEmitter { // `allChecksFinished` should be started before processing check results in `queue.start()`. // Otherwise, there could be a race condition causing check results to be missed by `allChecksFinished()`. const allChecksFinished = this.allChecksFinished() + // The server dispatches the check runs in the background after the 202; + // watch the scheduling operation so a dispatch failure fails the run + // immediately with the real error instead of waiting out the per-check + // result timeouts. The promise only ever rejects (successful dispatch + // means results keep arriving over MQTT), so the race stays decided by + // allChecksFinished in the happy path. + const schedulingFailed = this.watchScheduling(schedulingId, schedulingWatchAbort.signal) + // Defensive: keep the promise permanently handled so its rejection can + // never become a global unhandled rejection if a future refactor inserts + // an await (or a throwing emit) before the race below observes it. + schedulingFailed.catch(() => {}) /// / Need to structure the checks depending on how it went this.emit(Events.RUN_STARTED, checks, testSessionId) // Start the queue after the test session run rest call is completed to avoid race conditions this.queue.start() - await allChecksFinished + await Promise.race([allChecksFinished, schedulingFailed]) this.emit(Events.RUN_FINISHED, testSessionId) } catch (err) { this.disableAllTimeouts() this.emit(Events.ERROR, err) } finally { + // Stop any in-flight scheduling long-poll so it doesn't hold the + // process open after the run has settled. + schedulingWatchAbort.abort() if (sigintHandler) { process.off('SIGINT', sigintHandler) for (const listener of previousSigintListeners) { @@ -169,6 +201,51 @@ export default abstract class AbstractCheckRunner extends EventEmitter { } } + /** + * Watches the run's scheduling operation and rejects with + * {@link TestSessionSchedulingFailedError} if the background dispatch fails. + * Never resolves: on successful dispatch the results keep streaming over + * MQTT and there is nothing to report. Watching is best-effort — polling + * errors (abort, network issues) never fail the run. + */ + private watchScheduling (schedulingId: string | undefined, signal: AbortSignal): Promise { + return new Promise((_, reject) => { + if (!schedulingId) { + return + } + testSessions.pollSchedulingUntilComplete(schedulingId, { signal }) + .then(operation => { + if (operation.status === 'FAILED') { + reject(schedulingFailedError(operation)) + } + }) + .catch(() => { + // Best-effort: only a definitive FAILED operation may fail the run. + }) + }) + } + + /** + * Detach-mode counterpart of {@link watchScheduling}: waits for the + * scheduling operation to finish and throws if dispatch failed. Polling + * errors are swallowed (best-effort), matching the watch. + */ + private async assertSchedulingSucceeded (schedulingId: string | undefined): Promise { + if (!schedulingId) { + return + } + let operation + try { + operation = await testSessions.pollSchedulingUntilComplete(schedulingId) + } catch { + // Best-effort: only a definitive FAILED operation may fail the run. + return + } + if (operation.status === 'FAILED') { + throw schedulingFailedError(operation) + } + } + private async configureResultListener (checkRunSuiteId: string, socketClient: MqttClient): Promise { socketClient.on('message', (topic: string, rawMessage: string | Buffer) => { const message = JSON.parse(rawMessage.toString('utf8')) diff --git a/packages/cli/src/services/test-runner.ts b/packages/cli/src/services/test-runner.ts index 16c3c03de..7e927458d 100644 --- a/packages/cli/src/services/test-runner.ts +++ b/packages/cli/src/services/test-runner.ts @@ -60,6 +60,7 @@ export default class TestRunner extends AbstractCheckRunner { checkRunSuiteId: string, ): Promise<{ testSessionId?: string + schedulingId?: string checks: Array<{ check: any, sequenceId: SequenceId }> }> { const checkRunJobs = this.checkBundles.map(({ construct: check, bundle }) => { @@ -98,11 +99,11 @@ export default class TestRunner extends AbstractCheckRunner { streamLogs: this.streamLogs, refreshCache: this.refreshCache, }) - const { testSessionId, sequenceIds } = data + const { testSessionId, schedulingId, sequenceIds } = data const checks = this.checkBundles.map(({ construct: check }) => { return { check, sequenceId: sequenceIds?.[check.logicalId] } }) - return { testSessionId, checks } + return { testSessionId, schedulingId, checks } } async processCheckResult (result: any) {