diff --git a/packages/client/src/runtime/core/engines/client/RemoteExecutor.test.ts b/packages/client/src/runtime/core/engines/client/RemoteExecutor.test.ts new file mode 100644 index 000000000000..46e6218acd83 --- /dev/null +++ b/packages/client/src/runtime/core/engines/client/RemoteExecutor.test.ts @@ -0,0 +1,181 @@ +import { PrismaClientKnownRequestError } from '@prisma/client-runtime-utils' +import { EventEmitter } from 'events' + +import { disabledTracingHelper } from '../../tracing/TracingHelper' +import type { AccelerateExtensionFetch } from '../common/Engine' +import type { LogEmitter, LogEvent } from '../common/types/Events' +import { RemoteExecutor } from './RemoteExecutor' + +const ACCELERATE_URL = 'prisma://accelerate.prisma-data.net/?api_key=test-api-key' + +const connectionInfo = { provider: 'postgres', connectionInfo: { supportsRelationJoins: true } } + +function jsonResponse(body: unknown, status = 200): Response { + return new Response(JSON.stringify(body), { status, headers: { 'Content-Type': 'application/json' } }) +} + +function networkError(): TypeError { + return new TypeError('fetch failed', { cause: new Error('other side closed') }) +} + +function setup(responses: Array) { + const warnings: string[] = [] + const logEmitter = new EventEmitter() as unknown as LogEmitter + logEmitter.on('warn', (event: LogEvent) => warnings.push(event.message)) + + const executor = new RemoteExecutor({ + clientVersion: '0.0.0', + accelerateUrl: ACCELERATE_URL, + logEmitter, + logLevel: 'info', + logQueries: false, + tracingHelper: disabledTracingHelper, + }) + + const fetch = jest.fn, Parameters>(() => { + const next = responses.shift() + if (next === undefined) { + throw new Error('unexpected request') + } + return next instanceof Response ? Promise.resolve(next) : Promise.reject(next) + }) + + return { executor, fetch, warnings } +} + +function execute(executor: RemoteExecutor, fetch: AccelerateExtensionFetch) { + return executor.execute({ + plan: { type: 'value', args: 1 } as never, + placeholderValues: {}, + model: 'User', + operation: 'findMany', + transaction: undefined, + batchIndex: undefined, + customFetch: fetch, + queryInfo: { type: 'single', modelName: 'User', action: 'findMany', query: {} }, + }) +} + +describe('RemoteExecutor retries', () => { + let globalFetch: jest.SpyInstance + + beforeEach(() => { + globalFetch = jest.spyOn(globalThis, 'fetch') + }) + + afterEach(() => { + globalFetch.mockRestore() + }) + + test('retries a request whose fetch rejects with a network error', async () => { + const { executor, fetch, warnings } = setup([networkError(), networkError(), jsonResponse({ data: [] })]) + + await expect(execute(executor, fetch)).resolves.toEqual([]) + + expect(fetch).toHaveBeenCalledTimes(3) + expect(warnings).toEqual([ + 'Attempt 1/3 failed for POST /query: fetch failed', + expect.stringMatching(/^Retrying after \d+ms$/), + 'Attempt 2/3 failed for POST /query: fetch failed', + expect.stringMatching(/^Retrying after \d+ms$/), + ]) + }) + + test('sends the same request again on retry', async () => { + const { executor, fetch } = setup([networkError(), jsonResponse({ data: [] })]) + + await execute(executor, fetch) + + const [first, second] = fetch.mock.calls + expect(second[0]).toEqual(first[0]) + expect(second[1].method).toEqual(first[1].method) + expect(second[1].body).toEqual(first[1].body) + }) + + test.each([500, 502, 503])('retries a %d response', async (status) => { + const { executor, fetch, warnings } = setup([ + jsonResponse({ message: 'Worker exceeded resource limits' }, status), + jsonResponse({ data: [] }), + ]) + + await expect(execute(executor, fetch)).resolves.toEqual([]) + + expect(fetch).toHaveBeenCalledTimes(2) + expect(warnings[0]).toEqual(`Attempt 1/3 failed for POST /query: HTTP ${status}`) + }) + + test('does not put the response body of a retried request in the logs', async () => { + const { executor, fetch, warnings } = setup([ + new Response(JSON.stringify({ message: 'internal details: secret-value' }), { + status: 503, + statusText: 'Service Unavailable', + }), + jsonResponse({ data: [] }), + ]) + + await expect(execute(executor, fetch)).resolves.toEqual([]) + + expect(warnings[0]).toEqual('Attempt 1/3 failed for POST /query: HTTP 503 Service Unavailable') + expect(warnings.join('\n')).not.toContain('secret-value') + }) + + test.each([ + ['AbortError', 'This operation was aborted'], + ['TimeoutError', 'The operation was aborted due to timeout'], + ])('does not retry a request cancelled with %s', async (name, message) => { + const error = new DOMException(message, name) + const { executor, fetch, warnings } = setup([error, jsonResponse({ data: [] })]) + + await expect(execute(executor, fetch)).rejects.toBe(error) + + expect(fetch).toHaveBeenCalledTimes(1) + expect(warnings).toEqual([]) + }) + + test('retries requests that use the global fetch', async () => { + const { executor } = setup([]) + globalFetch.mockRejectedValueOnce(networkError()).mockResolvedValueOnce(jsonResponse(connectionInfo)) + + await expect(executor.getConnectionInfo()).resolves.toEqual(connectionInfo) + expect(globalFetch).toHaveBeenCalledTimes(2) + }) + + test('rethrows the network error after the last retry', async () => { + const error = networkError() + const { executor, fetch, warnings } = setup([error, error, error, error]) + + await expect(execute(executor, fetch)).rejects.toBe(error) + + expect(fetch).toHaveBeenCalledTimes(4) + expect(warnings.filter((message) => message.startsWith('Attempt'))).toHaveLength(3) + }) + + test('reports the last server error after the last retry', async () => { + const { executor, fetch } = setup([ + jsonResponse({ message: 'first' }, 503), + jsonResponse({ message: 'second' }, 503), + jsonResponse({ message: 'third' }, 503), + jsonResponse({ message: 'last' }, 503), + ]) + + const result = execute(executor, fetch) + + await expect(result).rejects.toBeInstanceOf(PrismaClientKnownRequestError) + await expect(result).rejects.toMatchObject({ message: 'last', code: 'P6000' }) + expect(fetch).toHaveBeenCalledTimes(4) + }) + + test.each([ + [400, { code: 'P2002', error: 'Unique constraint failed on the fields: (`email`)' }], + [409, { code: 'P2028', error: 'Transaction already closed' }], + [422, { error: 'Query exceeded the response size limit' }], + [504, { message: 'Query timed out' }], + ])('does not retry a %d response', async (status, body) => { + const { executor, fetch, warnings } = setup([jsonResponse(body, status)]) + + await expect(execute(executor, fetch)).rejects.toBeInstanceOf(PrismaClientKnownRequestError) + + expect(fetch).toHaveBeenCalledTimes(1) + expect(warnings).toEqual([]) + }) +}) diff --git a/packages/client/src/runtime/core/engines/client/RemoteExecutor.ts b/packages/client/src/runtime/core/engines/client/RemoteExecutor.ts index 1688937ec48d..3f2228287038 100644 --- a/packages/client/src/runtime/core/engines/client/RemoteExecutor.ts +++ b/packages/client/src/runtime/core/engines/client/RemoteExecutor.ts @@ -13,10 +13,17 @@ import type { LogEmitter } from '../common/types/Events' import type { QueryEngineResultExtensions } from '../common/types/QueryEngine' import type { InteractiveTransactionInfo } from '../common/types/Transaction' import type { ExecutePlanParams, Executor, ProviderAndConnectionInfo } from './Executor' +import { backOff } from './utils/backoff' import { dateFromEngineTimestamp } from './utils/engine-timestamp' const debug = Debug('prisma:client:clientEngine:remoteExecutor') +/** + * How many times a request is retried after a transient failure before the + * error is passed on. Matches the `DataProxyEngine` of Prisma 6. + */ +const MAX_RETRIES = 3 + export interface RemoteExecutorOptions { clientVersion: string logEmitter: LogEmitter @@ -144,13 +151,47 @@ export class RemoteExecutor implements Executor { fetch?: AccelerateExtensionFetch batchRequestIdx?: number }): Promise { - const response = await this.#httpClient.request({ - method, - path, - headers: this.#headerBuilder.build(), - body, - fetch, - }) + let response: Response + + for (let attempt = 0; ; attempt++) { + try { + response = await this.#httpClient.request({ + method, + path, + headers: this.#headerBuilder.build(), + body, + fetch, + }) + } catch (error) { + // `fetch` rejects when the request could not be completed at the + // network level (connection reset, socket closed, DNS failure, ...). + // A cancelled request is not a transient failure, so it is passed on + // straight away. + if (attempt >= MAX_RETRIES || isAbortError(error)) { + throw error + } + await this.#waitBeforeRetry({ attempt, method, path, reason: (error as Error)?.message ?? String(error) }) + continue + } + + if (!response.ok && isRetryableStatus(response.status) && attempt < MAX_RETRIES) { + // Read the body so that the connection can be reused. Only the status + // goes into the log message: the body is whatever the server or a proxy + // in front of it returned, and it is not ours to put in the logs. If + // the last attempt fails too, its body is reported as usual by + // `#throwErrorFromResponse`. + await response.text().catch(() => {}) + await this.#waitBeforeRetry({ + attempt, + method, + path, + reason: `HTTP ${response.status}${response.statusText ? ` ${response.statusText}` : ''}`, + }) + continue + } + + break + } if (!response.ok) { await this.#throwErrorFromResponse(response, batchRequestIdx) @@ -165,6 +206,32 @@ export class RemoteExecutor implements Executor { return responseJson } + async #waitBeforeRetry({ + attempt, + method, + path, + reason, + }: { + attempt: number + method: string + path: string + reason: string + }): Promise { + this.#logEmitter.emit('warn', { + message: `Attempt ${attempt + 1}/${MAX_RETRIES} failed for ${method} ${path}: ${reason}`, + timestamp: new Date(), + target: 'RemoteExecutor', + }) + + const delay = await backOff(attempt) + + this.#logEmitter.emit('warn', { + message: `Retrying after ${delay}ms`, + timestamp: new Date(), + target: 'RemoteExecutor', + }) + } + async #throwErrorFromResponse(response: Response, batchRequestIdx?: number): Promise { let errorCode = response.headers.get('Prisma-Error-Code') @@ -260,6 +327,34 @@ export class RemoteExecutor implements Executor { } } +/** + * Server errors that are retried, the same set the `DataProxyEngine` of + * Prisma 6 retried: any 5xx (an overloaded or restarting instance, a gateway + * or Cloudflare error, ...) except `504 Gateway Timeout`, where the query may + * still be running and sending it again would only make the timeout more + * likely. + * + * Errors the Query Plan Executor reports for the query itself (validation, + * constraint violations, transaction errors, resource limits) use 4xx statuses + * and are never retried. + */ +function isRetryableStatus(status: number): boolean { + return status >= 500 && status !== 504 +} + +/** + * Whether `fetch` rejected because the request was cancelled through an + * `AbortSignal`, either explicitly (`AbortError`) or by a deadline such as + * `AbortSignal.timeout()` (`TimeoutError`). Such requests are not retried. + */ +function isAbortError(error: unknown): boolean { + if (typeof error !== 'object' || error === null) { + return false + } + const name = (error as { name?: unknown }).name + return name === 'AbortError' || name === 'TimeoutError' +} + /** * HTTP client with support for cookies and machine hint headers. * diff --git a/packages/client/src/runtime/core/engines/client/utils/backoff.ts b/packages/client/src/runtime/core/engines/client/utils/backoff.ts new file mode 100644 index 000000000000..a6f5e9913fa4 --- /dev/null +++ b/packages/client/src/runtime/core/engines/client/utils/backoff.ts @@ -0,0 +1,16 @@ +const BACKOFF_INTERVAL_MS = 50 + +/** + * Waits for an exponentially growing, jittered delay before retry number + * `attempt` (0-based) and resolves with the delay that was used. + * + * Same schedule as the `DataProxyEngine` of Prisma 6: roughly 50ms, 100ms, + * 200ms, ... each with up to ±50% jitter. + */ +export function backOff(attempt: number): Promise { + const baseDelay = Math.pow(2, attempt) * BACKOFF_INTERVAL_MS + const jitter = Math.ceil(Math.random() * baseDelay) - Math.ceil(baseDelay / 2) + const delay = baseDelay + jitter + + return new Promise((resolve) => setTimeout(() => resolve(delay), delay)) +}