Function fetchWithRetry

apps/sim/lib/data-drains/destinations/gcs.ts:135–188

(input: RetryRequestInput)

}

async function fetchWithRetry(input: RetryRequestInput): Promise<void> {
  let lastError: unknown
  for (let attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) {
    if (input.signal.aborted) throw input.signal.reason ?? new Error('Aborted')
    const perAttempt = AbortSignal.any([input.signal, AbortSignal.timeout(PER_ATTEMPT_TIMEOUT_MS)])
    let response: Response
    try {
      const headers = await input.buildHeaders()
      response = await fetch(input.url, {
        method: input.method,
        body: input.body as BodyInit | undefined,
        headers,
        signal: perAttempt,
      })
    } catch (error) {
      lastError = error
      logger.debug('GCS request failed', {
        action: input.action,
        attempt,
        bucket: input.bucket,
        error: toError(error).message,
      })
      if (attempt < MAX_ATTEMPTS) {
        await sleepUntilAborted(backoffWithJitter(attempt, null), input.signal)
        continue
      }
      throw error
    }
    if (response.ok || input.successStatuses?.includes(response.status)) {
      /** Drain the success body so undici can return the socket to the keep-alive pool. */
      await response.text().catch(() => '')
      return
    }
    if (!isRetryableStatus(response.status) || attempt === MAX_ATTEMPTS) {
      const text = await response.text().catch(() => '')
      logger.warn('GCS operation failed', {
        action: input.action,
        bucket: input.bucket,
        status: response.status,
      })
      throw new Error(
        `GCS ${input.action} failed (HTTP ${response.status}): ${text || response.statusText}`
      )
    }
    lastError = new Error(`GCS ${input.action} responded with HTTP ${response.status}`)
    const retryAfterMs = parseRetryAfter(response.headers.get('retry-after'))
    /** Drain the retryable response body so undici can return the socket to the keep-alive pool. */
    await response.text().catch(() => '')
    await sleepUntilAborted(backoffWithJitter(attempt, retryAfterMs), input.signal)
  }
  throw lastError instanceof Error
    ? lastError
    : new Error(`GCS ${input.action} failed after retries`)
}

/** GCS uses HTTP headers (x-goog-meta-*) to carry custom metadata; the spec forbids non-ASCII. */
const ASCII_ONLY_RE = /^[\x20-\x7e]*$/