Retry a D1 overload once before failing that provider

src/server/ingest/d1-window.ts

@@ -0,0 +1,50 @@
+/**
+ * D1's primary rejects further statements for about 10 seconds after
+ * "too many resources" or the internal-error reset (issue #625, #635).
+ */
+import { errorMessage } from '#/server/errors.ts'
+
+/**
+ * How long to wait after D1 says its 10-second resource window is spent,
+ * before retrying that provider once, and before the next provider (or
+ * the same-as reconcile) touches the primary.
+ */
+export const D1_WINDOW_WAIT_MS = 10_000
+
+/**
+ * D1 overload, and the internal-error sibling that resets the same primary.
+ * Either one means the next statement in the next few seconds fails too.
+ */
+export function d1WindowExhausted(message: string): boolean {
+  const text = message.toLowerCase()
+  if (!text.includes('d1')) return false
+  return text.includes('overloaded') || text.includes('internal error')
+}
+
+export const sleep = (ms: number) =>
+  new Promise<void>((resolve) => {
+    setTimeout(resolve, ms)
+  })
+
+/**
+ * Run `work` once. A spent D1 window waits, then runs it once more.
+ * Anything else, including a second spent window, throws to the caller.
+ */
+export async function retryOnceAfterD1Window<T>(
+  job: 'models-poll' | 'spec-sync',
+  providerId: string,
+  work: () => Promise<T>,
+  wait: (ms: number) => Promise<void>,
+): Promise<T> {
+  try {
+    return await work()
+  } catch (error) {
+    const message = errorMessage(error)
+    if (!d1WindowExhausted(message)) throw error
+    console.error(
+      JSON.stringify({ job, providerId, error: message, retry: 'd1-window' }),
+    )
+    await wait(D1_WINDOW_WAIT_MS)
+    return await work()
+  }
+}

src/server/ingest/poll-models.ts

@@ -58,6 +58,12 @@ import {
   updateStormOf,
 } from './docs-failing.ts'
 import type { DocsFailing } from './docs-failing.ts'
