Skip to content

fix(core): retry transient fetch network errors and buffer compressed multipart payloads #602

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
jhssilva wants to merge 1 commit into
base: master
Choose a base branch
from
Draft

fix(core): retry transient fetch network errors and buffer compressed multipart payloads #602

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
84 changes: 84 additions & 0 deletions packages/core/src/helpers/request.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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')
Expand Down Expand Up @@ -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']);
});
});
});
69 changes: 65 additions & 4 deletions packages/core/src/helpers/request.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()),
};
Expand All @@ -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 <T>(opts: RequestOpts): Promise<T> => {
const {
Expand All @@ -109,15 +158,17 @@ export const doRequest = async <T>(opts: RequestOpts): Promise<T> => {

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',
};

Expand All @@ -141,10 +192,20 @@ export const doRequest = async <T>(opts: RequestOpts): Promise<T> => {
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;
Expand Down
Loading