fix(client): retry transient Accelerate failures in RemoteExecutor

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<Response | Error>) {
+  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<ReturnType<AccelerateExtensionFetch>, Parameters<AccelerateExtensionFetch>>(() => {
+    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([])
+  })
+})

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<unknown> {
-    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<void> {
+    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<never> {
     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.
  *

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<number> {
+  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))
+}