+import {
+  D1_WINDOW_WAIT_MS,
+  d1WindowExhausted,
+  retryOnceAfterD1Window,
+  sleep,
+} from './d1-window.ts'
 import {
   captureIngestEvents,
   runIngestScope,
@@ -1165,26 +1171,7 @@ export async function pollProviderModels(
   return outcome
 }
 
-/**
- * How long to wait after D1 says its 10-second resource window is spent,
- * before the next provider (or the same-as reconcile) touches the primary.
- */
-export const D1_WINDOW_WAIT_MS = 10_000
-
-/**
- * D1 overload, and the internal-error sibling that resets the same primary.
- * Either one means the next statement in the next few seconds fails too.
- */
-export function d1WindowExhausted(message: string): boolean {
-  const text = message.toLowerCase()
-  if (!text.includes('d1')) return false
-  return text.includes('overloaded') || text.includes('internal error')
-}
-
-const sleep = (ms: number) =>
-  new Promise<void>((resolve) => {
-    setTimeout(resolve, ms)
-  })
+export { D1_WINDOW_WAIT_MS, d1WindowExhausted }
 
 /** Poll every registered provider with per-provider failure isolation. */
 export async function pollAllProviders(
@@ -1194,9 +1181,18 @@ export async function pollAllProviders(
 ): Promise<Array<PollOutcome>> {
   const outcomes: Array<PollOutcome> = []
   for (const provider of registry) {
+    // A spent window is retried once (issue #635). The failure recorded
+    // below is the second attempt, or a failure that was not a spent window.
     const exhausted = await runIngestScope(async () => {
       try {
-        outcomes.push(await pollProviderModels(deps, provider))
+        outcomes.push(
+          await retryOnceAfterD1Window(
+            'models-poll',
+            provider.id,
+            () => pollProviderModels(deps, provider),
+            wait,
+          ),
+        )
         return false
       } catch (error) {
         const message = errorMessage(error)
@@ -1220,13 +1216,12 @@ export async function pollAllProviders(
           failures: 1,
           error: message,
         })
-        // Not a retry of this provider. The next one would fail inside the
-        // same window, so let the window close first.
         return d1WindowExhausted(message)
       } finally {
         await captureIngestEvents(deps.secrets.POSTHOG_PROJECT_KEY)
       }
     })
+    // The next provider would fail inside the same window.
     if (exhausted) await wait(D1_WINDOW_WAIT_MS)
   }
   // Links span providers, so resolve them once per run, outside any one

src/server/ingest/poll-models.worker.test.ts

@@ -1,4 +1,5 @@
-import { describe, expect, it } from 'vitest'
+import { bodyText, requestUrl } from '#/lib/text.ts'
+import { describe, expect, it, vi } from 'vitest'
 import { compileTokenCard } from '@modelschemas/rate-card'
 import type { RateCard } from '@modelschemas/rate-card'
 import { FIXTURE_CARDS } from '../../../packages/rate-card/src/fixtures/index.ts'
@@ -32,7 +33,12 @@ import {
   readIngestRecords,
   recordDocsFailing,
 } from './docs-failing.ts'
-import { runIngestScope, takeIngestEvents } from './ingest-signals.ts'
+import {
+  POSTHOG_CAPTURE_URL,
+  runIngestScope,
+  takeIngestEvents,
+} from './ingest-signals.ts'
+import type { PostHogBatch } from './ingest-signals.ts'
 import {
   D1_WINDOW_WAIT_MS,
   LAST_SEEN_CHUNK,
@@ -3052,8 +3058,9 @@ describe('D1 overload window (issue #625)', () => {
       'D1_ERROR: D1 DB is overloaded. Too many resources consumed within a 10 second window',
       'spent',
     )
-    expect(calls).toBe(1)
-    expect(waits).toEqual([D1_WINDOW_WAIT_MS])
+    // One retry, then a wait before the next provider. Not a third attempt.
+    expect(calls).toBe(2)
+    expect(waits).toEqual([D1_WINDOW_WAIT_MS, D1_WINDOW_WAIT_MS])
     expect(D1_WINDOW_WAIT_MS).toBe(10_000)
     expect(outcomes[0]).toMatchObject({
       providerId: 'overload-a-spent',
@@ -3073,7 +3080,8 @@ describe('D1 overload window (issue #625)', () => {
       'D1_ERROR: Internal error in D1 DB storage caused object to be reset.',
       'reset',
     )
-    expect(internal.waits).toEqual([D1_WINDOW_WAIT_MS])
+    expect(internal.calls).toBe(2)
+    expect(internal.waits).toEqual([D1_WINDOW_WAIT_MS, D1_WINDOW_WAIT_MS])
     expect(internal.outcomes[1]?.added).toBe(1)
 
     const other = await twoProviders('listing 503', 'other')
@@ -3082,4 +3090,99 @@ describe('D1 overload window (issue #625)', () => {
     expect(other.outcomes[0]?.error).toContain('listing 503')
     expect(other.outcomes[1]).toMatchObject({ added: 1, failures: 0 })
   })
+
+  const overload =
+    'D1_ERROR: D1 DB is overloaded. Too many resources consumed within a 10 second window'
+
+  async function postedEvents(
+    run: () => Promise<void>,
+  ): Promise<PostHogBatch['batch']> {
+    const batches: Array<PostHogBatch> = []
+    vi.stubGlobal(
+      'fetch',
+      vi.fn<typeof fetch>(async (input, init) => {
+        if (requestUrl(input) === POSTHOG_CAPTURE_URL) {
+          batches.push(JSON.parse(bodyText(init?.body)) as PostHogBatch)
+        }
+        return new Response('', { status: 200 })
+      }),
+    )
+    try {
+      await run()
+    } finally {
+      vi.unstubAllGlobals()
+    }
+    return batches.flatMap((batch) => batch.batch)
+  }
+
+  it('retries one D1 overload and records nothing when the second poll succeeds', async () => {
+    const id = 'overload-retry-ok'
+    const waits: Array<number> = []
+    let calls = 0
+    const provider = stubProvider(id, [{ rawId: 'm' }])
+    provider.listModels = () => {
+      calls++
+      if (calls === 1) return Promise.reject(new Error(overload))
+      return Promise.resolve({ models: [{ rawId: 'm' }] })
+    }
+    const deps = await freshDeps(id)
+    deps.secrets.POSTHOG_PROJECT_KEY = 'phc_test'
+    const events = await postedEvents(async () => {
+      const outcomes = await pollAllProviders(deps, [provider], (ms) => {
+        waits.push(ms)
+        return Promise.resolve()
+      })
+      expect(outcomes).toMatchObject([
+        { providerId: id, added: 1, failures: 0 },
+      ])
+      expect(outcomes[0]?.error).toBeUndefined()
+    })
+    expect(calls).toBe(2)
+    expect(waits).toEqual([D1_WINDOW_WAIT_MS])
+    expect(events.filter((event) => event.event === 'ingest_failed')).toEqual(
+      [],
+    )
+    const row = await deps.db.query.providers.findFirst({
+      where: eq(providers.id, id),
+    })
+    expect(row?.status).toBe('active')
+    const model = await deps.db.query.models.findFirst({
+      where: eq(models.id, modelDbId(id, 'm')),
+    })
+    expect(model?.rawId).toBe('m')
+  })
+
+  it('records ingest_failed when the retry overloads too', async () => {
+    const id = 'overload-retry-fail'
+    let calls = 0
+    const provider = stubProvider(id, [{ rawId: 'm' }])
+    provider.listModels = () => {
+      calls++
+      return Promise.reject(new Error(overload))
+    }
+    const deps = await freshDeps(id)
+    deps.secrets.POSTHOG_PROJECT_KEY = 'phc_test'
+    const events = await postedEvents(async () => {
+      const outcomes = await pollAllProviders(deps, [provider], () =>
+        Promise.resolve(),
+      )
+      expect(outcomes).toMatchObject([
+        { providerId: id, failures: 1, modelsSeen: 0 },
+      ])
+      expect(outcomes[0]?.error).toContain('D1 DB is overloaded')
+    })
+    expect(calls).toBe(2)
+    expect(
+      events.filter((event) => event.event === 'ingest_failed'),
+    ).toMatchObject([
+      {
+        event: 'ingest_failed',
+        properties: { job: 'models-poll', providerId: id, error: overload },
+      },
+    ])
+    const row = await deps.db.query.providers.findFirst({
+      where: eq(providers.id, id),
+    })
+    expect(row?.status).toBe('active')
+  })
 })

src/server/ingest/sync.ts

@@ -35,6 +35,12 @@ import type {
   SpecSource,
 } from '#/server/providers/types.ts'
 import { hasAsyncApiFlag, withAsyncApiFlag } from './asyncapi.ts'
+import {
+  D1_WINDOW_WAIT_MS,
+  d1WindowExhausted,
+  retryOnceAfterD1Window,
+  sleep,
+} from './d1-window.ts'
 import {
   captureIngestEvents,
   ingestFailedEvent,
@@ -654,20 +660,54 @@ export function specSyncShard(shardIndex: number): Array<ProviderConfig> {
   return rest.filter((_, i) => i % restShards === shardIndex - 1)
 }
 
+/**
+ * Best-effort. A spent D1 window fails this write the same way the sync
+ * failed; the caller waits that window out before calling it. A failure
+ * here must not sink the rest of the shard.
+ */
+async function markProviderDegraded(
+  deps: SyncDeps,
+  providerId: string,
+): Promise<void> {
+  try {
+    await deps.db
+      .update(providers)
+      .set({ status: 'degraded' })
+      .where(eq(providers.id, providerId))
+  } catch (statusError) {
+    const message = `degraded-status write failed: ${errorMessage(statusError)}`
+    console.error(
+      JSON.stringify({ job: 'spec-sync', providerId, error: message }),
+    )
+    noteIngest(ingestFailedEvent('spec-sync', providerId, message))
+  }
+}
+
 /**
  * Sync the given providers (default: the full registry), sequentially
  * (subrequest limits), with per-provider isolation: one provider's outage
- * doesn't sink the run.
+ * doesn't sink the run. A spent D1 window is retried once after
+ * `D1_WINDOW_WAIT_MS` (issue #635). A second one still fails, and the
+ * degraded write waits out that window so it is not lost on the same overload.
  */
 export async function syncAllProviders(
   deps: SyncDeps,
   providersToSync: Array<ProviderConfig> = providerRegistry,
+  wait: (ms: number) => Promise<void> = sleep,
 ): Promise<Array<SyncOutcome>> {
   const outcomes: Array<SyncOutcome> = []
   for (const provider of providersToSync) {
-    await runIngestScope(async () => {
+    const exhausted = await runIngestScope(async () => {
       try {
-        outcomes.push(await syncProvider(deps, provider))
+        outcomes.push(
+          await retryOnceAfterD1Window(
+            'spec-sync',
+            provider.id,
+            () => syncProvider(deps, provider),
+            wait,
+          ),
+        )
+        return false
       } catch (error) {
         const message = errorMessage(error)
         // Own log line per failure: the aggregate outcomes blob can exceed
@@ -688,33 +728,25 @@ export async function syncAllProviders(
           error: message,
           warnings: [],
         })
-        try {
-          await deps.db
-            .update(providers)
-            .set({ status: 'degraded' })
-            .where(eq(providers.id, provider.id))
-        } catch (statusError) {
-          // Best-effort: if the failure was resource exhaustion (subrequest
-          // budget), this write fails too — don't let it sink the run.
-          console.error(
-            JSON.stringify({
-              job: 'spec-sync',
-              providerId: provider.id,
-              error: `degraded-status write failed: ${errorMessage(statusError)}`,
-            }),
-          )
-          noteIngest(
-            ingestFailedEvent(
-              'spec-sync',
-              provider.id,
-              `degraded-status write failed: ${errorMessage(statusError)}`,
-            ),
-          )
-        }
+        const spent = d1WindowExhausted(message)
+        // A non-window failure can write status now. A spent window cannot:
+        // that write failed in production on the same overload (issue #635).
+        if (!spent) await markProviderDegraded(deps, provider.id)
+        return spent
       } finally {
         await captureIngestEvents(deps.secrets.POSTHOG_PROJECT_KEY)
       }
     })
+    if (exhausted) {
+      await wait(D1_WINDOW_WAIT_MS)
+      await runIngestScope(async () => {
+        try {
+          await markProviderDegraded(deps, provider.id)
+        } finally {
+          await captureIngestEvents(deps.secrets.POSTHOG_PROJECT_KEY)
+        }
+      })
+    }
   }
   return outcomes
 }