From c1276e0d1cc3566037408784ab22835eda8507a9 Mon Sep 17 00:00:00 2001 From: Tom Beckenham <34339192+tombeckenham@users.noreply.github.com> Date: Tue, 10 Nov 2026 18:33:17 +1100 Subject: [PATCH] Retry a D1 overload once before failing that provider (#635) A spent D1 window waits D1_WINDOW_WAIT_MS and runs the provider once more, in the models poll and the spec-sync shard. A second overload still records ingest_failed. Spec sync writes degraded only after that second failure, and only after the wait, so a recovered retry stays active. --- src/server/ingest/d1-window.ts | 50 +++++ src/server/ingest/poll-models.ts | 41 ++-- src/server/ingest/poll-models.worker.test.ts | 113 ++++++++++- src/server/ingest/sync.ts | 84 ++++++--- src/server/ingest/sync.worker.test.ts | 188 ++++++++++++++++++- 5 files changed, 420 insertions(+), 56 deletions(-) create mode 100644 src/server/ingest/d1-window.ts diff --git a/src/server/ingest/d1-window.ts b/src/server/ingest/d1-window.ts new file mode 100644 index 00000000..5c28394e --- /dev/null +++ b/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((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( + job: 'models-poll' | 'spec-sync', + providerId: string, + work: () => Promise, + wait: (ms: number) => Promise, +): Promise { + 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() + } +} diff --git a/src/server/ingest/poll-models.ts b/src/server/ingest/poll-models.ts index 436e1285..82c0b3ae 100644 --- a/src/server/ingest/poll-models.ts +++ b/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((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> { const outcomes: Array = [] 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 diff --git a/src/server/ingest/poll-models.worker.test.ts b/src/server/ingest/poll-models.worker.test.ts index 6d825960..21447d6d 100644 --- a/src/server/ingest/poll-models.worker.test.ts +++ b/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, + ): Promise { + const batches: Array = [] + vi.stubGlobal( + 'fetch', + vi.fn(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 = [] + 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') + }) }) diff --git a/src/server/ingest/sync.ts b/src/server/ingest/sync.ts index fd132118..3deec3f9 100644 --- a/src/server/ingest/sync.ts +++ b/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 { 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 { + 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 = providerRegistry, + wait: (ms: number) => Promise = sleep, ): Promise> { const outcomes: Array = [] 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 } diff --git a/src/server/ingest/sync.worker.test.ts b/src/server/ingest/sync.worker.test.ts index 4c658578..8acea6dc 100644 --- a/src/server/ingest/sync.worker.test.ts +++ b/src/server/ingest/sync.worker.test.ts @@ -1,9 +1,10 @@ +import { bodyText, requestUrl } from '#/lib/text.ts' import { providerRegistry } from '../providers/index.ts' import { assembleProviderOpenApi } from '../provider-openapi.ts' import { pollProviderModels } from './poll-models.ts' import { provider as gatewayProvider } from '../providers/adapters/cloudflare-ai-gateway.ts' import { getActivitySchemaMap, getEndpointSchema } from '../schemas-api.ts' -import { describe, expect, it } from 'vitest' +import { describe, expect, it, vi } from 'vitest' import { env } from 'cloudflare:test' import { and, eq } from 'drizzle-orm' @@ -17,8 +18,11 @@ import { import type { OpenApiDocument, ProviderConfig } from '../providers/types.ts' import { EXTRACTOR_VERSION } from './bundle.ts' +import { D1_WINDOW_WAIT_MS } from './d1-window.ts' +import { POSTHOG_CAPTURE_URL } from './ingest-signals.ts' +import type { PostHogBatch } from './ingest-signals.ts' import { MODELS_DEV_API_URL } from './retire-models-dev.ts' -import { syncProvider } from './sync.ts' +import { syncAllProviders, syncProvider } from './sync.ts' import type { SyncDeps } from './sync.ts' const ref = (name: string) => ({ $ref: `#/components/schemas/${name}` }) @@ -632,3 +636,183 @@ it('keeps per-model logical schema identities on one real multi-activity HTTP pa providerRegistry.splice(providerRegistry.indexOf(configured), 1) } }) + +describe('D1 overload window on spec sync (issue #635)', () => { + const overload = + 'D1_ERROR: D1 DB is overloaded. Too many resources consumed within a 10 second window' + + async function postedEvents( + run: () => Promise, + ): Promise { + const batches: Array = [] + vi.stubGlobal( + 'fetch', + vi.fn(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) + } + + function overloadProvider(id: string, order: Array, fail: boolean) { + let calls = 0 + const provider = stubProvider(id, fixtureSpec(false)) + provider.fetchSpec = () => { + calls++ + order.push(id) + if (fail || calls === 1) return Promise.reject(new Error(overload)) + return Promise.resolve({ + specs: [fixtureSpec(false)], + sources: [STUB_SOURCE], + outputStrategy: 'post-200' as const, + }) + } + return { provider, calls: () => calls } + } + + it('retries one D1 overload and leaves the provider active', async () => { + const id = 'sync-overload-ok' + const nextId = 'sync-overload-ok-next' + const deps = await freshDeps(id) + await deps.db.insert(providers).values({ + id: nextId, + displayName: 'Stub', + specSourceUrl: 'https://example.com/spec.json', + status: 'degraded', + }) + await deps.db + .update(providers) + .set({ status: 'degraded' }) + .where(eq(providers.id, id)) + deps.secrets.POSTHOG_PROJECT_KEY = 'phc_test' + const order: Array = [] + const first = overloadProvider(id, order, false) + const next = stubProvider(nextId, fixtureSpec(false)) + const original = next.fetchSpec + next.fetchSpec = (providerEnv) => { + order.push(nextId) + return original(providerEnv) + } + const waits: Array = [] + const events = await postedEvents(async () => { + const outcomes = await syncAllProviders( + deps, + [first.provider, next], + (ms) => { + waits.push(ms) + order.push('wait') + return Promise.resolve() + }, + ) + expect(outcomes[0]).toMatchObject({ + providerId: id, + endpointsSeen: 1, + }) + expect(outcomes[0]?.error).toBeUndefined() + expect(outcomes[1]).toMatchObject({ + providerId: nextId, + endpointsSeen: 1, + }) + }) + expect(first.calls()).toBe(2) + expect(order).toEqual([id, 'wait', id, nextId]) + 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 still = await deps.db.query.providers.findFirst({ + where: eq(providers.id, nextId), + }) + expect(still?.status).toBe('active') + }) + + it('records ingest_failed and degraded when the retry overloads too', async () => { + const id = 'sync-overload-fail' + const nextId = 'sync-overload-fail-next' + const deps = await freshDeps(id) + await deps.db.insert(providers).values({ + id: nextId, + displayName: 'Stub', + specSourceUrl: 'https://example.com/spec.json', + }) + deps.secrets.POSTHOG_PROJECT_KEY = 'phc_test' + const order: Array = [] + const first = overloadProvider(id, order, true) + const next = stubProvider(nextId, fixtureSpec(false)) + const original = next.fetchSpec + next.fetchSpec = (providerEnv) => { + order.push(nextId) + return original(providerEnv) + } + const events = await postedEvents(async () => { + const outcomes = await syncAllProviders( + deps, + [first.provider, next], + (ms) => { + order.push('wait') + expect(ms).toBe(D1_WINDOW_WAIT_MS) + return Promise.resolve() + }, + ) + expect(outcomes[0]).toMatchObject({ + providerId: id, + endpointsSeen: 0, + error: overload, + }) + expect(outcomes[1]).toMatchObject({ + providerId: nextId, + endpointsSeen: 1, + }) + }) + expect(first.calls()).toBe(2) + expect(order).toEqual([id, 'wait', id, 'wait', nextId]) + expect( + events.filter((event) => event.event === 'ingest_failed'), + ).toMatchObject([ + { + event: 'ingest_failed', + properties: { job: 'spec-sync', providerId: id, error: overload }, + }, + ]) + const row = await deps.db.query.providers.findFirst({ + where: eq(providers.id, id), + }) + expect(row?.status).toBe('degraded') + expect(await providerEndpoints(deps.db, nextId)).toHaveLength(1) + }) + + it('does not retry a failure that is not a spent D1 window', async () => { + const id = 'sync-overload-other' + const deps = await freshDeps(id) + let calls = 0 + const provider = stubProvider(id, fixtureSpec(false)) + provider.fetchSpec = () => { + calls++ + return Promise.reject(new Error('listing 503')) + } + const waits: Array = [] + const outcomes = await syncAllProviders(deps, [provider], (ms) => { + waits.push(ms) + return Promise.resolve() + }) + expect(calls).toBe(1) + expect(waits).toEqual([]) + expect(outcomes[0]?.error).toContain('listing 503') + const row = await deps.db.query.providers.findFirst({ + where: eq(providers.id, id), + }) + expect(row?.status).toBe('degraded') + }) +})