From 19289684744be0e809b29c33b1d3c57aee05f747 Mon Sep 17 00:00:00 2001 From: finalerock44 Date: Fri, 16 Oct 2026 12:43:10 +0100 Subject: [PATCH 1/2] fix(cloud): re-send a flow submit whose response was lost When the API stalls, a `dcd cloud` submit can wait 300 s and then get a 502 from Railway's edge. The CLI can't tell whether the run was created, so it failed the command even when the run went ahead (dcd#1617). The API now answers a repeat of the same submit (same body, same upload id) with the run it already created, or a 409 while that run is still being created, and never creates a second run or charge. So the CLI sends a submit again while its outcome is unknown: - submitWithRetry (services/flow-submit.service.ts) re-sends the same body on a 409/500/502/503/504 or on any transport failure (network error, timeout, a body cut off), waiting 2 s and then up to 30 s between attempts, about three minutes in all. The flow zip is not uploaded again. - Each attempt has a 90 s timeout, well under the 300 s at which Railway's edge cuts it with a 502. - The multipart fallback for an API without the client-direct endpoint runs only when the first attempt 404s or 405s. A 404 after a retry comes from an API that took an earlier attempt. - `dcd cloud` prints one warning per retry (not in --json); dcd-mcp logs to stderr. Both record `submit_attempts` on the lifecycle telemetry event. - enhanceFetchError keeps the undici cause, so a timeout, a reset and a DNS failure no longer all read as "fetch failed". No new flags: `--retry` already means re-running failed tests. Co-Authored-By: Claude Opus 5.5 --- src/commands/cloud.ts | 61 +++-- src/gateways/api-gateway.ts | 31 ++- src/mcp/tools/run-cloud-test.ts | 58 +++-- src/services/flow-submit.service.ts | 122 ++++++++++ src/services/telemetry.service.ts | 19 ++ test/unit/flow-submit.test.ts | 250 ++++++++++++++++++++ test/unit/telemetry-submit-attempts.test.ts | 57 +++++ 7 files changed, 552 insertions(+), 46 deletions(-) create mode 100644 src/services/flow-submit.service.ts create mode 100644 test/unit/flow-submit.test.ts create mode 100644 test/unit/telemetry-submit-attempts.test.ts diff --git a/src/commands/cloud.ts b/src/commands/cloud.ts index 816f60c..d9d7087 100644 --- a/src/commands/cloud.ts +++ b/src/commands/cloud.ts @@ -13,6 +13,10 @@ import { } from '../methods.js'; import { DeviceValidationService } from '../services/device-validation.service.js'; import { plan } from '../services/execution-plan.service.js'; +import { + isSubmitEndpointMissing, + submitWithRetry, +} from '../services/flow-submit.service.js'; import { buildTestMetadataMap, computeCommonRoot, @@ -929,9 +933,27 @@ export const cloudCommand = defineCommand({ // 404s (route undefined), some proxies 405 (path/method not allowed); in // either case fall back to the legacy multipart POST /uploads/flow, which // is byte-identical bar the storage reference. - let response: Awaited>; - try { - const storageRef = await uploadFlowZip({ apiUrl, auth, buffer, debug }); + const submitMultipart = (status: number) => { + if (debug) { + out( + `[DEBUG] Client-direct flow upload unavailable (HTTP ${status}); falling back to multipart ${apiUrl}/uploads/flow`, + ); + } + + const testFormData = testSubmissionService.buildFormData(fields, buffer); + return ApiGateway.uploadFlow(apiUrl, auth, testFormData); + }; + + const submitFlows = async (): Promise< + Awaited> + > => { + let storageRef: Awaited>; + try { + storageRef = await uploadFlowZip({ apiUrl, auth, buffer, debug }); + } catch (error) { + if (!isSubmitEndpointMissing(error)) throw error; + return submitMultipart((error as ApiError).status); + } if (debug) { out( @@ -940,27 +962,22 @@ export const cloudCommand = defineCommand({ out(`[DEBUG] Submitting flow test to ${apiUrl}/uploads/submitFlowTest`); } - response = await ApiGateway.submitFlowTest(apiUrl, auth, { - ...fields, - ...storageRef, + // The same body on every attempt: its upload id is what lets the API + // answer a repeat with the run it already created (dcd#1617 follow-up). + const submitBody = { ...fields, ...storageRef }; + const submitted = await submitWithRetry({ + fallback: (error) => submitMultipart(error.status), + onRetry: ({ attempt, attempts, delayMs, reason }) => + warnOut( + `Could not confirm the test submission (${reason}); sending it again in ${delayMs / 1000}s (attempt ${attempt}/${attempts}). It will not run twice.`, + ), + submit: () => ApiGateway.submitFlowTest(apiUrl, auth, submitBody), }); - } catch (error) { - if ( - error instanceof ApiError && - (error.status === 404 || error.status === 405) - ) { - if (debug) { - out( - `[DEBUG] Client-direct flow upload unavailable (HTTP ${error.status}); falling back to multipart ${apiUrl}/uploads/flow`, - ); - } + telemetry.recordSubmitAttempts(submitted.attempts); + return submitted.response; + }; - const testFormData = testSubmissionService.buildFormData(fields, buffer); - response = await ApiGateway.uploadFlow(apiUrl, auth, testFormData); - } else { - throw error; - } - } + const response = await submitFlows(); const { message, results } = response; diff --git a/src/gateways/api-gateway.ts b/src/gateways/api-gateway.ts index e7c550e..d08eb1d 100644 --- a/src/gateways/api-gateway.ts +++ b/src/gateways/api-gateway.ts @@ -82,9 +82,14 @@ export const ApiGateway = { message += ` • Verify API URL is correct: ${origin}\n`; message += ` • Check for proxy/VPN interference\n`; message += ` • Try again in a few moments if server is temporarily down\n\n`; - message += `Original error: ${error.message}`; - - const enhancedError = new Error(message); + // undici puts the real reason on `cause` (UND_ERR_HEADERS_TIMEOUT, + // ECONNRESET, ENOTFOUND, …); without it a timeout, a reset and a DNS + // failure all read as the same "fetch failed". + const cause = (error as { cause?: unknown }).cause; + const code = (cause as { code?: unknown } | undefined)?.code; + message += `Original error: ${error.message}${typeof code === 'string' ? ` (${code})` : ''}`; + + const enhancedError = new Error(message, { cause: cause ?? error }); enhancedError.name = 'NetworkError'; enhancedError.stack = error.stack; return enhancedError; @@ -761,6 +766,7 @@ export const ApiGateway = { ...auth.headers, }, method: 'POST', + signal: AbortSignal.timeout(this.submitAttemptTimeoutMs), }); if (!res.ok) { await this.handleApiError(res, 'Failed to submit test flows'); @@ -776,10 +782,29 @@ export const ApiGateway = { throw this.enhanceFetchError(error, `${baseUrl}/uploads/submitFlowTest`); } + if (error instanceof Error && error.name === 'TimeoutError') { + const timeout = new Error( + `Failed to submit test flows: no response from the API within ${this.submitAttemptTimeoutMs / 1000}s`, + { cause: error }, + ); + timeout.name = 'SubmitTimeoutError'; + throw timeout; + } + throw error; } }, + /** + * How long one submit attempt may take (dcd#1617 follow-up). A submit + * normally answers in a second or two; one held by a Supabase stall is + * abandoned and re-sent by `submitWithRetry`, which is safe because the API + * answers a repeat with the run it already created. Well under the 300s at + * which Railway's edge would cut it with a 502. Settable so tests can + * shorten it. + */ + submitAttemptTimeoutMs: 90_000, + /** * Dry-run cost + cell-count estimate for a (possibly device-matrix) * submission. Runs the same resolve → validate → fan-out → price core as the diff --git a/src/mcp/tools/run-cloud-test.ts b/src/mcp/tools/run-cloud-test.ts index 0eb3bc1..1a29495 100644 --- a/src/mcp/tools/run-cloud-test.ts +++ b/src/mcp/tools/run-cloud-test.ts @@ -3,8 +3,12 @@ import * as path from 'node:path'; import type { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js'; import { z } from 'zod'; -import { ApiError, ApiGateway } from '../../gateways/api-gateway.js'; +import { ApiGateway } from '../../gateways/api-gateway.js'; import { plan } from '../../services/execution-plan.service.js'; +import { + isSubmitEndpointMissing, + submitWithRetry, +} from '../../services/flow-submit.service.js'; import { computeCommonRoot, buildTestMetadataMap } from '../../services/flow-paths.js'; import { DeviceValidationService } from '../../services/device-validation.service.js'; import { platformFromAppFile } from '../../services/notices.service.js'; @@ -282,27 +286,39 @@ export function registerRunCloudTest(server: McpServer): void { // referencing it. Older API deployments lack these endpoints (404/405); // fall back to the legacy multipart POST /uploads/flow. Mirrors // `dcd cloud`. - let response: Awaited>; - try { - const storageRef = await uploadFlowZip({ apiUrl, auth, buffer }); - response = await ApiGateway.submitFlowTest(apiUrl, auth, { - ...fields, - ...storageRef, - }); - } catch (error) { - if ( - error instanceof ApiError && - (error.status === 404 || error.status === 405) - ) { - const testFormData = testSubmissionService.buildFormData( - fields, - buffer, - ); - response = await ApiGateway.uploadFlow(apiUrl, auth, testFormData); - } else { - throw error; + const submitMultipart = () => + ApiGateway.uploadFlow( + apiUrl, + auth, + testSubmissionService.buildFormData(fields, buffer), + ); + const submitFlows = async (): Promise< + Awaited> + > => { + let storageRef: Awaited>; + try { + storageRef = await uploadFlowZip({ apiUrl, auth, buffer }); + } catch (error) { + if (!isSubmitEndpointMissing(error)) throw error; + return submitMultipart(); } - } + + // Re-sent while its outcome is unknown, always with the same body; + // see submitWithRetry (dcd#1617 follow-up). + const submitBody = { ...fields, ...storageRef }; + const submitted = await submitWithRetry({ + fallback: submitMultipart, + onRetry: ({ attempt, attempts, delayMs, reason }) => + logStderr( + `Could not confirm the test submission (${reason}); sending it again in ${delayMs / 1000}s (attempt ${attempt}/${attempts}). It will not run twice.`, + ), + submit: () => ApiGateway.submitFlowTest(apiUrl, auth, submitBody), + }); + telemetryExtra.submitAttempts = submitted.attempts; + return submitted.response; + }; + + const response = await submitFlows(); const { message, results } = response; if (!results?.length) { diff --git a/src/services/flow-submit.service.ts b/src/services/flow-submit.service.ts new file mode 100644 index 0000000..cabeeeb --- /dev/null +++ b/src/services/flow-submit.service.ts @@ -0,0 +1,122 @@ +import { ApiError } from '../gateways/api-gateway.js'; + +/** + * Submitting a flow test when the response may be lost (dcd#1617 follow-up). + * + * Supabase's ingress sometimes holds a request for minutes before answering, + * and Railway's edge cuts a request that is silent for 300s with a 502. Either + * way the CLI cannot tell whether its run was created. The API makes that safe + * to find out: it answers a repeat of the same submission (same body, same + * upload id from getFlowUploadUrl) with the run it already created, or a 409 + * while that run is still being created. It never creates a second run or a + * second charge. So a submit whose outcome is unknown is simply sent again. + */ + +/** + * Waits before each re-send: 2s, then up to 30s apart, about three minutes in + * all. That is long enough for a write held in a stall to land, after which the + * repeat is answered with the run. + */ +export const SUBMIT_RETRY_DELAYS_MS: readonly number[] = [ + 2000, 5000, 10_000, 20_000, 30_000, 30_000, 30_000, 30_000, +]; + +/** + * Statuses that leave the outcome unknown (502/503/504 from the edge or the + * API), say the API could not finish (500), or say the first attempt is still + * creating the run (409 SUBMISSION_IN_PROGRESS). Re-sending is only safe + * because the API dedupes on the upload id. + */ +const RETRYABLE_STATUSES = new Set([409, 500, 502, 503, 504]); + +/** + * Whether a failed attempt is worth sending again. Anything that is not an API + * answer is a transport failure: a network error, the attempt timeout, or a + * body that died mid-read after a 2xx. In every case the run may already exist, + * which a repeat finds out. The npm build throws undici's errors and the bun + * binaries their own; neither is an ApiError, so both retry. + */ +export function isRetryableSubmitError(error: unknown): boolean { + if (error instanceof ApiError) return RETRYABLE_STATUSES.has(error.status); + return true; +} + +/** A one-line reason for the retry notice. Network errors carry a long help text. */ +export function describeSubmitError(error: unknown): string { + if (error instanceof ApiError) return `HTTP ${error.status}`; + if (error instanceof Error) { + if (error.name === 'SubmitTimeoutError') return 'no response in time'; + const code = (error.cause as { code?: unknown } | undefined)?.code; + if (typeof code === 'string') return code; + return error.name === 'NetworkError' ? 'network error' : error.name; + } + + return String(error); +} + +/** The API predates the client-direct submit: a real API 404s, some proxies 405. */ +export function isSubmitEndpointMissing(error: unknown): boolean { + return ( + error instanceof ApiError && (error.status === 404 || error.status === 405) + ); +} + +export interface SubmitRetryNotice { + /** The attempt about to be sent, counting from 1. */ + attempt: number; + attempts: number; + delayMs: number; + reason: string; +} + +/** + * Send a submission, sending it again while its outcome is unknown. + * + * `fallback` is the legacy multipart submit for an API that predates this + * endpoint. It is used only when the *first* attempt says the endpoint is + * missing. A 404 on a later attempt comes from an API that took the earlier + * attempt, so a multipart submit then would create a second run. + * + * Resolves with the response and how many attempts it took. Throws the last + * error when it is not retryable or the attempts run out. + */ +export async function submitWithRetry(opts: { + delaysMs?: readonly number[]; + fallback: (error: ApiError) => Promise; + onRetry?: (notice: SubmitRetryNotice) => void; + sleep?: (ms: number) => Promise; + submit: () => Promise; +}): Promise<{ attempts: number; response: T }> { + const delaysMs = opts.delaysMs ?? SUBMIT_RETRY_DELAYS_MS; + const sleep = + opts.sleep ?? + ((ms: number) => + new Promise((resolve) => { + setTimeout(resolve, ms); + })); + const attempts = delaysMs.length + 1; + + for (let attempt = 1; ; attempt++) { + try { + return { attempts: attempt, response: await opts.submit() }; + } catch (error) { + if (attempt === 1 && isSubmitEndpointMissing(error)) { + return { + attempts: attempt, + response: await opts.fallback(error as ApiError), + }; + } + + const delayMs = delaysMs[attempt - 1]; + if (delayMs === undefined || !isRetryableSubmitError(error)) throw error; + + opts.onRetry?.({ + attempt: attempt + 1, + attempts, + delayMs, + reason: describeSubmitError(error), + }); + await sleep(delayMs); + } + } +} diff --git a/src/services/telemetry.service.ts b/src/services/telemetry.service.ts index 41d0778..6e3df8e 100644 --- a/src/services/telemetry.service.ts +++ b/src/services/telemetry.service.ts @@ -53,6 +53,7 @@ class Telemetry { private config: TelemetryConfig | null = null; private command: string = inferCommandFromArgv(); private encryption: EncryptionTelemetry | null = null; + private submitAttempts: number | undefined; private sessionId: string = randomUUID(); private startedAt: number = Date.now(); private readonly disabled: boolean = !!process.env.DCD_TELEMETRY_DISABLED; @@ -139,11 +140,22 @@ class Telemetry { }; } + /** + * Record how many attempts the flow submit took (dcd#1617 follow-up), sent + * as `submit_attempts` on `command completed` / `command failed`. Anything + * above 1 means a response was lost and the submit was sent again, which is + * what to count when judging how often the API stalls under the CLI. + */ + recordSubmitAttempts(attempts: number) { + this.submitAttempts = attempts; + } + recordCommandSuccess() { this.enqueue('info', 'cli.lifecycle', 'command completed', { duration_ms: Date.now() - this.startedAt, exit_code: 0, ...this.encryptionExtra(), + ...this.submitExtra(), }); } @@ -162,6 +174,7 @@ class Telemetry { error_stack: opts.error instanceof Error ? opts.error.stack : undefined, ...this.encryptionExtra(), + ...this.submitExtra(), }); } @@ -169,6 +182,12 @@ class Telemetry { return this.encryption ? { encrypt: this.encryption } : {}; } + private submitExtra(): { submit_attempts?: number } { + return this.submitAttempts === undefined + ? {} + : { submit_attempts: this.submitAttempts }; + } + private enqueue( level: TelemetryLevel, context: string, diff --git a/test/unit/flow-submit.test.ts b/test/unit/flow-submit.test.ts new file mode 100644 index 0000000..0d12add --- /dev/null +++ b/test/unit/flow-submit.test.ts @@ -0,0 +1,250 @@ +import { expect } from 'chai'; +import * as http from 'node:http'; +import type { AddressInfo } from 'node:net'; + +import { ApiError, ApiGateway } from '../../src/gateways/api-gateway.js'; +import { + isRetryableSubmitError, + submitWithRetry, +} from '../../src/services/flow-submit.service.js'; +import type { AuthContext } from '../../src/types/domain/auth.types.js'; + +/** + * Re-sending a flow submit whose outcome is unknown (dcd#1617 follow-up). + * + * The API answers a repeat of the same body (same upload id) with the run it + * already created, so the CLI can send a submit again after a 502, a timeout + * or a dropped connection without creating a second run. These go through the + * real gateway and the real fetch to a local server, so the network-error and + * timeout cases are the ones undici actually produces. + */ + +const AUTH: AuthContext = { mode: 'apiKey', headers: { 'x-app-api-key': 'k' } }; +const BODY = { id: 'upload-1', path: 'uploads/upload-1/upload-1.zip', name: 'run' }; +type SubmitResponse = Awaited>; +// Only the fields these tests look at; the gateway passes the body through. +const CREATED = { + message: 'Test scheduled successfully.', + results: [{ id: 41, test_file_name: 'flow.yaml', test_upload_id: 'upload-1' }], +} as unknown as SubmitResponse; + +/** What the server does with each request, in order. */ +type Reply = 'drop' | 'hold' | { body: unknown; status: number }; + +let server: http.Server; +let baseUrl: string; +let replies: Reply[]; +let bodies: string[]; + +const submit = () => ApiGateway.submitFlowTest(baseUrl, AUTH, BODY); +const noSleep = () => Promise.resolve(); + +describe('flow submit: retry while the outcome is unknown', () => { + before(async () => { + server = http.createServer((req, res) => { + let raw = ''; + req.on('data', (chunk) => { + raw += chunk; + }); + req.on('end', () => { + bodies.push(raw); + const reply = replies.shift() ?? { body: { message: 'no reply queued' }, status: 500 }; + if (reply === 'drop') { + req.socket.destroy(); + return; + } + + if (reply === 'hold') return; // never answers, like a held request + + res.writeHead(reply.status, { 'content-type': 'application/json' }); + res.end(JSON.stringify(reply.body)); + }); + }); + await new Promise((resolve) => { + server.listen(0, '127.0.0.1', resolve); + }); + baseUrl = `http://127.0.0.1:${(server.address() as AddressInfo).port}`; + }); + + after(async () => { + server.closeAllConnections(); + await new Promise((resolve) => { + server.close(() => resolve()); + }); + }); + + beforeEach(() => { + replies = []; + bodies = []; + }); + + it('sends the same body again after a 502 and returns the run', async () => { + replies.push({ body: { message: 'upstream error' }, status: 502 }, { body: CREATED, status: 201 }); + const notices: string[] = []; + + const { attempts, response } = await submitWithRetry({ + fallback: () => Promise.reject(new Error('must not fall back')), + onRetry: ({ attempt, reason }) => notices.push(`${attempt}:${reason}`), + sleep: noSleep, + submit, + }); + + expect(response).to.deep.equal(CREATED); + expect(attempts).to.equal(2); + expect(bodies).to.have.length(2); + // The upload id is what makes the repeat safe, so it must not change. + expect(bodies[1]).to.equal(bodies[0]); + expect(notices).to.deep.equal(['2:HTTP 502']); + }); + + it('waits out a 409 while the first attempt is still creating the run', async () => { + replies.push( + { body: { code: 'SUBMISSION_IN_PROGRESS', message: 'still being created' }, status: 409 }, + { body: CREATED, status: 201 }, + ); + + const { attempts } = await submitWithRetry({ + fallback: () => Promise.reject(new Error('must not fall back')), + sleep: noSleep, + submit, + }); + + expect(attempts).to.equal(2); + }); + + it('sends again after a dropped connection', async () => { + replies.push('drop', { body: CREATED, status: 201 }); + + const { attempts, response } = await submitWithRetry({ + fallback: () => Promise.reject(new Error('must not fall back')), + sleep: noSleep, + submit, + }); + + expect(response).to.deep.equal(CREATED); + expect(attempts).to.equal(2); + }); + + it('abandons an attempt with no response and sends again', async () => { + const original = ApiGateway.submitAttemptTimeoutMs; + ApiGateway.submitAttemptTimeoutMs = 200; + replies.push('hold', { body: CREATED, status: 201 }); + try { + const reasons: string[] = []; + const { attempts } = await submitWithRetry({ + fallback: () => Promise.reject(new Error('must not fall back')), + onRetry: ({ reason }) => reasons.push(reason), + sleep: noSleep, + submit, + }); + + expect(attempts).to.equal(2); + expect(reasons).to.deep.equal(['no response in time']); + } finally { + ApiGateway.submitAttemptTimeoutMs = original; + server.closeAllConnections(); + } + }); + + it('does not send again when the API refused the request (400)', async () => { + replies.push({ body: { message: 'bad flows' }, status: 400 }); + + const error = await submitWithRetry({ + fallback: () => Promise.reject(new Error('must not fall back')), + sleep: noSleep, + submit, + }).catch((error_: unknown) => error_); + + expect(error).to.be.instanceOf(ApiError); + expect((error as ApiError).status).to.equal(400); + expect(bodies).to.have.length(1); + }); + + it('falls back to multipart only when the first attempt finds no endpoint', async () => { + replies.push({ body: { message: 'Cannot POST' }, status: 404 }); + + const { attempts, response } = await submitWithRetry({ + fallback: (error) => Promise.resolve({ ...CREATED, message: `multipart after ${error.status}` }), + sleep: noSleep, + submit, + }); + + expect(attempts).to.equal(1); + expect(response.message).to.equal('multipart after 404'); + }); + + it('never falls back after a retry: that API already took an attempt', async () => { + replies.push({ body: { message: 'upstream error' }, status: 502 }, { body: { message: 'Cannot POST' }, status: 404 }); + let fellBack = false; + + const error = await submitWithRetry({ + fallback: () => { + fellBack = true; + return Promise.resolve(CREATED); + }, + sleep: noSleep, + submit, + }).catch((error_: unknown) => error_); + + expect(fellBack).to.equal(false); + expect((error as ApiError).status).to.equal(404); + }); + + it('gives up with the last error once the attempts run out', async () => { + replies.push( + { body: { message: 'upstream error' }, status: 503 }, + { body: { message: 'upstream error' }, status: 503 }, + { body: { message: 'upstream error' }, status: 503 }, + ); + + const error = await submitWithRetry({ + delaysMs: [0, 0], + fallback: () => Promise.reject(new Error('must not fall back')), + sleep: noSleep, + submit, + }).catch((error_: unknown) => error_); + + expect((error as ApiError).status).to.equal(503); + expect(bodies).to.have.length(3); + }); + + it('waits the given delays between attempts', async () => { + replies.push( + { body: {}, status: 502 }, + { body: {}, status: 502 }, + { body: CREATED, status: 201 }, + ); + const slept: number[] = []; + + await submitWithRetry({ + delaysMs: [2000, 5000], + fallback: () => Promise.reject(new Error('must not fall back')), + sleep: (ms) => { + slept.push(ms); + return Promise.resolve(); + }, + submit, + }); + + expect(slept).to.deep.equal([2000, 5000]); + }); + + describe('isRetryableSubmitError', () => { + for (const status of [409, 500, 502, 503, 504]) { + it(`retries HTTP ${status}`, () => { + expect(isRetryableSubmitError(new ApiError('x', status))).to.equal(true); + }); + } + + for (const status of [400, 401, 402, 403, 404, 429]) { + it(`does not retry HTTP ${status}`, () => { + expect(isRetryableSubmitError(new ApiError('x', status))).to.equal(false); + }); + } + + it('retries anything that is not an API answer (network, timeout, a body cut off)', () => { + expect(isRetryableSubmitError(new TypeError('terminated'))).to.equal(true); + expect(isRetryableSubmitError(new Error('ECONNRESET'))).to.equal(true); + }); + }); +}); diff --git a/test/unit/telemetry-submit-attempts.test.ts b/test/unit/telemetry-submit-attempts.test.ts new file mode 100644 index 0000000..414aca1 --- /dev/null +++ b/test/unit/telemetry-submit-attempts.test.ts @@ -0,0 +1,57 @@ +import { expect } from 'chai'; + +import { telemetry } from '../../src/services/telemetry.service.js'; + +/** + * How many attempts the flow submit took (dcd#1617 follow-up) rides on the + * lifecycle event as `extra.submit_attempts`. Anything above 1 is a response + * that was lost and a submit that was sent again, which is the number to count + * when judging how often the API stalls under the CLI. + */ +interface SentEvent { + extra?: Record; + message: string; +} + +describe('submit attempts telemetry', () => { + const realFetch = globalThis.fetch; + let events: SentEvent[]; + + beforeEach(function () { + // The singleton reads this once, at import. + if (process.env.DCD_TELEMETRY_DISABLED) this.skip(); + + events = []; + globalThis.fetch = (async (_input: unknown, init?: RequestInit) => { + events.push(...(JSON.parse(String(init?.body)) as { events: SentEvent[] }).events); + return new Response('{}', { status: 202 }); + }) as typeof fetch; + telemetry.configure({ + apiUrl: 'http://localhost:9999', + auth: { headers: { 'x-app-api-key': 'test-key' }, mode: 'apiKey' }, + }); + }); + + afterEach(() => { + globalThis.fetch = realFetch; + }); + + const last = (message: string) => + events.filter((e) => e.message === message).pop(); + + it('puts the attempt count on `command completed`', async () => { + telemetry.recordSubmitAttempts(3); + telemetry.recordCommandSuccess(); + await telemetry.flush(); + + expect(last('command completed')?.extra?.submit_attempts).to.equal(3); + }); + + it('and on `command failed`', async () => { + telemetry.recordSubmitAttempts(9); + telemetry.recordCommandFailure({ error: new Error('boom'), exitCode: 1 }); + await telemetry.flush(); + + expect(last('command failed')?.extra?.submit_attempts).to.equal(9); + }); +}); From 0c8c967cc8c1001031efdc14c2304a8e1493cfbc Mon Sep 17 00:00:00 2001 From: finalerock44 Date: Fri, 16 Oct 2026 14:37:42 +0100 Subject: [PATCH 2/2] fix(cloud): refresh auth between submit retries, and count failed attempts Review on #226: - Every retry reused the auth resolved at the start of the command, and the retries can outlast a `dcd login` access token. The next attempt then got a non-retryable 401, possibly after an earlier attempt had already created the run. Each attempt now calls refreshAuth first, as the poll loops do. A session that can't be refreshed (definitive SessionRefreshError, or no stored session) is not retried. - `submit_attempts` was only recorded on success, so `command failed` never carried it. That is the worst-stall case it exists to measure. submitWithRetry now calls onAttempt before each attempt, and both `dcd cloud` and dcd-mcp record the count there. Also bounds the worst case. No attempt starts more than 5 minutes after the first. With every attempt timing out at 90 s the retries could otherwise hold the command for about 16 minutes; now the worst case is about 6.5. Co-Authored-By: Claude Opus 5.5 --- src/commands/cloud.ts | 13 ++++-- src/mcp/tools/run-cloud-test.ts | 13 +++++- src/services/flow-submit.service.ts | 30 ++++++++++++- test/unit/flow-submit.test.ts | 68 +++++++++++++++++++++++++++++ 4 files changed, 118 insertions(+), 6 deletions(-) diff --git a/src/commands/cloud.ts b/src/commands/cloud.ts index d9d7087..8dabab3 100644 --- a/src/commands/cloud.ts +++ b/src/commands/cloud.ts @@ -32,7 +32,7 @@ import { telemetry } from '../services/telemetry.service.js'; import { TestSubmissionService } from '../services/test-submission.service.js'; import { VersionService } from '../services/version.service.js'; import { isIosMatrixConfig } from '../types/domain/device.types.js'; -import { resolveAuth } from '../utils/auth.js'; +import { refreshAuth, resolveAuth } from '../utils/auth.js'; import { assertMatrixSupported, matrixIsIos, @@ -965,15 +965,22 @@ export const cloudCommand = defineCommand({ // The same body on every attempt: its upload id is what lets the API // answer a repeat with the run it already created (dcd#1617 follow-up). const submitBody = { ...fields, ...storageRef }; + // The retries can outlast a `dcd login` access token, so each attempt + // re-checks it, as the poll loop does. + let submitAuth = auth; const submitted = await submitWithRetry({ fallback: (error) => submitMultipart(error.status), + // Recorded before each attempt, so `command failed` carries it too. + onAttempt: (attempt) => telemetry.recordSubmitAttempts(attempt), onRetry: ({ attempt, attempts, delayMs, reason }) => warnOut( `Could not confirm the test submission (${reason}); sending it again in ${delayMs / 1000}s (attempt ${attempt}/${attempts}). It will not run twice.`, ), - submit: () => ApiGateway.submitFlowTest(apiUrl, auth, submitBody), + async submit() { + submitAuth = await refreshAuth(submitAuth); + return ApiGateway.submitFlowTest(apiUrl, submitAuth, submitBody); + }, }); - telemetry.recordSubmitAttempts(submitted.attempts); return submitted.response; }; diff --git a/src/mcp/tools/run-cloud-test.ts b/src/mcp/tools/run-cloud-test.ts index 1a29495..c57137f 100644 --- a/src/mcp/tools/run-cloud-test.ts +++ b/src/mcp/tools/run-cloud-test.ts @@ -306,15 +306,24 @@ export function registerRunCloudTest(server: McpServer): void { // Re-sent while its outcome is unknown, always with the same body; // see submitWithRetry (dcd#1617 follow-up). const submitBody = { ...fields, ...storageRef }; + // The retries can outlast a `dcd login` access token, so each + // attempt re-checks it, as the wait loop below does. + let submitAuth = auth; const submitted = await submitWithRetry({ fallback: submitMultipart, + // Recorded before each attempt, so a failed tool call carries it. + onAttempt: (attempt) => { + telemetryExtra.submitAttempts = attempt; + }, onRetry: ({ attempt, attempts, delayMs, reason }) => logStderr( `Could not confirm the test submission (${reason}); sending it again in ${delayMs / 1000}s (attempt ${attempt}/${attempts}). It will not run twice.`, ), - submit: () => ApiGateway.submitFlowTest(apiUrl, auth, submitBody), + async submit() { + submitAuth = await refreshAuth(submitAuth); + return ApiGateway.submitFlowTest(apiUrl, submitAuth, submitBody); + }, }); - telemetryExtra.submitAttempts = submitted.attempts; return submitted.response; }; diff --git a/src/services/flow-submit.service.ts b/src/services/flow-submit.service.ts index cabeeeb..8c81f7f 100644 --- a/src/services/flow-submit.service.ts +++ b/src/services/flow-submit.service.ts @@ -1,4 +1,5 @@ import { ApiError } from '../gateways/api-gateway.js'; +import { isDefinitiveAuthFailure } from '../utils/auth.js'; /** * Submitting a flow test when the response may be lost (dcd#1617 follow-up). @@ -21,6 +22,13 @@ export const SUBMIT_RETRY_DELAYS_MS: readonly number[] = [ 2000, 5000, 10_000, 20_000, 30_000, 30_000, 30_000, 30_000, ]; +/** + * No attempt starts more than this long after the first. Each attempt can take + * up to its 90s timeout, so without a cap every attempt timing out would hold + * the command for about 16 minutes; with it, the worst case is about 6.5. + */ +export const SUBMIT_RETRY_BUDGET_MS = 5 * 60_000; + /** * Statuses that leave the outcome unknown (502/503/504 from the edge or the * API), say the API could not finish (500), or say the first attempt is still @@ -38,6 +46,9 @@ const RETRYABLE_STATUSES = new Set([409, 500, 502, 503, 504]); */ export function isRetryableSubmitError(error: unknown): boolean { if (error instanceof ApiError) return RETRYABLE_STATUSES.has(error.status); + // The session can't be refreshed (refused, or logged out): only `dcd login` + // fixes that, so another attempt would fail the same way. + if (isDefinitiveAuthFailure(error)) return false; return true; } @@ -81,13 +92,23 @@ export interface SubmitRetryNotice { * error when it is not retryable or the attempts run out. */ export async function submitWithRetry(opts: { + budgetMs?: number; delaysMs?: readonly number[]; fallback: (error: ApiError) => Promise; + now?: () => number; + /** + * Called before each attempt with its number, counting from 1, so a caller + * can record how many were made whether the submit succeeds or not. + */ + onAttempt?: (attempt: number) => void; onRetry?: (notice: SubmitRetryNotice) => void; sleep?: (ms: number) => Promise; submit: () => Promise; }): Promise<{ attempts: number; response: T }> { const delaysMs = opts.delaysMs ?? SUBMIT_RETRY_DELAYS_MS; + const budgetMs = opts.budgetMs ?? SUBMIT_RETRY_BUDGET_MS; + const now = opts.now ?? Date.now; + const startedAt = now(); const sleep = opts.sleep ?? ((ms: number) => @@ -97,6 +118,7 @@ export async function submitWithRetry(opts: { const attempts = delaysMs.length + 1; for (let attempt = 1; ; attempt++) { + opts.onAttempt?.(attempt); try { return { attempts: attempt, response: await opts.submit() }; } catch (error) { @@ -108,7 +130,13 @@ export async function submitWithRetry(opts: { } const delayMs = delaysMs[attempt - 1]; - if (delayMs === undefined || !isRetryableSubmitError(error)) throw error; + if ( + delayMs === undefined || + !isRetryableSubmitError(error) || + now() - startedAt + delayMs > budgetMs + ) { + throw error; + } opts.onRetry?.({ attempt: attempt + 1, diff --git a/test/unit/flow-submit.test.ts b/test/unit/flow-submit.test.ts index 0d12add..868762a 100644 --- a/test/unit/flow-submit.test.ts +++ b/test/unit/flow-submit.test.ts @@ -3,10 +3,13 @@ import * as http from 'node:http'; import type { AddressInfo } from 'node:net'; import { ApiError, ApiGateway } from '../../src/gateways/api-gateway.js'; +import { SessionRefreshError } from '../../src/gateways/cli-auth-gateway.js'; import { isRetryableSubmitError, + SUBMIT_RETRY_BUDGET_MS, submitWithRetry, } from '../../src/services/flow-submit.service.js'; +import { CliError } from '../../src/utils/cli.js'; import type { AuthContext } from '../../src/types/domain/auth.types.js'; /** @@ -229,6 +232,71 @@ describe('flow submit: retry while the outcome is unknown', () => { expect(slept).to.deep.equal([2000, 5000]); }); + it('reports each attempt before it is sent, so a failure carries the count too', async () => { + replies.push({ body: {}, status: 502 }, { body: { message: 'bad' }, status: 400 }); + const seen: number[] = []; + + await submitWithRetry({ + fallback: () => Promise.reject(new Error('must not fall back')), + onAttempt: (attempt) => seen.push(attempt), + sleep: noSleep, + submit, + }).catch(() => undefined); + + expect(seen).to.deep.equal([1, 2]); + }); + + it('starts no attempt after the time budget is spent', async () => { + // Every attempt "takes" 60s of fake time; the budget allows two waits. + let clock = 0; + let calls = 0; + const error = await submitWithRetry({ + budgetMs: 150_000, + delaysMs: [1000, 1000, 1000, 1000], + fallback: () => Promise.reject(new Error('must not fall back')), + now: () => clock, + sleep: (ms) => { + clock += ms; + return Promise.resolve(); + }, + submit() { + calls += 1; + clock += 60_000; + return Promise.reject(new ApiError('upstream error', 502)); + }, + }).catch((error_: unknown) => error_); + + expect((error as ApiError).status).to.equal(502); + // Attempts start at 0s, 61s and 122s; a fourth would start at 183s. + expect(calls).to.equal(3); + }); + + it('keeps the worst case to minutes, not a quarter of an hour', () => { + // The last attempt starts inside the budget and is cut by its own timeout. + const worstCaseMs = SUBMIT_RETRY_BUDGET_MS + ApiGateway.submitAttemptTimeoutMs; + expect(worstCaseMs).to.be.at.most(7 * 60_000); + }); + + it('does not send again when the session cannot be refreshed', async () => { + for (const refusal of [ + new SessionRefreshError('refresh token revoked', true), + new CliError('No stored session. Run `dcd login`.'), + ]) { + let calls = 0; + const error = await submitWithRetry({ + fallback: () => Promise.reject(new Error('must not fall back')), + sleep: noSleep, + submit() { + calls += 1; + return Promise.reject(refusal); + }, + }).catch((error_: unknown) => error_); + + expect(error).to.equal(refusal); + expect(calls).to.equal(1); + } + }); + describe('isRetryableSubmitError', () => { for (const status of [409, 500, 502, 503, 504]) { it(`retries HTTP ${status}`, () => {