diff --git a/packages/core/src/helpers/request.test.ts b/packages/core/src/helpers/request.test.ts index b23b263c1..bc43a1979 100644 --- a/packages/core/src/helpers/request.test.ts +++ b/packages/core/src/helpers/request.test.ts @@ -211,6 +211,59 @@ describe('Request Helpers', () => { expect(scope.isDone()).toBe(true); }); + test('Should retry on transient fetch network errors (e.g. ECONNRESET / EPIPE)', async () => { + const { doRequest } = await import('@dd/core/helpers/request'); + const onRetryMock = jest.fn(); + const connResetCause = Object.assign(new Error('write ECONNRESET'), { + code: 'ECONNRESET', + }); + const fetchFailedError = Object.assign(new TypeError('fetch failed'), { + cause: connResetCause, + }); + const fetchImpl = jest + .fn() + .mockRejectedValueOnce(fetchFailedError) + .mockResolvedValueOnce(new Response('{"data":"ok"}')); + + const response = await doRequest({ + ...requestOpts, + retries: 2, + minTimeout: 1, + maxTimeout: 5, + onRetry: onRetryMock, + fetchImpl, + }); + + expect(response).toEqual({ data: 'ok' }); + expect(fetchImpl).toHaveBeenCalledTimes(2); + expect(onRetryMock).toHaveBeenCalledTimes(1); + expect(onRetryMock).toHaveBeenCalledWith(fetchFailedError, 1); + }); + + test('Should bail without retrying when AbortSignal is aborted', async () => { + const { doRequest } = await import('@dd/core/helpers/request'); + const onRetryMock = jest.fn(); + const controller = new AbortController(); + controller.abort(); + const abortError = Object.assign(new Error('The operation was aborted'), { + name: 'AbortError', + }); + const fetchImpl = jest.fn().mockRejectedValue(abortError); + + await expect( + doRequest({ + ...requestOpts, + retries: 3, + signal: controller.signal, + onRetry: onRetryMock, + fetchImpl, + }), + ).rejects.toThrow('The operation was aborted'); + + expect(fetchImpl).toHaveBeenCalledTimes(1); + expect(onRetryMock).not.toHaveBeenCalled(); + }); + test('Should add authentication headers when using API and APP keys.', async () => { const fetchMock = jest .spyOn(global, 'fetch') @@ -292,4 +345,35 @@ describe('Request Helpers', () => { ); }); }); + + describe('createRequestData', () => { + test('Should materialize compressed body and set Content-Length before returning', async () => { + const { createRequestData } = await import('@dd/core/helpers/request'); + let formConsumed = false; + + const { data, headers } = await createRequestData({ + getForm: () => { + const form = new FormData(); + form.append( + 'event', + new Blob([JSON.stringify({ service: 'web-ui' })], { + type: 'application/json', + }), + 'event', + ); + formConsumed = true; + return form; + }, + defaultHeaders: { 'DD-EVP-ORIGIN': 'rspack-build-plugin_sourcemaps' }, + zip: true, + }); + + expect(formConsumed).toBe(true); + expect(headers['Content-Encoding']).toBe('gzip'); + expect(Number(headers['Content-Length'])).toBeGreaterThan(0); + + const compressedBytes = new Uint8Array(await new Response(data).arrayBuffer()); + expect(String(compressedBytes.byteLength)).toBe(headers['Content-Length']); + }); + }); }); diff --git a/packages/core/src/helpers/request.ts b/packages/core/src/helpers/request.ts index b01a100ca..f6e1e9e33 100644 --- a/packages/core/src/helpers/request.ts +++ b/packages/core/src/helpers/request.ts @@ -74,10 +74,17 @@ export const createRequestData = async (options: { // Node.js pipe() immediately puts the source into flowing mode (starts reading blobs // via process.nextTick), which races with cleanup of file-backed blobs after the // request completes. Web Streams only start reading when the output is consumed. - const data = zip ? req.body!.pipeThrough(new CompressionStream('gzip')) : req.body!; + const rawStream = zip ? req.body!.pipeThrough(new CompressionStream('gzip')) : req.body!; + + // Materialize the payload in memory before opening the HTTP request so disk I/O and + // gzip compression do not stall an already-open request stream against intake's + // 30s/60s body-receive deadlines. + const bodyBuffer = new Uint8Array(await new Response(rawStream).arrayBuffer()); + const data = new Blob([bodyBuffer]).stream(); const headers = { 'Content-Encoding': zip ? 'gzip' : 'multipart/form-data', + 'Content-Length': String(bodyBuffer.byteLength), ...defaultHeaders, ...Object.fromEntries(req.headers.entries()), }; @@ -88,6 +95,48 @@ export const createRequestData = async (options: { export const ERROR_CODES_NO_RETRY = [400, 401, 403, 404, 405, 409, 413]; export const NB_RETRIES = 5; +const RETRIABLE_NETWORK_ERROR_CODES = new Set([ + 'ECONNRESET', + 'EPIPE', + 'ETIMEDOUT', + 'ECONNREFUSED', + 'ECONNABORTED', + 'EHOSTUNREACH', + 'ENETUNREACH', + 'EAI_AGAIN', +]); + +const hasRetriableErrorCode = (err: unknown): boolean => { + if (err === null || typeof err !== 'object') { + return false; + } + const code = 'code' in err && typeof err.code === 'string' ? err.code : undefined; + if (!code) { + return false; + } + return RETRIABLE_NETWORK_ERROR_CODES.has(code) || code.startsWith('UND_ERR_'); +}; + +export const isRetriableFetchError = (error: unknown, signal?: AbortSignal): boolean => { + if (signal?.aborted) { + return false; + } + if (error === null || typeof error !== 'object') { + return false; + } + if ('name' in error && (error.name === 'AbortError' || error.name === 'TimeoutError')) { + return false; + } + const cause = 'cause' in error ? error.cause : undefined; + if (hasRetriableErrorCode(error) || hasRetriableErrorCode(cause)) { + return true; + } + if ('message' in error && error.message === 'fetch failed') { + return true; + } + return false; +}; + // Do a retriable fetch. export const doRequest = async (opts: RequestOpts): Promise => { const { @@ -109,15 +158,17 @@ export const doRequest = async (opts: RequestOpts): Promise => { return retry(async (bail: (e: Error) => void, attempt: number) => { let response: Response; + let requestInit: RequestInit; + let requestHeaders: RequestInit['headers']; try { - const requestInit: RequestInit = { + requestInit = { method, // This is needed for sending body in NodeJS' Fetch. // https://github.com/nodejs/node/issues/46221 duplex: 'half', signal, }; - let requestHeaders: RequestInit['headers'] = { + requestHeaders = { 'X-Datadog-Origin': 'build-plugins', }; @@ -141,10 +192,20 @@ export const doRequest = async (opts: RequestOpts): Promise => { requestInit.body = data; requestHeaders = { ...requestHeaders, ...headers }; } + } catch (error: any) { + // We don't want to retry if there is a non-fetch related error. + bail(error); + // bail(error) throws so the return is never executed. + return {} as T; + } + try { response = await fetchImpl(url, { ...requestInit, headers: requestHeaders }); } catch (error: any) { - // We don't want to retry if there is a non-fetch related error. + if (isRetriableFetchError(error, signal)) { + throw error; + } + // Bail on non-retriable fetch errors (e.g. locked stream, AbortSignal timeout). bail(error); // bail(error) throws so the return is never executed. return {} as T;