diff --git a/.github/workflows/test-build.yml b/.github/workflows/test-build.yml index d72ced902c8..61505fbc98b 100644 --- a/.github/workflows/test-build.yml +++ b/.github/workflows/test-build.yml @@ -102,6 +102,7 @@ jobs: bunx vitest run scripts/retired-columns.postgres.test.ts scripts/connector-sync-schedule-precision.postgres.test.ts + scripts/database-failure-classification.postgres.test.ts - name: Verify OAuth lifecycle and SCIM membership guards in PostgreSQL working-directory: apps/sim diff --git a/apps/sim/background/knowledge-processing.test.ts b/apps/sim/background/knowledge-processing.test.ts index e9c327d988f..6da47b1438e 100644 --- a/apps/sim/background/knowledge-processing.test.ts +++ b/apps/sim/background/knowledge-processing.test.ts @@ -46,6 +46,10 @@ import { MAX_PROVIDER_CONTINUATION_ATTEMPTS } from '@/lib/knowledge/documents/pr import { MAX_QUOTA_CONTINUATION_ATTEMPTS } from '@/lib/knowledge/documents/processing-quota-continuation' import type { DocumentProcessingAttemptContext } from '@/lib/knowledge/documents/service' import { + DOCUMENT_PROCESSING_RETRY_POLICY, + DocumentProcessingDatabaseRetryError, + getDocumentProcessingRetry, + processDocument, resolveQuotaContinuationDelayMs, runDocumentProcessing, } from '@/background/knowledge-processing' @@ -607,10 +611,83 @@ describe('knowledge processing worker', () => { ) expect(failure).toBeInstanceOf(Error) expect(failure).toMatchObject({ message: 'Database request failed (SQLSTATE 57014).' }) - expect(failure).not.toHaveProperty('cause') + /** Trigger records only the name, message and stack; the cause stays for classification. */ + expect((failure as Error).cause).toBe(error) + expect((failure as Error).stack).not.toContain('private') expect(JSON.stringify(failure)).not.toContain('private') }) + describe('transient database failures', () => { + const MINUTE = 60 * 1000 + const statementTimeout = () => + new DrizzleQueryError( + 'insert private SQL', + ['private bound content'], + Object.assign(new Error('canceling statement due to statement timeout'), { + code: '57014', + }) + ) + + /** A service failure that asks the worker whether to schedule a database retry, as the service does. */ + function failProcessingWith(error: Error): { scheduled: Array } { + const scheduled: Array = [] + mockProcessDocumentAsync.mockImplementation(async (...args: unknown[]) => { + const context = args[6] as DocumentProcessingAttemptContext + scheduled.push(context.scheduleDatabaseRetry?.(error) ?? null) + throw error + }) + return { scheduled } + } + + it('schedules a minute-scale retry and hands Trigger the same time the document records', async () => { + const error = statementTimeout() + const { scheduled } = failProcessingWith(error) + const startedAt = Date.now() + + const failure = await runDocumentProcessing(WORKSPACE_PAYLOAD, 1).catch( + (caught: unknown) => caught + ) + + expect(failure).toBeInstanceOf(DocumentProcessingDatabaseRetryError) + expect(failure).toMatchObject({ message: 'Database request failed (SQLSTATE 57014).' }) + expect((failure as Error).cause).toBe(error) + expect((failure as Error).stack).not.toContain('private') + const retryAt = scheduled[0] + expect(retryAt).toBeInstanceOf(Date) + expect(retryAt!.getTime() - startedAt).toBeGreaterThanOrEqual(2 * MINUTE * 0.8) + expect(retryAt!.getTime() - startedAt).toBeLessThanOrEqual(2 * MINUTE * 1.2 + 1000) + expect(getDocumentProcessingRetry(failure, 1)).toEqual({ retryAt }) + }) + + it('records the failure and stops once the database attempts are spent', async () => { + const error = statementTimeout() + const { scheduled } = failProcessingWith(error) + const lastAttempt = DOCUMENT_PROCESSING_RETRY_POLICY.database.maxAttempts + + const failure = await runDocumentProcessing(WORKSPACE_PAYLOAD, lastAttempt).catch( + (caught: unknown) => caught + ) + + expect(scheduled).toEqual([null]) + expect(failure).not.toBeInstanceOf(DocumentProcessingDatabaseRetryError) + expect((failure as Error).cause).toBe(error) + expect(getDocumentProcessingRetry(failure, lastAttempt)).toEqual({ skipRetrying: true }) + }) + + it('leaves other failures on the task retry settings and attempt count', async () => { + const error = new Error('Storage request timed out') + const { scheduled } = failProcessingWith(error) + + await expect(runDocumentProcessing(WORKSPACE_PAYLOAD, 1)).rejects.toBe(error) + + expect(scheduled).toEqual([null]) + expect(getDocumentProcessingRetry(error, 1)).toBeUndefined() + expect( + getDocumentProcessingRetry(error, DOCUMENT_PROCESSING_RETRY_POLICY.maxAttempts) + ).toEqual({ skipRetrying: true }) + }) + }) + it('retries failed provider continuation dispatch instead of reporting a successful deferral', async () => { const error = new Error('Trigger dispatch unavailable') mockTrigger.mockRejectedValue(error) @@ -745,11 +822,29 @@ describe('knowledge-process-document task configuration', () => { * `attempt_count = 1`, so each was left `failed` having never been retried. */ it('escalates to a larger machine on an out-of-memory kill', async () => { - const { processDocument } = await import('@/background/knowledge-processing') - expect(processDocument.retry?.outOfMemory?.machine).toBe('large-2x') }) + it('declares enough attempts for database retries and routes failures through catchError', async () => { + expect(processDocument.retry?.maxAttempts).toBe( + Math.max( + DOCUMENT_PROCESSING_RETRY_POLICY.maxAttempts, + DOCUMENT_PROCESSING_RETRY_POLICY.database.maxAttempts + ) + ) + const retryAt = new Date('2026-01-01T00:02:00.000Z') + const scheduled = new DocumentProcessingDatabaseRetryError( + 'Database request failed.', + retryAt, + { + cause: new Error('private'), + } + ) + await expect( + processDocument.catchError?.({ error: scheduled, ctx: { attempt: { number: 1 } } } as never) + ).resolves.toEqual({ retryAt }) + }) + it('backs durable quota continuations off to a bounded polling interval', () => { const first = resolveQuotaContinuationDelayMs(1) const second = resolveQuotaContinuationDelayMs(2) diff --git a/apps/sim/background/knowledge-processing.ts b/apps/sim/background/knowledge-processing.ts index 1964e315958..e4352cf4968 100644 --- a/apps/sim/background/knowledge-processing.ts +++ b/apps/sim/background/knowledge-processing.ts @@ -1,6 +1,14 @@ import { createLogger } from '@sim/logger' +import { findCause } from '@sim/utils/errors' import { queue, task } from '@trigger.dev/sdk' import { env, envNumber } from '@/lib/core/config/env' +import { + type BackgroundRetryDecision, + type BackgroundRetryPolicy, + backgroundRetryAttemptCeiling, + getBackgroundRetryDecision, + getDatabaseRetryAt, +} from '@/lib/core/errors/background-retry' import { BYOK_EMBEDDING_CREDENTIAL_REJECTION_MESSAGE, EMBEDDING_QUOTA_EXHAUSTED_MESSAGE, @@ -39,6 +47,46 @@ import { processDocumentAsync } from '@/lib/knowledge/documents/service' const logger = createLogger('TriggerKnowledgeProcessing') export { resolveQuotaContinuationDelayMs } +/** + * Ordinary failures keep the configured short retries. A transient database failure backs off for + * minutes, about an hour in total, so a slow database window does not exhaust every attempt inside + * it and leave an uploaded document failed for good. + */ +export const DOCUMENT_PROCESSING_RETRY_POLICY: BackgroundRetryPolicy = { + maxAttempts: envNumber(env.KB_CONFIG_MAX_ATTEMPTS, 3), + database: { maxAttempts: 6, baseDelayMs: 2 * 60 * 1000, maxDelayMs: 30 * 60 * 1000 }, +} + +/** + * A database failure whose next attempt is already scheduled, and recorded on the document as + * `pending` until {@link retryAt}. The message names only the database code; the driver error + * stays in `cause`, which the task runner does not record. + */ +export class DocumentProcessingDatabaseRetryError extends Error { + constructor( + message: string, + readonly retryAt: Date, + options: { cause: unknown } + ) { + super(message, options) + this.name = 'DocumentProcessingDatabaseRetryError' + } +} + +/** The `catchError` decision for `knowledge-process-document` after `attempt` (1-based) failed. */ +export function getDocumentProcessingRetry( + error: unknown, + attempt: number +): BackgroundRetryDecision { + const scheduled = findCause( + error, + (value): value is DocumentProcessingDatabaseRetryError => + value instanceof DocumentProcessingDatabaseRetryError + ) + if (scheduled) return { retryAt: scheduled.retryAt } + return getBackgroundRetryDecision(error, attempt, DOCUMENT_PROCESSING_RETRY_POLICY) +} + export async function runDocumentProcessing( rawPayload: DocumentProcessingPayload, attemptNumber = 1 @@ -56,6 +104,8 @@ export async function runDocumentProcessing( payload.processingSliceCount === undefined logger.info(`[${requestId}] Starting Trigger.dev processing for document: ${docData.filename}`) + /** Set from the service's callback, so control-flow narrowing cannot see it change. */ + let databaseRetryAt = null as Date | null try { const result = await processDocumentAsync( @@ -87,6 +137,14 @@ export async function runDocumentProcessing( : { quotaContinuationExhausted: true }), scheduleProviderContinuation: (error) => scheduleDocumentProcessingProviderContinuation(payload, error, true, chargedAtDispatch), + scheduleDatabaseRetry: (error) => { + databaseRetryAt = getDatabaseRetryAt( + error, + attemptNumber, + DOCUMENT_PROCESSING_RETRY_POLICY + ) + return databaseRetryAt + }, } ) @@ -100,6 +158,20 @@ export async function runDocumentProcessing( processingTime: Date.now() - startedAt, } } catch (error) { + if (databaseRetryAt) { + const diagnostic = getConnectorFailureDiagnostic(error) + logger.warn(`[${requestId}] Document processing will retry after a database failure`, { + documentId, + diagnostic, + attempt: attemptNumber, + retryAt: databaseRetryAt.toISOString(), + }) + throw new DocumentProcessingDatabaseRetryError( + diagnostic?.message ?? 'Database request failed.', + databaseRetryAt, + { cause: error } + ) + } const providerDeferral = getProviderCapacityDeferral(error) if (providerDeferral || error instanceof ProviderCapacityContinuationExhaustedError) { const outcome = @@ -205,7 +277,8 @@ export async function runDocumentProcessing( `[${requestId}] Failed to process document: ${docData.filename}`, diagnostic ?? error ) - if (diagnostic?.category === 'database') throw new Error(diagnostic.message) + /** Trigger records the thrown message and stack, never `cause`; Drizzle's message carries SQL. */ + if (diagnostic?.category === 'database') throw new Error(diagnostic.message, { cause: error }) throw error } } @@ -253,7 +326,13 @@ export const processDocument = task({ */ machine: 'medium-2x', retry: { - maxAttempts: envNumber(env.KB_CONFIG_MAX_ATTEMPTS, 3), + /** + * The ceiling for thrown errors: database retries use all of it, and + * `catchError` stops every other thrown error at `KB_CONFIG_MAX_ATTEMPTS`. + * A crashed or timed-out run is not retried; an out-of-memory kill is + * retried once, on the `outOfMemory` machine below. + */ + maxAttempts: backgroundRetryAttemptCeiling(DOCUMENT_PROCESSING_RETRY_POLICY), factor: envNumber(env.KB_CONFIG_RETRY_FACTOR, 2), minTimeoutInMs: envNumber(env.KB_CONFIG_MIN_TIMEOUT, 1000), maxTimeoutInMs: envNumber(env.KB_CONFIG_MAX_TIMEOUT, 10000), @@ -272,4 +351,5 @@ export const processDocument = task({ queue: interactiveProcessingQueue, run: (payload: DocumentProcessingPayload, { ctx }) => runDocumentProcessing(payload, ctx.attempt.number), + catchError: async ({ error, ctx }) => getDocumentProcessingRetry(error, ctx.attempt.number), }) diff --git a/apps/sim/lib/core/errors/background-retry.test.ts b/apps/sim/lib/core/errors/background-retry.test.ts new file mode 100644 index 00000000000..cf97bea7019 --- /dev/null +++ b/apps/sim/lib/core/errors/background-retry.test.ts @@ -0,0 +1,95 @@ +/** + * @vitest-environment node + */ +import { DrizzleQueryError } from 'drizzle-orm/errors' +import { describe, expect, it } from 'vitest' +import { + type BackgroundRetryDecision, + type BackgroundRetryPolicy, + backgroundRetryAttemptCeiling, + getBackgroundRetryDecision, + getDatabaseRetryAt, +} from '@/lib/core/errors/background-retry' + +const MINUTE = 60 * 1000 +const POLICY: BackgroundRetryPolicy = { + maxAttempts: 3, + database: { maxAttempts: 6, baseDelayMs: 2 * MINUTE, maxDelayMs: 30 * MINUTE }, +} +const NOW = Date.parse('2026-01-01T00:00:00.000Z') + +function failedQuery(code: string, message = 'private driver detail'): DrizzleQueryError { + return new DrizzleQueryError( + 'private SQL', + ['private'], + Object.assign(new Error(message), { code }) + ) +} + +function delayOf(decision: BackgroundRetryDecision): number { + if (!decision || !('retryAt' in decision)) throw new Error('expected a scheduled retry') + return decision.retryAt.getTime() - NOW +} + +describe('getBackgroundRetryDecision', () => { + it.each([ + ['capacity', failedQuery('57014', 'canceling statement due to statement timeout')], + ['conflict', failedQuery('40P01', 'deadlock detected')], + ['connection', failedQuery('CONNECTION_CLOSED')], + ])('waits minutes after a %s failure', (_label, error) => { + const delay = delayOf(getBackgroundRetryDecision(error, 1, POLICY, NOW)) + expect(delay).toBeGreaterThanOrEqual(2 * MINUTE * 0.8) + expect(delay).toBeLessThanOrEqual(2 * MINUTE * 1.2) + }) + + it('doubles the delay per attempt up to the ceiling', () => { + const error = failedQuery('55P03') + const delays = [1, 2, 3, 4, 5].map((attempt) => + delayOf(getBackgroundRetryDecision(error, attempt, POLICY, NOW)) + ) + const bases = [2, 4, 8, 16, 30].map((minutes) => minutes * MINUTE) + delays.forEach((delay, index) => { + expect(delay).toBeGreaterThanOrEqual(bases[index] * 0.8) + expect(delay).toBeLessThanOrEqual(bases[index] * 1.2) + }) + const longPolicy = { ...POLICY, database: { ...POLICY.database, maxAttempts: 20 } } + expect(delayOf(getBackgroundRetryDecision(error, 12, longPolicy, NOW))).toBeLessThanOrEqual( + 30 * MINUTE * 1.2 + ) + }) + + it('stops database retries at their own attempt ceiling', () => { + const error = failedQuery('53300') + expect(getBackgroundRetryDecision(error, 5, POLICY, NOW)).toHaveProperty('retryAt') + expect(getBackgroundRetryDecision(error, 6, POLICY, NOW)).toEqual({ skipRetrying: true }) + }) + + it('keeps the task default for other failures until their attempt ceiling', () => { + const error = failedQuery('23505') + expect(getBackgroundRetryDecision(error, 1, POLICY, NOW)).toBeUndefined() + expect(getBackgroundRetryDecision(error, 2, POLICY, NOW)).toBeUndefined() + expect(getBackgroundRetryDecision(error, 3, POLICY, NOW)).toEqual({ skipRetrying: true }) + }) + + it('does not stretch an explicit cancellation onto the database pacing', () => { + const cancelled = failedQuery('57014', 'canceling statement due to user request') + expect(getBackgroundRetryDecision(cancelled, 1, POLICY, NOW)).toBeUndefined() + }) +}) + +describe('getDatabaseRetryAt', () => { + it('returns null for a failure that is not a transient database failure', () => { + expect(getDatabaseRetryAt(new Error('parser failed'), 1, POLICY, NOW)).toBeNull() + }) + + it('returns null once the database attempts are spent', () => { + expect(getDatabaseRetryAt(failedQuery('40001'), 6, POLICY, NOW)).toBeNull() + }) +}) + +describe('backgroundRetryAttemptCeiling', () => { + it('covers whichever kind of failure retries longer', () => { + expect(backgroundRetryAttemptCeiling(POLICY)).toBe(6) + expect(backgroundRetryAttemptCeiling({ ...POLICY, maxAttempts: 8 })).toBe(8) + }) +}) diff --git a/apps/sim/lib/core/errors/background-retry.ts b/apps/sim/lib/core/errors/background-retry.ts new file mode 100644 index 00000000000..6ac52429e4a --- /dev/null +++ b/apps/sim/lib/core/errors/background-retry.ts @@ -0,0 +1,65 @@ +import { getTransientDatabaseFailure } from '@sim/utils/errors' +import { backoffWithJitter } from '@sim/utils/retry' + +/** + * What a Trigger.dev `catchError` hook returns: retry at a chosen time, stop retrying, or + * (`undefined`) fall back to the task's own `retry` settings. + */ +export type BackgroundRetryDecision = { retryAt: Date } | { skipRetrying: true } | undefined + +export interface BackgroundRetryPolicy { + /** Attempts for every failure other than a transient database failure. */ + maxAttempts: number + /** + * Attempts and pacing when the database was transiently unavailable (capacity, conflict, or + * connection). Delays are minutes, not seconds, so the retries outlast a slow window instead of + * all landing inside it; the attempt ceiling keeps a failure that only looks transient bounded. + */ + database: { + maxAttempts: number + baseDelayMs: number + maxDelayMs: number + } +} + +/** The `retry.maxAttempts` a task must declare so neither kind of failure is cut short. */ +export function backgroundRetryAttemptCeiling(policy: BackgroundRetryPolicy): number { + return Math.max(policy.maxAttempts, policy.database.maxAttempts) +} + +/** + * When to run the next attempt after `attempt` (1-based) failed on a transient database failure, + * or `null` when the error is not one or its attempts are spent. + */ +export function getDatabaseRetryAt( + error: unknown, + attempt: number, + policy: BackgroundRetryPolicy, + now = Date.now() +): Date | null { + if (!getTransientDatabaseFailure(error)) return null + if (attempt >= policy.database.maxAttempts) return null + const delayMs = backoffWithJitter(attempt, null, { + baseMs: policy.database.baseDelayMs, + maxMs: policy.database.maxDelayMs, + }) + return new Date(now + delayMs) +} + +/** + * Chooses the next attempt after `attempt` (1-based) failed. A transient database failure backs + * off on the policy's database pacing up to its own ceiling; anything else keeps the task's + * ordinary retries and stops at `maxAttempts`. + */ +export function getBackgroundRetryDecision( + error: unknown, + attempt: number, + policy: BackgroundRetryPolicy, + now = Date.now() +): BackgroundRetryDecision { + if (getTransientDatabaseFailure(error)) { + const retryAt = getDatabaseRetryAt(error, attempt, policy, now) + return retryAt ? { retryAt } : { skipRetrying: true } + } + return attempt >= policy.maxAttempts ? { skipRetrying: true } : undefined +} diff --git a/apps/sim/lib/knowledge/__integration__/stored-document-recovery.integration.ts b/apps/sim/lib/knowledge/__integration__/stored-document-recovery.integration.ts index b7a8b45e035..dc7b1dcecd0 100644 --- a/apps/sim/lib/knowledge/__integration__/stored-document-recovery.integration.ts +++ b/apps/sim/lib/knowledge/__integration__/stored-document-recovery.integration.ts @@ -82,7 +82,11 @@ import { searchScopedKnowledge } from '@/lib/knowledge/application/workspace-sea import { createContentSyncLease } from '@/lib/knowledge/connectors/sync-lock' import { addDocument } from '@/lib/knowledge/connectors/sync-persistence' import { sweepStuckDocuments } from '@/lib/knowledge/connectors/sync-primitives' -import { enqueueKnowledgeDocumentProcessing } from '@/lib/knowledge/documents/processing-outbox-event' +import { DEFERRED_RETRY_LOST_ERROR } from '@/lib/knowledge/documents/deferred-retry-check' +import { + enqueueKnowledgeDocumentProcessing, + KNOWLEDGE_DOCUMENT_DEFERRED_RETRY_CHECK_EVENT, +} from '@/lib/knowledge/documents/processing-outbox-event' import { knowledgeDocumentProcessingOutboxHandlers } from '@/lib/knowledge/documents/processing-outbox-handler' import { DOCUMENT_RECOVERY_BATCH_SIZE, @@ -1053,3 +1057,211 @@ describe('independent recovery of retained connector documents', () => { } }) }) + +describe('uploaded documents whose scheduled database retry is lost', () => { + const GENERATION = 'deferred-upload-generation' + + async function retryChecksFor(documentId: string) { + return db + .select() + .from(outboxEvent) + .where( + and( + eq(outboxEvent.eventType, KNOWLEDGE_DOCUMENT_DEFERRED_RETRY_CHECK_EVENT), + sql`${outboxEvent.payload}->>'documentId' = ${documentId}` + ) + ) + } + + /** Throws once, right after the claim, the way a database capacity window fails a run. */ + function failNextRunAfterClaim(error: Error) { + return vi + .spyOn(billingAttribution, 'assertBillingAttributionOwner') + .mockImplementationOnce(() => { + throw error + }) + } + + /** An uploaded document whose run hit a lock timeout and scheduled a retry of the same run. */ + async function deferredUpload(queuedAt: Date | null) { + const ids = await seed() + const file = await failedFile(ids) + await db + .update(document) + .set({ + connectorId: null, + processingStatus: 'pending', + processingQueueToken: GENERATION, + processingQueuedAt: queuedAt, + processingCompletedAt: null, + processingError: null, + uploadedAt: new Date(), + }) + .where(eq(document.id, file.documentId)) + const billing = await resolveSystemBillingAttribution(ids.workspaceId) + const retryAt = new Date(Date.now() + 120_000) + const runRetry = (options: { onClaimed?: () => void } = {}) => + processDocumentAsync(ids.knowledgeBaseId, file.documentId, file, {}, billing, GENERATION, { + processingQueueToken: GENERATION, + chargedAtDispatch: false, + scheduleDatabaseRetry: () => retryAt, + ...options, + }) + const spy = failNextRunAfterClaim( + Object.assign(new Error('canceling statement due to lock timeout'), { code: '55P03' }) + ) + try { + await expect(runRetry()).rejects.toMatchObject({ code: '55P03' }) + } finally { + spy.mockRestore() + } + const [check] = await retryChecksFor(file.documentId) + return { ids, file, billing, retryAt, check, runRetry } + } + + async function runCheckAt(eventId: string, at: number) { + vi.useFakeTimers({ toFake: ['Date'] }) + vi.setSystemTime(at) + try { + return await outbox.processOutboxEventById(eventId, knowledgeDocumentProcessingOutboxHandlers) + } finally { + vi.useRealTimers() + } + } + + const overdue = (retryAt: Date) => retryAt.getTime() + QUEUED_DISPATCH_GRACE_MS + 60_000 + + it('commits the deferral and its check together, due once the retry is past the grace', async () => { + const { file, retryAt, check } = await deferredUpload(new Date()) + const [row] = await db.select().from(document).where(eq(document.id, file.documentId)) + expect(row).toMatchObject({ processingStatus: 'pending', processingDeferredUntil: retryAt }) + expect(check.availableAt).toEqual(new Date(retryAt.getTime() + QUEUED_DISPATCH_GRACE_MS)) + expect(check.payload).toMatchObject({ + documentId: file.documentId, + processingQueueToken: GENERATION, + processingDeferredUntil: retryAt.toISOString(), + }) + expect( + await outbox.processOutboxEventById(check.id, knowledgeDocumentProcessingOutboxHandlers) + ).toBe('pending') + }) + + it('fails the document once its scheduled retry is overdue and no run is live', async () => { + const { file, retryAt, check } = await deferredUpload(new Date()) + expect(await runCheckAt(check.id, overdue(retryAt))).toBe('completed') + const [row] = await db.select().from(document).where(eq(document.id, file.documentId)) + expect(row).toMatchObject({ + processingStatus: 'failed', + processingError: DEFERRED_RETRY_LOST_ERROR, + processingDeferredUntil: null, + processingQueueToken: GENERATION, + }) + expect(row.processingCompletedAt).not.toBeNull() + }) + + it('checks again later, without spending an attempt, while the retry run is live', async () => { + const { file, retryAt, check } = await deferredUpload(new Date()) + fixture.useTrigger = true + fixture.listRuns.mockResolvedValue({ + data: [{ id: 'run-delayed', status: 'DELAYED' }], + hasNextPage: () => false, + }) + expect(await runCheckAt(check.id, overdue(retryAt))).toBe('pending') + const [event] = await retryChecksFor(file.documentId) + expect(event.attempts).toBe(0) + const [row] = await db.select().from(document).where(eq(document.id, file.documentId)) + expect(row).toMatchObject({ processingStatus: 'pending', processingDeferredUntil: retryAt }) + }) + + it.each([ + ['claims', { processingStatus: 'processing', processingStartedAt: new Date() }], + ['claims and re-defers', { processingDeferredUntil: new Date(Date.now() + 600_000) }], + ] as const)( + 'never overwrites a retry that %s the document while the check inspects it', + async (_label, change) => { + const { file, retryAt, check } = await deferredUpload(new Date()) + fixture.useTrigger = true + fixture.listRuns.mockImplementation(async () => { + await db.update(document).set(change).where(eq(document.id, file.documentId)) + return { data: [], hasNextPage: () => false } + }) + expect(await runCheckAt(check.id, overdue(retryAt))).toBe('completed') + const [row] = await db.select().from(document).where(eq(document.id, file.documentId)) + expect(row).toMatchObject(change) + expect(row.processingStatus).not.toBe('failed') + } + ) + + it('schedules no check for a connector document, which the recovery sweep covers', async () => { + const ids = await seed() + const file = await failedFile(ids) + await db + .update(document) + .set({ processingStatus: 'pending', processingQueueToken: GENERATION }) + .where(eq(document.id, file.documentId)) + const spy = failNextRunAfterClaim( + Object.assign(new Error('canceling statement due to lock timeout'), { code: '55P03' }) + ) + try { + await expect( + processDocumentAsync( + ids.knowledgeBaseId, + file.documentId, + file, + {}, + await resolveSystemBillingAttribution(ids.workspaceId), + GENERATION, + { + processingQueueToken: GENERATION, + chargedAtDispatch: false, + scheduleDatabaseRetry: () => new Date(Date.now() + 120_000), + } + ) + ).rejects.toMatchObject({ code: '55P03' }) + } finally { + spy.mockRestore() + } + const [row] = await db.select().from(document).where(eq(document.id, file.documentId)) + expect(row.processingStatus).toBe('pending') + expect(await retryChecksFor(file.documentId)).toHaveLength(0) + }) + + it('keeps a dispatch from claiming a deferred run that was never stamped, and the retry still claims it', async () => { + const { ids, file, billing, retryAt, runRetry } = await deferredUpload(null) + const [deferred] = await db.select().from(document).where(eq(document.id, file.documentId)) + expect(deferred.processingQueuedAt).toEqual(retryAt) + + await processDocumentsWithQueue( + [ + { + documentId: file.documentId, + filename: deferred.filename, + fileUrl: deferred.fileUrl, + fileSize: deferred.fileSize, + mimeType: deferred.mimeType, + }, + ], + ids.knowledgeBaseId, + {}, + generateId(), + billing, + 'interactive' + ) + const [afterDispatch] = await db.select().from(document).where(eq(document.id, file.documentId)) + expect(afterDispatch).toMatchObject({ + processingStatus: 'pending', + processingQueueToken: GENERATION, + processingAttempts: deferred.processingAttempts, + processingDeferredUntil: retryAt, + }) + + const onClaimed = vi.fn() + const spy = failNextRunAfterClaim(new Error('Synthetic failure after the retry claimed')) + try { + await runRetry({ onClaimed }).catch(() => undefined) + } finally { + spy.mockRestore() + } + expect(onClaimed).toHaveBeenCalledTimes(1) + }) +}) diff --git a/apps/sim/lib/knowledge/connectors/member-sync-engine.test.ts b/apps/sim/lib/knowledge/connectors/member-sync-engine.test.ts index 9341083baba..d6d42a6d8d7 100644 --- a/apps/sim/lib/knowledge/connectors/member-sync-engine.test.ts +++ b/apps/sim/lib/knowledge/connectors/member-sync-engine.test.ts @@ -1,7 +1,9 @@ /** * @vitest-environment node */ -import { describe, expect, it, vi } from 'vitest' +import { queueTableRows, resetDbChainMock, schemaMock } from '@sim/testing' +import { DrizzleQueryError } from 'drizzle-orm/errors' +import { beforeEach, describe, expect, it, vi } from 'vitest' vi.mock('@/connectors/registry.server', () => ({ CONNECTOR_REGISTRY: {} })) vi.mock('@/lib/knowledge/documents/service', () => ({ @@ -20,11 +22,15 @@ vi.mock('@/lib/billing/core/workspace-access', () => ({ vi.mock('@/lib/credential-groups/availability', () => ({ isCredentialGroupsAvailable: vi.fn() })) import { + buildMemberSyncDatabaseRetryUpdate, buildMemberSyncFailureUpdate, deriveMemberActive, + type MemberSyncResult, memberFailureBackoffMs, memberNextAttemptAt, + memberRunMadeProgress, nextMemberSyncTime, + resolveMemberSyncFailureUpdate, shouldListFully, } from '@/lib/knowledge/connectors/member-sync-engine' import { @@ -254,6 +260,131 @@ describe('member sync engine decisions', () => { }) }) + describe('buildMemberSyncDatabaseRetryUpdate', () => { + const now = new Date('2026-09-01T12:00:00Z') + const minutesAfter = (mins: number) => now.getTime() + mins * 60 * 1000 + + it('keeps the error visible without advancing the breaker', () => { + const update = buildMemberSyncDatabaseRetryUpdate( + now, + MAX_CONSECUTIVE_FAILURES - 1, + 'db timeout', + 40 + ) + expect(update).toMatchObject({ + memberSyncStatus: 'error', + lastMemberSyncError: 'db timeout', + memberSyncConsecutiveFailures: MAX_CONSECUTIVE_FAILURES - 1, + memberSyncLockToken: null, + memberSyncLockLeaseAt: null, + }) + }) + + it('schedules the next run after the resolved retry delay', () => { + expect( + buildMemberSyncDatabaseRetryUpdate(now, 0, 'db timeout', 120 * 60 * 1000).nextMemberSyncAt + ).toEqual(new Date(minutesAfter(120))) + }) + }) + + describe('memberRunMadeProgress', () => { + const idle = { + membersCompleted: 0, + docsAdded: 0, + docsUpdated: 0, + docsDeleted: 0, + } as MemberSyncResult + + it('reports no progress for a run that wrote nothing', () => { + expect(memberRunMadeProgress(idle)).toBe(false) + }) + + it.each([ + ['completed a member', { membersCompleted: 1 }], + ['added documents', { docsAdded: 2 }], + ['updated documents', { docsUpdated: 1 }], + ['purged documents in the lifecycle pass', { docsDeleted: 3 }], + ])('reports progress for a run that %s', (_label, writes) => { + expect(memberRunMadeProgress({ ...idle, ...writes })).toBe(true) + }) + }) + + describe('resolveMemberSyncFailureUpdate', () => { + beforeEach(() => { + resetDbChainMock() + }) + + const failure = { + connectorId: 'c-1', + runId: 'run-1', + previousFailures: MAX_CONSECUTIVE_FAILURES - 1, + errorMessage: 'failed', + madeProgress: false, + } + const run = (status: string, membersCompleted = 0) => ({ + status, + membersCompleted, + docsAdded: 0, + docsUpdated: 0, + }) + const deadlock = () => + new DrizzleQueryError( + 'update private SQL', + ['private'], + Object.assign(new Error('deadlock detected'), { code: '40P01' }) + ) + + it('does not disable a connector one failure from the breaker over a database timeout', async () => { + queueTableRows(schemaMock.knowledgeConnectorMemberSyncLog, [run('failed'), run('completed')]) + const timeout = new DrizzleQueryError( + 'select private SQL', + ['private'], + Object.assign(new Error('canceling statement due to statement timeout'), { + code: '57014', + }) + ) + const update = await resolveMemberSyncFailureUpdate(timeout, failure) + expect(update).toMatchObject({ + memberSyncStatus: 'error', + memberSyncConsecutiveFailures: MAX_CONSECUTIVE_FAILURES - 1, + }) + expect(update.nextMemberSyncAt).not.toBeNull() + }) + + it('reads the members-mode run log for the streak', async () => { + queueTableRows(schemaMock.knowledgeConnectorMemberSyncLog, [ + run('failed'), + run('failed'), + run('completed'), + ]) + const before = Date.now() + const update = await resolveMemberSyncFailureUpdate(deadlock(), { + ...failure, + previousFailures: 0, + }) + expect(update.nextMemberSyncAt!.getTime() - before).toBeGreaterThanOrEqual(90 * 60 * 1000) + }) + + it('retries within minutes after a run that completed members before the database failed', async () => { + queueTableRows(schemaMock.knowledgeConnectorMemberSyncLog, [run('failed'), run('failed')]) + const before = Date.now() + const update = await resolveMemberSyncFailureUpdate(deadlock(), { + ...failure, + madeProgress: true, + }) + expect(update.nextMemberSyncAt!.getTime() - before).toBeLessThanOrEqual(5 * 60 * 1000) + expect(update.memberSyncConsecutiveFailures).toBe(MAX_CONSECUTIVE_FAILURES - 1) + }) + + it('still disables at the breaker for a failure the database did not cause', async () => { + const update = await resolveMemberSyncFailureUpdate(new Error('source broke'), failure) + expect(update).toMatchObject({ + memberSyncStatus: 'disabled', + memberSyncConsecutiveFailures: MAX_CONSECUTIVE_FAILURES, + }) + }) + }) + describe('nextMemberSyncTime', () => { const now = new Date('2026-09-01T12:00:00Z') diff --git a/apps/sim/lib/knowledge/connectors/member-sync-engine.ts b/apps/sim/lib/knowledge/connectors/member-sync-engine.ts index 53c9de10a09..f031c25cfc8 100644 --- a/apps/sim/lib/knowledge/connectors/member-sync-engine.ts +++ b/apps/sim/lib/knowledge/connectors/member-sync-engine.ts @@ -11,7 +11,7 @@ import { knowledgeDocumentObservation, } from '@sim/db/schema' import { createLogger } from '@sim/logger' -import { getErrorMessage, toError } from '@sim/utils/errors' +import { getErrorMessage, getTransientDatabaseFailure, toError } from '@sim/utils/errors' import { generateId } from '@sim/utils/id' import { randomInt } from '@sim/utils/random' import { and, asc, eq, gt, inArray, isNull, lte, notExists, sql } from 'drizzle-orm' @@ -65,6 +65,7 @@ import { } from '@/lib/knowledge/connectors/member-observations' import { inviteWorkspaceMembersToCredentialGroup } from '@/lib/knowledge/connectors/member-provisioning' import { runConnectorContentPass } from '@/lib/knowledge/connectors/sync-content-pass' +import { resolveDatabaseRetryDelayMs } from '@/lib/knowledge/connectors/sync-database-retry' import { deferConnectorSync, getConnectorSyncDeferral, @@ -327,6 +328,87 @@ export function buildMemberSyncFailureUpdate( } } +/** + * The connector row written after the database, not the source, failed a members-mode run. The + * members-mode counterpart of `buildSyncDatabaseRetryUpdate`: the breaker keeps only the source + * failures already counted, and the retry waits the delay `resolveDatabaseRetryDelayMs` chose. + */ +export function buildMemberSyncDatabaseRetryUpdate( + now: Date, + previousFailures: number | null | undefined, + errorMessage: string, + retryDelayMs: number +) { + return { + memberSyncStatus: 'error' as const, + lastMemberSyncError: errorMessage, + nextMemberSyncAt: new Date(now.getTime() + retryDelayMs), + memberSyncConsecutiveFailures: previousFailures ?? 0, + memberSyncLockToken: null, + memberSyncLockLeaseAt: null, + updatedAt: now, + } +} + +/** + * Whether a members-mode run moved the sync forward: it completed a member, or wrote documents. + * `docsDeleted` holds the document lifecycle's purges, which the run log records as `docs_purged`. + */ +export function memberRunMadeProgress(result: MemberSyncResult): boolean { + return result.membersCompleted + result.docsAdded + result.docsUpdated + result.docsDeleted > 0 +} + +/** + * The connector row a failed members-mode run writes. A deterministic capacity rejection waits for + * an operator, a transient database failure retries without touching the breaker, and anything + * else climbs the ladder toward auto-disable. + */ +export async function resolveMemberSyncFailureUpdate( + error: unknown, + failure: { + connectorId: string + runId: string + previousFailures: number + errorMessage: string + retryAfterMs?: number + /** Whether the run completed a member or wrote documents before it failed. */ + madeProgress: boolean + } +) { + const now = new Date() + if (error instanceof ConnectorSyncCapacityError) { + return { + memberSyncStatus: 'error' as const, + lastMemberSyncError: failure.errorMessage, + nextMemberSyncAt: null, + memberSyncConsecutiveFailures: failure.previousFailures, + memberSyncLockToken: null, + memberSyncLockLeaseAt: null, + updatedAt: now, + } + } + if (getTransientDatabaseFailure(error)) { + return buildMemberSyncDatabaseRetryUpdate( + now, + failure.previousFailures, + failure.errorMessage, + await resolveDatabaseRetryDelayMs({ + kind: 'member', + connectorId: failure.connectorId, + runId: failure.runId, + previousFailures: failure.previousFailures, + madeProgress: failure.madeProgress, + }) + ) + } + return buildMemberSyncFailureUpdate( + now, + failure.previousFailures, + failure.errorMessage, + failure.retryAfterMs + ) +} + /** * When a member who completed is next due: exactly one interval on, with no * jitter, so they are due whenever the connector's own (jittered) run lands. @@ -2422,23 +2504,14 @@ export async function executeMemberSync( logger.error('Member sync failed', { connectorId, runId, error: errorMessage, diagnostic }) try { await failMemberSyncLog(runId, result, errorMessage) - const failureUpdate = - error instanceof ConnectorSyncCapacityError - ? { - memberSyncStatus: 'error' as const, - lastMemberSyncError: errorMessage, - nextMemberSyncAt: null, - memberSyncConsecutiveFailures: connector.memberSyncConsecutiveFailures, - memberSyncLockToken: null, - memberSyncLockLeaseAt: null, - updatedAt: new Date(), - } - : buildMemberSyncFailureUpdate( - new Date(), - connector.memberSyncConsecutiveFailures, - errorMessage, - retryAfterMs - ) + const failureUpdate = await resolveMemberSyncFailureUpdate(error, { + connectorId, + runId, + previousFailures: connector.memberSyncConsecutiveFailures, + errorMessage, + retryAfterMs, + madeProgress: memberRunMadeProgress(result), + }) const written = await db .update(knowledgeConnector) .set(failureUpdate) diff --git a/apps/sim/lib/knowledge/connectors/sync-database-retry.test.ts b/apps/sim/lib/knowledge/connectors/sync-database-retry.test.ts new file mode 100644 index 00000000000..b3de5f7ecbb --- /dev/null +++ b/apps/sim/lib/knowledge/connectors/sync-database-retry.test.ts @@ -0,0 +1,247 @@ +/** + * @vitest-environment node + */ +import { + dbChainMockFns, + hasMockCondition, + queueTableRows, + resetDbChainMock, + schemaMock, +} from '@sim/testing' +import { beforeEach, describe, expect, it, vi } from 'vitest' + +const { mockLogError } = vi.hoisted(() => ({ mockLogError: vi.fn() })) +vi.mock('@sim/logger', async () => { + const { createMockLogger } = await import('@sim/testing/mocks/logger.mock') + return { createLogger: () => ({ ...createMockLogger(), error: mockLogError }) } +}) + +import { + countZeroProgressFailedRuns, + DATABASE_FAILURE_ALERT_STREAK, + DATABASE_RETRY_AFTER_PROGRESS_MS, + databaseRetryDelayMs, + RUN_HISTORY_LOCK_TIMEOUT_MS, + RUN_HISTORY_STATEMENT_TIMEOUT_MS, + resolveDatabaseRetryDelayMs, +} from '@/lib/knowledge/connectors/sync-database-retry' +import { + CONNECTOR_FAILURE_BACKOFF_CAP_MINUTES, + MAX_CONSECUTIVE_FAILURES, +} from '@/lib/knowledge/connectors/sync-limits' + +const MINUTE = 60 * 1000 +const NO_WRITES = { docsAdded: 0, docsUpdated: 0, docsDeleted: 0 } +const contentRun = (status: string, writes: Partial = {}) => ({ + status, + ...NO_WRITES, + ...writes, +}) +const memberRun = (status: string, writes: Record = {}) => ({ + status, + membersCompleted: 0, + docsAdded: 0, + docsUpdated: 0, + docsPurged: 0, + ...writes, +}) + +describe('countZeroProgressFailedRuns', () => { + beforeEach(() => { + resetDbChainMock() + }) + + it('counts this run plus the failed runs before it, up to the last one that did not fail', async () => { + queueTableRows(schemaMock.knowledgeConnectorSyncLog, [ + contentRun('failed'), + contentRun('failed'), + contentRun('completed'), + contentRun('failed'), + ]) + expect(await countZeroProgressFailedRuns('content', 'c-1', 'run-1')).toBe(3) + }) + + it.each([ + ['added', { docsAdded: 5 }], + ['updated', { docsUpdated: 1 }], + ['deleted', { docsDeleted: 2 }], + ])('ends the streak at a failed run that %s documents', async (_label, writes) => { + queueTableRows(schemaMock.knowledgeConnectorSyncLog, [ + contentRun('failed'), + contentRun('failed', writes), + contentRun('failed'), + ]) + expect(await countZeroProgressFailedRuns('content', 'c-1', 'run-1')).toBe(2) + }) + + it('counts only this run after a success', async () => { + queueTableRows(schemaMock.knowledgeConnectorSyncLog, [ + contentRun('completed'), + contentRun('failed'), + ]) + expect(await countZeroProgressFailedRuns('content', 'c-1', 'run-1')).toBe(1) + }) + + it('counts an unbroken history in full', async () => { + queueTableRows(schemaMock.knowledgeConnectorSyncLog, [ + contentRun('failed'), + contentRun('failed'), + ]) + expect(await countZeroProgressFailedRuns('content', 'c-1', 'run-1')).toBe(3) + }) + + it('reads the members-mode run log for a members-mode run', async () => { + queueTableRows(schemaMock.knowledgeConnectorMemberSyncLog, [ + memberRun('failed'), + memberRun('started'), + ]) + expect(await countZeroProgressFailedRuns('member', 'c-1', 'run-1')).toBe(2) + }) + + it.each([ + ['completed a member', { membersCompleted: 1 }], + ['added documents', { docsAdded: 3 }], + ['updated documents', { docsUpdated: 1 }], + ['purged documents', { docsPurged: 4 }], + ])('ends a members-mode streak at a failed run that %s', async (_label, writes) => { + queueTableRows(schemaMock.knowledgeConnectorMemberSyncLog, [ + memberRun('failed'), + memberRun('failed', writes), + memberRun('failed'), + ]) + expect(await countZeroProgressFailedRuns('member', 'c-1', 'run-1')).toBe(2) + }) + + it('excludes the current run and reads only as far back as the ladder climbs', async () => { + queueTableRows(schemaMock.knowledgeConnectorSyncLog, []) + await countZeroProgressFailedRuns('content', 'c-1', 'run-1') + const where = dbChainMockFns.where.mock.calls.at(-1)?.[0] + expect( + hasMockCondition( + where, + (node) => + node.type === 'ne' && + node.left === schemaMock.knowledgeConnectorSyncLog.id && + node.right === 'run-1' + ) + ).toBe(true) + expect(dbChainMockFns.limit).toHaveBeenCalledWith( + CONNECTOR_FAILURE_BACKOFF_CAP_MINUTES / 30 - 1 + ) + }) + + it('bounds the history read with its own statement and lock timeouts', async () => { + queueTableRows(schemaMock.knowledgeConnectorSyncLog, []) + await countZeroProgressFailedRuns('content', 'c-1', 'run-1') + const bound = dbChainMockFns.execute.mock.calls[0]?.[0] as { + toSQL: () => { sql: string; params: unknown[] } + } + const { sql, params } = bound.toSQL() + expect(sql).toContain("set_config('statement_timeout'") + expect(sql).toContain("set_config('lock_timeout'") + expect(params).toEqual([ + String(RUN_HISTORY_STATEMENT_TIMEOUT_MS), + String(RUN_HISTORY_LOCK_TIMEOUT_MS), + ]) + expect(dbChainMockFns.execute.mock.invocationCallOrder[0]).toBeLessThan( + dbChainMockFns.select.mock.invocationCallOrder[0] + ) + }) + + it('falls back to this run alone when the history read times out', async () => { + queueTableRows(schemaMock.knowledgeConnectorSyncLog, [ + contentRun('failed'), + contentRun('failed'), + ]) + dbChainMockFns.limit.mockRejectedValueOnce( + Object.assign(new Error('canceling statement due to statement timeout'), { code: '57014' }) + ) + expect(await countZeroProgressFailedRuns('content', 'c-1', 'run-1')).toBe(1) + }) + + it('falls back to this run alone when the history cannot be read', async () => { + dbChainMockFns.limit.mockRejectedValueOnce(new Error('canceling statement')) + expect(await countZeroProgressFailedRuns('content', 'c-1', 'run-1')).toBe(1) + }) +}) + +describe('databaseRetryDelayMs', () => { + it.each([ + [1, 30], + [2, 60], + [3, 90], + [10, 300], + ])('climbs the failure ladder with the streak (%i failed runs → %i min)', (streak, minutes) => { + const delay = databaseRetryDelayMs(streak, 0) + expect(delay).toBeGreaterThanOrEqual(minutes * MINUTE) + expect(delay).toBeLessThanOrEqual(minutes * MINUTE + MINUTE) + }) + + it('stops at the ladder ceiling', () => { + expect(databaseRetryDelayMs(1_000, 0)).toBeLessThanOrEqual( + CONNECTOR_FAILURE_BACKOFF_CAP_MINUTES * MINUTE + MINUTE + ) + }) + + it('never waits less than the rung the breaker count already earned', () => { + const delay = databaseRetryDelayMs(1, MAX_CONSECUTIVE_FAILURES - 1) + expect(delay).toBeGreaterThanOrEqual(MAX_CONSECUTIVE_FAILURES * 30 * MINUTE) + }) +}) + +describe('resolveDatabaseRetryDelayMs', () => { + beforeEach(() => { + vi.clearAllMocks() + resetDbChainMock() + }) + + const retry = { + kind: 'content' as const, + connectorId: 'c-1', + runId: 'run-1', + previousFailures: 0, + } + + it('retries shortly after a run that made progress, without reading the streak', async () => { + queueTableRows( + schemaMock.knowledgeConnectorSyncLog, + Array.from({ length: 20 }, () => contentRun('failed')) + ) + const delay = await resolveDatabaseRetryDelayMs({ ...retry, madeProgress: true }) + expect(delay).toBeGreaterThanOrEqual(DATABASE_RETRY_AFTER_PROGRESS_MS) + expect(delay).toBeLessThanOrEqual(DATABASE_RETRY_AFTER_PROGRESS_MS + MINUTE) + expect(dbChainMockFns.select).not.toHaveBeenCalled() + }) + + it('climbs the ladder by the zero-progress streak for a run that made none', async () => { + queueTableRows(schemaMock.knowledgeConnectorSyncLog, [ + contentRun('failed'), + contentRun('failed'), + contentRun('completed'), + ]) + const delay = await resolveDatabaseRetryDelayMs({ ...retry, madeProgress: false }) + expect(delay).toBeGreaterThanOrEqual(90 * MINUTE) + expect(delay).toBeLessThanOrEqual(91 * MINUTE) + }) + + it('reports a streak that reaches the alert threshold at error level, without disabling', async () => { + queueTableRows( + schemaMock.knowledgeConnectorSyncLog, + Array.from({ length: DATABASE_FAILURE_ALERT_STREAK - 1 }, () => contentRun('failed')) + ) + await resolveDatabaseRetryDelayMs({ ...retry, madeProgress: false }) + expect(mockLogError).toHaveBeenCalledWith( + 'Connector sync keeps failing on the database without progress', + { connectorId: 'c-1', kind: 'content', zeroProgressFailedRuns: DATABASE_FAILURE_ALERT_STREAK } + ) + }) + + it('stays quiet below the alert threshold', async () => { + queueTableRows( + schemaMock.knowledgeConnectorSyncLog, + Array.from({ length: DATABASE_FAILURE_ALERT_STREAK - 2 }, () => contentRun('failed')) + ) + await resolveDatabaseRetryDelayMs({ ...retry, madeProgress: false }) + expect(mockLogError).not.toHaveBeenCalled() + }) +}) diff --git a/apps/sim/lib/knowledge/connectors/sync-database-retry.ts b/apps/sim/lib/knowledge/connectors/sync-database-retry.ts new file mode 100644 index 00000000000..aeea4535010 --- /dev/null +++ b/apps/sim/lib/knowledge/connectors/sync-database-retry.ts @@ -0,0 +1,177 @@ +import { db } from '@sim/db' +import { knowledgeConnectorMemberSyncLog, knowledgeConnectorSyncLog } from '@sim/db/schema' +import { createLogger } from '@sim/logger' +import { describeError } from '@sim/utils/errors' +import { randomInt } from '@sim/utils/random' +import { and, desc, eq, ne, sql } from 'drizzle-orm' +import type { DbTransaction } from '@/lib/db/types' +import { + CONNECTOR_FAILURE_BACKOFF_CAP_MINUTES, + CONNECTOR_FAILURE_BACKOFF_STEP_MINUTES, + connectorFailureBackoffMinutes, +} from '@/lib/knowledge/connectors/sync-limits' + +const logger = createLogger('ConnectorDatabaseRetry') + +/** Jitter added to a database retry, so connectors failed by one slow window do not return together. */ +const DATABASE_RETRY_JITTER_MAX_MS = 60_000 + +/** + * The retry after a run that failed on the database but moved the sync forward first. Its + * checkpoint is durable, so the next run resumes rather than repeats, and waiting a ladder rung + * per failure would crawl a large first sync that hits one slow statement per run. + */ +export const DATABASE_RETRY_AFTER_PROGRESS_MS = 3 * 60 * 1000 + +/** + * Failed runs in a row, none of them making progress, at which the retries are reported for + * alerting. The connector is never disabled for them; this is the signal that it is stuck. + */ +export const DATABASE_FAILURE_ALERT_STREAK = 10 + +/** The streak at which the failure ladder reaches its ceiling; longer streaks back off no further. */ +const LADDER_RUNGS = Math.ceil( + CONNECTOR_FAILURE_BACKOFF_CAP_MINUTES / CONNECTOR_FAILURE_BACKOFF_STEP_MINUTES +) + +/** + * Limits on the run-history read. It runs on the failure path, usually while the database is the + * thing failing, so it must give up fast rather than queue behind the slow window; a read that + * times out falls back to counting this run alone. + */ +export const RUN_HISTORY_STATEMENT_TIMEOUT_MS = 2_000 +export const RUN_HISTORY_LOCK_TIMEOUT_MS = 500 + +/** Which run log a sync writes: the content sync's, or the members-mode run's. */ +export type SyncRunLogKind = 'content' | 'member' + +interface LoggedRun { + status: string + progressed: boolean +} + +/** Earlier runs of this connector, newest first, and whether each one moved the sync forward. */ +async function readEarlierRuns( + tx: DbTransaction, + kind: SyncRunLogKind, + connectorId: string, + runId: string, + limit: number +): Promise { + if (kind === 'content') { + const log = knowledgeConnectorSyncLog + const rows = await tx + .select({ + status: log.status, + docsAdded: log.docsAdded, + docsUpdated: log.docsUpdated, + docsDeleted: log.docsDeleted, + }) + .from(log) + .where(and(eq(log.connectorId, connectorId), ne(log.id, runId))) + .orderBy(desc(log.startedAt)) + .limit(limit) + return rows.map((row) => ({ + status: row.status, + progressed: row.docsAdded + row.docsUpdated + row.docsDeleted > 0, + })) + } + const log = knowledgeConnectorMemberSyncLog + const rows = await tx + .select({ + status: log.status, + membersCompleted: log.membersCompleted, + docsAdded: log.docsAdded, + docsUpdated: log.docsUpdated, + docsPurged: log.docsPurged, + }) + .from(log) + .where(and(eq(log.connectorId, connectorId), ne(log.id, runId))) + .orderBy(desc(log.startedAt)) + .limit(limit) + return rows.map((row) => ({ + status: row.status, + progressed: row.membersCompleted + row.docsAdded + row.docsUpdated + row.docsPurged > 0, + })) +} + +/** + * How many runs in a row, ending with `runId`, have failed without moving the sync forward: the + * run itself plus every such run before it, back to the last run that succeeded or made progress, + * bounded by the ladder's ceiling. + * + * A database failure never advances the connector's failure counter, so that a slow database + * cannot spend the breaker that disables connectors for persistent source failures. The run log + * already records every attempt and what it wrote, so it measures the streak instead: a statement + * that fails every run without progress still backs off rung by rung, while a run that added, + * updated, or deleted documents (in members mode, purged by the document lifecycle and logged as + * `docs_purged`), or completed a member, before failing ends it. + * The read uses the log's `(connector_id, started_at DESC)` index and is bounded by + * {@link RUN_HISTORY_STATEMENT_TIMEOUT_MS}. If it fails or times out, the streak counts only this + * run and the caller's own floor applies. + */ +export async function countZeroProgressFailedRuns( + kind: SyncRunLogKind, + connectorId: string, + runId: string +): Promise { + try { + const earlier = await db.transaction(async (tx) => { + await tx.execute( + sql`SELECT set_config('statement_timeout', ${String(RUN_HISTORY_STATEMENT_TIMEOUT_MS)}, true), set_config('lock_timeout', ${String(RUN_HISTORY_LOCK_TIMEOUT_MS)}, true)` + ) + return readEarlierRuns(tx, kind, connectorId, runId, LADDER_RUNGS - 1) + }) + const streakEnd = earlier.findIndex((run) => run.status !== 'failed' || run.progressed) + return 1 + (streakEnd === -1 ? earlier.length : streakEnd) + } catch (error) { + logger.warn('Could not read the failed-run streak; backing off from this run alone', { + connectorId, + kind, + error: describeError(error), + }) + return 1 + } +} + +/** + * The delay before retrying a run the database failed with no progress: the failure ladder's + * rung for the longer of the zero-progress streak and the breaker's own count, plus jitter. + */ +export function databaseRetryDelayMs( + failedRunStreak: number, + previousFailures: number | null | undefined +): number { + const rung = Math.max(failedRunStreak, (previousFailures ?? 0) + 1) + return ( + connectorFailureBackoffMinutes(rung) * 60 * 1000 + + randomInt(0, DATABASE_RETRY_JITTER_MAX_MS + 1) + ) +} + +/** + * When to retry a run that failed on the database. A run that made progress retries after + * {@link DATABASE_RETRY_AFTER_PROGRESS_MS}; one that made none climbs the ladder by its + * zero-progress streak and, from {@link DATABASE_FAILURE_ALERT_STREAK} on, reports the streak at + * error level so a connector stuck on the database is visible without being disabled. + */ +export async function resolveDatabaseRetryDelayMs(retry: { + kind: SyncRunLogKind + connectorId: string + runId: string + previousFailures: number | null | undefined + madeProgress: boolean +}): Promise { + if (retry.madeProgress) { + return DATABASE_RETRY_AFTER_PROGRESS_MS + randomInt(0, DATABASE_RETRY_JITTER_MAX_MS + 1) + } + const streak = await countZeroProgressFailedRuns(retry.kind, retry.connectorId, retry.runId) + if (streak >= DATABASE_FAILURE_ALERT_STREAK) { + logger.error('Connector sync keeps failing on the database without progress', { + connectorId: retry.connectorId, + kind: retry.kind, + zeroProgressFailedRuns: streak, + }) + } + return databaseRetryDelayMs(streak, retry.previousFailures) +} diff --git a/apps/sim/lib/knowledge/connectors/sync-engine.test.ts b/apps/sim/lib/knowledge/connectors/sync-engine.test.ts index 6b7e35e018f..20968907f68 100644 --- a/apps/sim/lib/knowledge/connectors/sync-engine.test.ts +++ b/apps/sim/lib/knowledge/connectors/sync-engine.test.ts @@ -17,8 +17,16 @@ import { generateShortId } from '@sim/utils/id' import { DrizzleQueryError } from 'drizzle-orm/errors' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import * as connectorTokens from '@/lib/knowledge/connectors/access-token' -import { executeSync, isConnectorRunnableStatus } from '@/lib/knowledge/connectors/sync-engine' -import { CREDENTIAL_REVOKED_SYNC_ERROR } from '@/lib/knowledge/connectors/sync-limits' +import { + buildSyncDatabaseRetryUpdate, + buildSyncFailureUpdate, + executeSync, + isConnectorRunnableStatus, +} from '@/lib/knowledge/connectors/sync-engine' +import { + CREDENTIAL_REVOKED_SYNC_ERROR, + MAX_CONSECUTIVE_FAILURES, +} from '@/lib/knowledge/connectors/sync-limits' import { classifySuspectListing, evaluateListingSafety, @@ -1102,6 +1110,148 @@ describe('executeSync deferred hydration rate limits', () => { }) }) +describe('executeSync database failures', () => { + const NOW = new Date('2026-08-29T03:00:00.000Z') + + async function failSyncWith( + error: Error, + consecutiveFailures?: number, + firstPage?: { documents: ExternalDocument[] } + ) { + const connector = { + id: 'c-1', + knowledgeBaseId: 'kb-1', + connectorType: 'paged', + credentialId: null, + encryptedApiKey: null, + sourceConfig: {}, + syncMode: 'full', + syncIntervalMinutes: 1440, + accessMode: 'workspace', + status: 'active', + lastSyncAt: null, + lastSyncDocCount: null, + consecutiveFailures: consecutiveFailures ?? MAX_CONSECUTIVE_FAILURES - 1, + syncLockToken: null, + } + queueTableRows(schemaMock.knowledgeConnector, [connector]) + for (let i = 0; i < 20; i++) + queueTableRows(schemaMock.knowledgeConnector, [ + { id: 'c-1', connectorArchivedAt: null, connectorDeletedAt: null, kbDeletedAt: null }, + ]) + queueTableRows(schemaMock.knowledgeBase, [{ userId: 'u-1', workspaceId: 'ws-1' }]) + for (let i = 0; i < 5; i++) queueTableRows(schemaMock.knowledgeBase, [{ id: 'kb-1' }]) + for (let i = 0; i < 4; i++) queueTableRows(schemaMock.document, []) + dbChainMockFns.returning.mockResolvedValue([{ id: 'c-1' }]).mockResolvedValueOnce([connector]) + if (firstPage) { + mockUploadFile.mockImplementation(async ({ customKey }: { customKey: string }) => ({ + key: customKey, + path: `/api/files/serve/${encodeURIComponent(customKey)}`, + })) + mockProcessDocumentsWithQueue.mockImplementation(async (documents: unknown[]) => ({ + accepted: documents.length, + failed: 0, + })) + mockListDocuments.mockResolvedValueOnce({ + documents: firstPage.documents, + hasMore: true, + nextCursor: 'page-2', + }) + } + mockListDocuments.mockRejectedValueOnce(error) + + const result = await executeSync('c-1', { + billingAttribution: { workspaceId: 'ws-1' } as never, + }) + const terminal = dbChainMockFns.set.mock.calls + .map(([value]) => value as Record) + .find((value) => 'consecutiveFailures' in value) + return { result, terminal, MAX_CONSECUTIVE_FAILURES } + } + + beforeEach(() => { + vi.clearAllMocks() + resetDbChainMock() + vi.useFakeTimers() + vi.setSystemTime(NOW) + }) + + afterEach(() => { + vi.useRealTimers() + }) + + it('does not disable a connector one failure from the breaker over a database timeout', async () => { + const timeout = new DrizzleQueryError( + 'select private SQL', + ['private'], + Object.assign(new Error('canceling statement due to statement timeout'), { code: '57014' }) + ) + const { result, terminal, MAX_CONSECUTIVE_FAILURES } = await failSyncWith(timeout) + + expect(result.error).toBe('Database request failed (SQLSTATE 57014).') + expect(terminal).toMatchObject({ + status: 'error', + lastSyncError: 'Database request failed (SQLSTATE 57014).', + consecutiveFailures: MAX_CONSECUTIVE_FAILURES - 1, + }) + expect((terminal?.nextSyncAt as Date).getTime()).toBeGreaterThan(NOW.getTime()) + }) + + it('backs a repeated database failure off by the streak in the run log', async () => { + queueTableRows(schemaMock.knowledgeConnectorSyncLog, [ + { status: 'failed' }, + { status: 'failed' }, + { status: 'completed' }, + ]) + const timeout = new DrizzleQueryError( + 'select private SQL', + ['private'], + Object.assign(new Error('canceling statement due to lock timeout'), { code: '55P03' }) + ) + const { terminal } = await failSyncWith(timeout, 0) + + /** Two failed runs before this one: the third rung, with the breaker still at zero. */ + const delay = (terminal?.nextSyncAt as Date).getTime() - NOW.getTime() + expect(delay).toBeGreaterThanOrEqual(90 * 60 * 1000) + expect(delay).toBeLessThanOrEqual(91 * 60 * 1000) + expect(terminal).toMatchObject({ status: 'error', consecutiveFailures: 0 }) + }) + + it('retries within minutes after a run that added documents before the database failed', async () => { + const timeout = new DrizzleQueryError( + 'select private SQL', + ['private'], + Object.assign(new Error('canceling statement due to lock timeout'), { code: '55P03' }) + ) + const { result, terminal } = await failSyncWith(timeout, 0, { + documents: [ + { + externalId: 'external-1', + title: 'Document 1', + content: 'hydrated', + contentHash: 'hash-1', + mimeType: 'text/plain', + metadata: { size: 8 }, + }, + ], + }) + + expect(result.docsAdded).toBeGreaterThan(0) + const delay = (terminal?.nextSyncAt as Date).getTime() - NOW.getTime() + expect(delay).toBeLessThanOrEqual(5 * 60 * 1000) + expect(terminal).toMatchObject({ status: 'error', consecutiveFailures: 0 }) + }) + + it('still disables at the breaker for a failure the database did not cause', async () => { + const { terminal, MAX_CONSECUTIVE_FAILURES } = await failSyncWith(new Error('source broke')) + + expect(terminal).toMatchObject({ + status: 'disabled', + consecutiveFailures: MAX_CONSECUTIVE_FAILURES, + }) + }) +}) + describe('previous complete listing evidence', () => { beforeEach(() => { vi.clearAllMocks() @@ -2019,6 +2169,44 @@ describe('buildSyncCapacityUpdate', () => { }) }) +describe('buildSyncDatabaseRetryUpdate', () => { + const now = new Date('2026-08-20T00:00:00.000Z') + const minutesAfter = (mins: number) => now.getTime() + mins * 60 * 1000 + + it('keeps the error visible without advancing the auto-disable counter', async () => { + const update = buildSyncDatabaseRetryUpdate(now, MAX_CONSECUTIVE_FAILURES - 1, 'db timeout', 40) + expect(update).toMatchObject({ + status: 'error', + lastSyncError: 'db timeout', + consecutiveFailures: MAX_CONSECUTIVE_FAILURES - 1, + syncLockToken: null, + syncLockLeaseAt: null, + updatedAt: now, + }) + }) + + it('schedules the next run after the resolved retry delay', async () => { + expect(buildSyncDatabaseRetryUpdate(now, 0, 'db timeout', 90 * 60 * 1000).nextSyncAt).toEqual( + new Date(minutesAfter(90)) + ) + }) + + it('leaves a later source failure to be judged on source failures alone', async () => { + let failures = 1 + for (let run = 0; run < 30; run++) { + failures = buildSyncDatabaseRetryUpdate( + now, + failures, + 'db timeout', + 30 * 60 * 1000 + ).consecutiveFailures + } + const sourceFailure = buildSyncFailureUpdate(now, failures, 'source broke') + expect(sourceFailure.status).toBe('error') + expect(sourceFailure.consecutiveFailures).toBe(2) + }) +}) + describe('sync lock lease', () => { const now = new Date('2026-08-20T00:00:00.000Z') @@ -2985,8 +3173,9 @@ describe('executeSync heartbeats during the listing phase', () => { }) expect(mockListDocuments).not.toHaveBeenCalled() expect(JSON.stringify(mockLogError.mock.calls)).not.toContain('private') + /** A database timeout is not the connector's failure, so it leaves the breaker alone. */ expect(dbChainMockFns.set).toHaveBeenCalledWith( - expect.objectContaining({ status: 'error', consecutiveFailures: 1 }) + expect.objectContaining({ status: 'error', consecutiveFailures: 0 }) ) } ) diff --git a/apps/sim/lib/knowledge/connectors/sync-engine.ts b/apps/sim/lib/knowledge/connectors/sync-engine.ts index faa486faf66..d8545ff6b32 100644 --- a/apps/sim/lib/knowledge/connectors/sync-engine.ts +++ b/apps/sim/lib/knowledge/connectors/sync-engine.ts @@ -6,7 +6,7 @@ import { knowledgeConnectorSyncLog, } from '@sim/db/schema' import { createLogger } from '@sim/logger' -import { getErrorMessage, toError } from '@sim/utils/errors' +import { getErrorMessage, getTransientDatabaseFailure, toError } from '@sim/utils/errors' import { generateId } from '@sim/utils/id' import { randomInt } from '@sim/utils/random' import { and, asc, eq, exists, gt, inArray, isNotNull, isNull, or, sql } from 'drizzle-orm' @@ -48,6 +48,7 @@ import { unansweredByListing, } from '@/lib/knowledge/connectors/mirrored-acls' import { runConnectorContentPass } from '@/lib/knowledge/connectors/sync-content-pass' +import { resolveDatabaseRetryDelayMs } from '@/lib/knowledge/connectors/sync-database-retry' import { deferConnectorSync, getConnectorSyncDeferral, @@ -715,6 +716,34 @@ export function buildSyncCapacityUpdate( } } +/** + * The connector row written after the database, not the source, failed the run: a statement, + * lock, or transaction timeout, a deadlock, or a dropped connection. + * + * A slow database window says nothing about the connector, so, like throttling, it must not + * consume the breaker that disables connectors after persistent failures: the counter keeps the + * source failures already counted, and a later source failure is judged on those alone. The retry + * still backs off by the delay {@link resolveDatabaseRetryDelayMs} chose: short after a run that + * made progress, and otherwise up the failure ladder, so a statement too heavy for its budget backs + * off to the ladder's ceiling instead of re-crawling the source every half hour. + */ +export function buildSyncDatabaseRetryUpdate( + now: Date, + previousFailures: number | null | undefined, + errorMessage: string, + retryDelayMs: number +) { + return { + status: 'error' as const, + lastSyncError: errorMessage, + nextSyncAt: new Date(now.getTime() + retryDelayMs), + consecutiveFailures: previousFailures ?? 0, + syncLockToken: null, + syncLockLeaseAt: null, + updatedAt: now, + } +} + /** * The connector row a successful sync writes. * @@ -1536,22 +1565,37 @@ export async function executeSync( try { await completeSyncLog(syncLogId, 'failed', result, { errorMessage }) + const databaseFailure = + !(error instanceof ConnectorSyncCapacityError) && getTransientDatabaseFailure(error) const failureUpdate = error instanceof ConnectorSyncCapacityError ? buildSyncCapacityUpdate(new Date(), connector.consecutiveFailures, errorMessage) - : rateLimited - ? buildSyncRateLimitUpdate( - new Date(), - connector.consecutiveFailures, - errorMessage, - retryAfterMs - ) - : buildSyncFailureUpdate( + : databaseFailure + ? buildSyncDatabaseRetryUpdate( new Date(), connector.consecutiveFailures, errorMessage, - retryAfterMs + await resolveDatabaseRetryDelayMs({ + kind: 'content', + connectorId, + runId: syncLogId, + previousFailures: connector.consecutiveFailures, + madeProgress: result.docsAdded + result.docsUpdated + result.docsDeleted > 0, + }) ) + : rateLimited + ? buildSyncRateLimitUpdate( + new Date(), + connector.consecutiveFailures, + errorMessage, + retryAfterMs + ) + : buildSyncFailureUpdate( + new Date(), + connector.consecutiveFailures, + errorMessage, + retryAfterMs + ) if (failureUpdate.status === 'disabled') { logger.warn('Connector disabled after repeated failures', { diff --git a/apps/sim/lib/knowledge/documents/deferred-retry-check.test.ts b/apps/sim/lib/knowledge/documents/deferred-retry-check.test.ts new file mode 100644 index 00000000000..f82a4178ed9 --- /dev/null +++ b/apps/sim/lib/knowledge/documents/deferred-retry-check.test.ts @@ -0,0 +1,236 @@ +/** + * @vitest-environment node + */ +import { + dbChainMockFns, + hasMockCondition, + queueTableRows, + resetDbChainMock, + schemaMock, +} from '@sim/testing' +import { beforeEach, describe, expect, it, vi } from 'vitest' + +const { mockInspect, mockSnapshotCondition } = vi.hoisted(() => ({ + mockInspect: vi.fn(), + mockSnapshotCondition: vi.fn((snapshot: { id: string }) => ({ + type: 'snapshot', + id: snapshot.id, + })), +})) +vi.mock('@/lib/knowledge/documents/processing-recovery-queue', () => ({ + processingSnapshotColumns: {}, + documentProcessingSnapshotCondition: mockSnapshotCondition, + inspectDocumentProcessingLiveness: mockInspect, +})) + +import { + checkDeferredDocumentRetry, + DEFERRED_RETRY_CHECK_TERMINAL_MS, + DEFERRED_RETRY_LOST_ERROR, + DEFERRED_RETRY_RECHECK_MS, +} from '@/lib/knowledge/documents/deferred-retry-check' +import type { DeferredRetryCheckPayload } from '@/lib/knowledge/documents/processing-outbox-event' +import { QUEUED_DISPATCH_GRACE_MS, RECOVERY_WINDOW_MS } from '@/lib/knowledge/documents/types' + +const QUEUED_AT = new Date('2026-09-01T00:00:00.000Z') +const DEFERRED_UNTIL = new Date('2026-09-01T00:02:00.000Z') +const OVERDUE = DEFERRED_UNTIL.getTime() + QUEUED_DISPATCH_GRACE_MS + 60_000 + +const PAYLOAD: DeferredRetryCheckPayload = { + knowledgeBaseId: 'kb-1', + documentId: 'doc-1', + processingQueueToken: 'token-1', + processingQueuedAt: QUEUED_AT.toISOString(), + processingDeferredUntil: DEFERRED_UNTIL.toISOString(), +} + +const DEFERRED_ROW = { + id: 'doc-1', + uploadedAt: new Date('2026-08-31T00:00:00.000Z'), + processingStatus: 'pending', + processingQueueToken: 'token-1', + processingQueuedAt: QUEUED_AT, + processingStartedAt: null, + processingDeferredUntil: DEFERRED_UNTIL, + processingCompletedAt: null, + processingRecoveryAfter: null, + connectorId: null, + archivedAt: null, + deletedAt: null, +} + +const context = { + eventId: 'event-1', + eventType: 'knowledge.document.deferred-retry-check', + attempts: 0, + maxAttempts: 5, + signal: new AbortController().signal, + checkpointPayload: vi.fn(), +} + +function failedWrite() { + return dbChainMockFns.set.mock.calls.find( + ([value]) => (value as Record).processingStatus === 'failed' + )?.[0] +} + +async function check(row: Record | null, now = OVERDUE) { + vi.setSystemTime(now) + queueTableRows(schemaMock.document, row ? [row] : []) + return checkDeferredDocumentRetry(PAYLOAD, context) +} + +describe('checkDeferredDocumentRetry', () => { + beforeEach(() => { + vi.clearAllMocks() + resetDbChainMock() + vi.useFakeTimers({ toFake: ['Date'] }) + mockInspect.mockImplementation(async (rows: unknown[]) => ({ abandoned: rows, live: [] })) + }) + + it('fails a document whose scheduled retry never ran, fenced on the snapshot it inspected', async () => { + expect(await check(DEFERRED_ROW)).toBeUndefined() + + expect(mockInspect).toHaveBeenCalledWith([DEFERRED_ROW], context.signal) + expect(failedWrite()).toEqual({ + processingStatus: 'failed', + processingError: DEFERRED_RETRY_LOST_ERROR, + processingDeferredUntil: null, + processingCompletedAt: new Date(OVERDUE), + }) + expect(failedWrite()).not.toHaveProperty('processingQueueToken') + const where = dbChainMockFns.where.mock.calls.at(-1)?.[0] + expect(mockSnapshotCondition).toHaveBeenCalledWith(DEFERRED_ROW) + expect(hasMockCondition(where, (node) => node.type === 'snapshot')).toBe(true) + expect( + hasMockCondition( + where, + (node) => + node.type === 'eq' && + node.left === schemaMock.document.processingStatus && + node.right === 'pending' + ) + ).toBe(true) + expect( + hasMockCondition( + where, + (node) => node.type === 'isNull' && node.column === schemaMock.document.connectorId + ) || + hasMockCondition( + where, + (node) => node.type === 'isNull' && node.left === schemaMock.document.connectorId + ) + ).toBe(true) + }) + + it('checks again later without spending an attempt while the run may be live', async () => { + mockInspect.mockResolvedValue({ abandoned: [], live: [DEFERRED_ROW] }) + + expect(await check(DEFERRED_ROW)).toEqual({ + outcome: 'deferred', + reason: 'Deferred retry may still be running', + minimumBackoffMs: DEFERRED_RETRY_RECHECK_MS, + consumeAttempt: false, + }) + expect(failedWrite()).toBeUndefined() + }) + + it('waits for the grace before inspecting a check that ran early', async () => { + const early = DEFERRED_UNTIL.getTime() + 1_000 + expect(await check(DEFERRED_ROW, early)).toMatchObject({ + outcome: 'deferred', + minimumBackoffMs: QUEUED_DISPATCH_GRACE_MS - 1_000, + consumeAttempt: false, + }) + expect(mockInspect).not.toHaveBeenCalled() + expect(failedWrite()).toBeUndefined() + }) + + it('stops rechecking past the recovery window, so the event always ends', async () => { + mockInspect.mockResolvedValue({ abandoned: [], live: [DEFERRED_ROW] }) + + expect(await check(DEFERRED_ROW, DEFERRED_UNTIL.getTime() + RECOVERY_WINDOW_MS)).toBeUndefined() + expect(mockInspect).not.toHaveBeenCalled() + expect(failedWrite()).toMatchObject({ processingStatus: 'failed' }) + }) + + it.each([ + ['a replaced queue token', { processingQueueToken: 'token-2' }], + ['a new dispatch generation', { processingQueuedAt: new Date('2026-09-02T00:00:00.000Z') }], + ['a later deferral', { processingDeferredUntil: new Date('2026-09-01T01:00:00.000Z') }], + ['a claimed retry', { processingStatus: 'processing', processingDeferredUntil: null }], + ['a claim that kept the deferral stamp', { processingStatus: 'processing' }], + ['a completed pass', { processingStatus: 'completed', processingDeferredUntil: null }], + ['a failed document', { processingStatus: 'failed', processingDeferredUntil: null }], + ['a deleted document', { deletedAt: new Date() }], + ['an archived document', { archivedAt: new Date() }], + ['a connector document', { connectorId: 'connector-1' }], + ])('completes as a no-op after %s', async (_label, change) => { + expect(await check({ ...DEFERRED_ROW, ...change })).toBeUndefined() + expect(mockInspect).not.toHaveBeenCalled() + expect(failedWrite()).toBeUndefined() + }) + + it('completes as a no-op for a document that no longer exists', async () => { + expect(await check(null)).toBeUndefined() + expect(failedWrite()).toBeUndefined() + }) + + describe('database failures while checking', () => { + const lockTimeout = () => + Object.assign(new Error('Failed query: private SQL'), { + query: 'private SQL', + cause: Object.assign(new Error('canceling statement due to lock timeout'), { + code: '55P03', + }), + }) + + function expectDatabaseDeferral(result: unknown) { + expect(result).toMatchObject({ outcome: 'deferred', consumeAttempt: false }) + const backoff = (result as { minimumBackoffMs: number }).minimumBackoffMs + expect(backoff).toBeGreaterThan(0) + expect(backoff).toBeLessThanOrEqual(DEFERRED_RETRY_RECHECK_MS) + } + + it('postpones the check without spending an attempt when the read fails', async () => { + vi.setSystemTime(OVERDUE) + dbChainMockFns.limit.mockRejectedValueOnce(lockTimeout()) + expectDatabaseDeferral(await checkDeferredDocumentRetry(PAYLOAD, context)) + }) + + it('postpones the check without spending an attempt when the fenced write fails', async () => { + dbChainMockFns.returning.mockRejectedValueOnce(lockTimeout()) + expectDatabaseDeferral(await check(DEFERRED_ROW)) + }) + + it('backs off longer the longer the check has been overdue, up to the recheck interval', async () => { + vi.setSystemTime(DEFERRED_UNTIL.getTime() + QUEUED_DISPATCH_GRACE_MS + 6 * 60 * 60 * 1000) + dbChainMockFns.limit.mockRejectedValueOnce(lockTimeout()) + const late = (await checkDeferredDocumentRetry(PAYLOAD, context)) as { + minimumBackoffMs: number + } + expect(late.minimumBackoffMs).toBeGreaterThanOrEqual(DEFERRED_RETRY_RECHECK_MS * 0.8) + expect(late.minimumBackoffMs).toBeLessThanOrEqual(DEFERRED_RETRY_RECHECK_MS * 1.2) + }) + + it('spends an attempt on an error that is not a transient database failure', async () => { + vi.setSystemTime(OVERDUE) + const constraint = Object.assign(new Error('duplicate key'), { code: '23505' }) + dbChainMockFns.limit.mockRejectedValueOnce(constraint) + await expect(checkDeferredDocumentRetry(PAYLOAD, context)).rejects.toBe(constraint) + }) + + it('spends attempts on a database failure once past the terminal bound, so the event ends', async () => { + vi.setSystemTime(DEFERRED_UNTIL.getTime() + DEFERRED_RETRY_CHECK_TERMINAL_MS) + const error = lockTimeout() + dbChainMockFns.limit.mockRejectedValueOnce(error) + await expect(checkDeferredDocumentRetry(PAYLOAD, context)).rejects.toBe(error) + }) + }) + + it('rejects a payload without its document or deferral', async () => { + await expect(checkDeferredDocumentRetry({ knowledgeBaseId: 'kb-1' }, context)).rejects.toThrow( + 'missing' + ) + }) +}) diff --git a/apps/sim/lib/knowledge/documents/deferred-retry-check.ts b/apps/sim/lib/knowledge/documents/deferred-retry-check.ts new file mode 100644 index 00000000000..847ea4f5b21 --- /dev/null +++ b/apps/sim/lib/knowledge/documents/deferred-retry-check.ts @@ -0,0 +1,191 @@ +import { db } from '@sim/db' +import { document } from '@sim/db/schema' +import { createLogger } from '@sim/logger' +import { toStringOrNull } from '@sim/utils/coerce' +import { getTransientDatabaseFailure } from '@sim/utils/errors' +import { toRecord } from '@sim/utils/object' +import { backoffWithJitter } from '@sim/utils/retry' +import { and, eq, isNotNull, isNull } from 'drizzle-orm' +import { + type DeferredOutboxHandlerResult, + deferOutboxHandler, + type OutboxHandler, +} from '@/lib/core/outbox/service' +import type { DeferredRetryCheckPayload } from '@/lib/knowledge/documents/processing-outbox-event' +import { + type DocumentProcessingSnapshot, + documentProcessingSnapshotCondition, + inspectDocumentProcessingLiveness, + processingSnapshotColumns, +} from '@/lib/knowledge/documents/processing-recovery-queue' +import { QUEUED_DISPATCH_GRACE_MS, RECOVERY_WINDOW_MS } from '@/lib/knowledge/documents/types' + +const logger = createLogger('KnowledgeDeferredRetryCheck') + +/** How often a document whose run still looks live, or could not be looked up, is checked again. */ +export const DEFERRED_RETRY_RECHECK_MS = 60 * 60 * 1000 + +/** + * How long after the deferral a database failure may still postpone the check without spending + * an attempt. The check itself stops asking about liveness at {@link RECOVERY_WINDOW_MS}; the extra + * day lets that final write outlast a slow database window too. Past it, a database failure spends + * attempts like any other error, so the event always reaches completion or dead letter. + */ +export const DEFERRED_RETRY_CHECK_TERMINAL_MS = RECOVERY_WINDOW_MS + 24 * 60 * 60 * 1000 + +/** First delay after a database failure; later ones grow with how overdue the check is. */ +const DATABASE_BACKOFF_STEP_MS = 2 * 60 * 1000 + +export const DEFERRED_RETRY_LOST_ERROR = + 'The scheduled retry for this document did not run. Retry the document to process it again.' + +function parsePayload(raw: unknown): DeferredRetryCheckPayload { + const record = toRecord(raw) + const knowledgeBaseId = toStringOrNull(record.knowledgeBaseId) + const documentId = toStringOrNull(record.documentId) + const processingDeferredUntil = toStringOrNull(record.processingDeferredUntil) + if (!knowledgeBaseId || !documentId || !processingDeferredUntil) { + throw new Error('Deferred retry check payload is missing its document or deferral') + } + return { + knowledgeBaseId, + documentId, + processingQueueToken: toStringOrNull(record.processingQueueToken), + processingQueuedAt: toStringOrNull(record.processingQueuedAt), + processingDeferredUntil, + } +} + +function sameInstant(value: Date | null, expected: string | null): boolean { + return (value?.getTime() ?? null) === (expected === null ? null : Date.parse(expected)) +} + +/** The uploaded document is still waiting on exactly the deferral this check was scheduled for. */ +function isSameDeferral( + row: DocumentProcessingSnapshot & { + connectorId: string | null + archivedAt: Date | null + deletedAt: Date | null + }, + payload: DeferredRetryCheckPayload +): row is typeof row & { processingDeferredUntil: Date } { + return ( + row.processingStatus === 'pending' && + row.connectorId === null && + row.archivedAt === null && + row.deletedAt === null && + row.processingQueueToken === payload.processingQueueToken && + sameInstant(row.processingQueuedAt, payload.processingQueuedAt) && + row.processingDeferredUntil !== null && + sameInstant(row.processingDeferredUntil, payload.processingDeferredUntil) + ) +} + +/** + * Fails an uploaded document whose scheduled database retry never ran. This is a state transition + * only: nothing is dispatched and no actor is needed, and the user's Retry re-dispatches it as + * them. The write is fenced on the exact snapshot it inspected, so a retry that claims or + * re-defers the document in the meantime is never overwritten. + * + * A run that still looks live, or whose status cannot be established, is checked again later + * without spending an attempt. Past {@link RECOVERY_WINDOW_MS} after the deferral no retry of that + * generation can still be running (a database retry is due within minutes and each run is bounded), + * so the check stops asking and fails the document, which is what guarantees the event ends. + * + * A transient database failure while checking, most likely the same slow window that deferred the + * document, postpones the check without spending an attempt, so the watchdog cannot dead-letter + * during the outage it exists to outlast. That stops at {@link DEFERRED_RETRY_CHECK_TERMINAL_MS}; + * any other error spends an attempt as before. + */ +export const checkDeferredDocumentRetry: OutboxHandler = async ( + rawPayload, + context +): Promise => { + const payload = parsePayload(rawPayload) + try { + return await checkDeferral(payload, context.signal) + } catch (error) { + context.signal.throwIfAborted() + const failure = getTransientDatabaseFailure(error) + const deferredUntil = Date.parse(payload.processingDeferredUntil) + const now = Date.now() + if (!failure || now >= deferredUntil + DEFERRED_RETRY_CHECK_TERMINAL_MS) throw error + /** Paced by how long the check has been overdue, since a deferral spends no attempt to count. */ + const overdueMs = Math.max(0, now - deferredUntil - QUEUED_DISPATCH_GRACE_MS) + return deferOutboxHandler( + `Database ${failure} failure while checking the deferred retry`, + backoffWithJitter(1 + Math.floor(overdueMs / DATABASE_BACKOFF_STEP_MS), null, { + baseMs: DATABASE_BACKOFF_STEP_MS, + maxMs: DEFERRED_RETRY_RECHECK_MS, + }), + false + ) + } +} + +async function checkDeferral( + payload: DeferredRetryCheckPayload, + signal: AbortSignal +): Promise { + signal.throwIfAborted() + const [row] = await db + .select({ + ...processingSnapshotColumns, + connectorId: document.connectorId, + archivedAt: document.archivedAt, + deletedAt: document.deletedAt, + }) + .from(document) + .where( + and( + eq(document.id, payload.documentId), + eq(document.knowledgeBaseId, payload.knowledgeBaseId) + ) + ) + .limit(1) + if (!row || !isSameDeferral(row, payload)) return undefined + + const now = Date.now() + const deferredUntil = row.processingDeferredUntil.getTime() + const dueAt = deferredUntil + QUEUED_DISPATCH_GRACE_MS + if (now < dueAt) { + return deferOutboxHandler('Deferred retry is not overdue yet', dueAt - now, false) + } + if (now < deferredUntil + RECOVERY_WINDOW_MS) { + const { abandoned } = await inspectDocumentProcessingLiveness([row], signal) + if (abandoned.length === 0) { + return deferOutboxHandler( + 'Deferred retry may still be running', + DEFERRED_RETRY_RECHECK_MS, + false + ) + } + } + signal.throwIfAborted() + + const failed = await db + .update(document) + .set({ + processingStatus: 'failed', + processingError: DEFERRED_RETRY_LOST_ERROR, + processingDeferredUntil: null, + processingCompletedAt: new Date(), + }) + .where( + and( + documentProcessingSnapshotCondition(row), + eq(document.processingStatus, 'pending'), + isNotNull(document.processingDeferredUntil), + isNull(document.connectorId), + isNull(document.archivedAt), + isNull(document.deletedAt) + ) + ) + .returning({ id: document.id }) + if (failed.length > 0) { + logger.warn('Uploaded document failed after its scheduled retry did not run', { + documentId: payload.documentId, + }) + } + return undefined +} diff --git a/apps/sim/lib/knowledge/documents/document-processing-source.test.ts b/apps/sim/lib/knowledge/documents/document-processing-source.test.ts index 32e7eed699e..29ceaba5ebf 100644 --- a/apps/sim/lib/knowledge/documents/document-processing-source.test.ts +++ b/apps/sim/lib/knowledge/documents/document-processing-source.test.ts @@ -94,11 +94,15 @@ import { UsageLimitDocumentProcessingError, } from '@/lib/knowledge/documents/document-processing-error' import { KNOWLEDGE_DOCUMENT_CONTINUATION_OUTBOX_EVENT } from '@/lib/knowledge/documents/processing-continuation-dispatch' +import { + DEFERRED_RETRY_CHECK_MAX_ATTEMPTS, + KNOWLEDGE_DOCUMENT_DEFERRED_RETRY_CHECK_EVENT, +} from '@/lib/knowledge/documents/processing-outbox-event' import { knowledgeDocumentProcessingOutboxHandlers } from '@/lib/knowledge/documents/processing-outbox-handler' import type { DocumentProcessingPayload } from '@/lib/knowledge/documents/processing-payload' import { ProviderCapacityContinuationExhaustedError } from '@/lib/knowledge/documents/processing-provider-deferral' import { processDocumentAsync, processDocumentsWithQueue } from '@/lib/knowledge/documents/service' -import { MAX_PROCESSING_ATTEMPTS } from '@/lib/knowledge/documents/types' +import { MAX_PROCESSING_ATTEMPTS, QUEUED_DISPATCH_GRACE_MS } from '@/lib/knowledge/documents/types' const mockEmbeddingCapacity = vi.fn() beforeEach(() => { @@ -747,6 +751,151 @@ describe('processDocumentAsync write guards', () => { expect(guardForStatusWrite('failed')).toBeDefined() }) + describe('a transient database failure', () => { + const databaseError = () => + new DrizzleQueryError( + 'insert private SQL', + ['private bound content'], + Object.assign(new Error('canceling statement due to statement timeout'), { + code: '57014', + }) + ) + + async function failWith(error: Error, scheduleDatabaseRetry: (error: unknown) => Date | null) { + armProviderSource() + mockProcessDocument.mockRejectedValueOnce(error) + return processDocumentAsync( + 'knowledge-base-1', + 'document-1', + PERSISTED_CONTEXT, + {}, + BILLING_ATTRIBUTION, + 'pass-1', + { + chargedAtDispatch: true, + processingQueueToken: 'pass-1', + processingQueuedAt: new Date(), + scheduleDatabaseRetry, + } + ).catch((caught: unknown) => caught) + } + + it('leaves the document pending until its scheduled retry instead of failed', async () => { + const error = databaseError() + const retryAt = new Date(Date.now() + 120_000) + const schedule = vi.fn().mockReturnValue(retryAt) + + expect(await failWith(error, schedule)).toBe(error) + + expect(schedule).toHaveBeenCalledWith(error) + const pending = dbChainMockFns.set.mock.calls.find( + ([value]) => value.processingDeferredUntil === retryAt + )?.[0] + expect(pending).toMatchObject({ + processingStatus: 'pending', + processingError: null, + processingDeferredUntil: retryAt, + processingStartedAt: null, + processingCompletedAt: null, + }) + /** The same run retries, so its queue token and retry budget stay as they are. */ + expect(pending).not.toHaveProperty('processingQueueToken') + expect(pending).not.toHaveProperty('processingAttempts') + /** Stamped only when unset, so a dispatch cannot claim it as never queued meanwhile. */ + expect(pending.processingQueuedAt.toSQL().sql).toMatch(/^COALESCE\(.+, \?\)$/) + expect( + dbChainMockFns.set.mock.calls.some(([value]) => value.processingStatus === 'failed') + ).toBe(false) + expect(guardForStatusWrite('pending')).toBeDefined() + }) + + /** The row the deferral write returns: the document as it now stands. */ + function deferredRow(connectorId: string | null, retryAt: Date, queuedAt: Date) { + return { + id: 'document-1', + knowledgeBaseId: 'knowledge-base-1', + connectorId, + uploadedAt: new Date(0), + processingStatus: 'pending', + processingQueueToken: 'pass-1', + processingQueuedAt: queuedAt, + processingStartedAt: null, + processingDeferredUntil: retryAt, + processingCompletedAt: null, + processingRecoveryAfter: null, + } + } + + function deferredRetryChecks() { + return dbChainMockFns.values.mock.calls + .map(([value]) => value as Record) + .filter((value) => value.eventType === KNOWLEDGE_DOCUMENT_DEFERRED_RETRY_CHECK_EVENT) + } + + it('schedules the lost-retry check for an uploaded document in the deferral transaction', async () => { + const retryAt = new Date(Date.now() + 120_000) + const queuedAt = new Date(Date.now() - 1_000) + dbChainMockFns.returning.mockResolvedValue([deferredRow(null, retryAt, queuedAt)]) + + await failWith(databaseError(), () => retryAt) + + const [check] = deferredRetryChecks() + expect(check).toMatchObject({ + availableAt: new Date(retryAt.getTime() + QUEUED_DISPATCH_GRACE_MS), + maxAttempts: DEFERRED_RETRY_CHECK_MAX_ATTEMPTS, + payload: { + knowledgeBaseId: 'knowledge-base-1', + documentId: 'document-1', + processingQueueToken: 'pass-1', + processingQueuedAt: queuedAt.toISOString(), + processingDeferredUntil: retryAt.toISOString(), + }, + }) + /** The deferral write and the check commit together. */ + const transactionOrder = dbChainMockFns.transaction.mock.invocationCallOrder.at(-1)! + const pendingSetOrder = + dbChainMockFns.set.mock.invocationCallOrder[ + dbChainMockFns.set.mock.calls.findIndex( + ([value]) => value.processingDeferredUntil === retryAt + ) + ] + expect(pendingSetOrder).toBeGreaterThan(transactionOrder) + }) + + it('schedules no check for a connector document, which the recovery sweep covers', async () => { + const retryAt = new Date(Date.now() + 120_000) + dbChainMockFns.returning.mockResolvedValue([deferredRow('connector-1', retryAt, new Date())]) + + await failWith(databaseError(), () => retryAt) + + expect(deferredRetryChecks()).toHaveLength(0) + }) + + it('schedules no check when the deferral write did not land', async () => { + const retryAt = new Date(Date.now() + 120_000) + dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'document-1' }]).mockResolvedValue([]) + + await failWith(databaseError(), () => retryAt) + + expect(deferredRetryChecks()).toHaveLength(0) + }) + + it('records the failure once no retry is scheduled', async () => { + const error = databaseError() + dbChainMockFns.returning.mockResolvedValue([deferredRow(null, new Date(), new Date())]) + expect(await failWith(error, () => null)).toBe(error) + expect(deferredRetryChecks()).toHaveLength(0) + + expect(dbChainMockFns.set).toHaveBeenCalledWith( + expect.objectContaining({ + processingStatus: 'failed', + processingError: 'Database request failed (SQLSTATE 57014).', + processingDeferredUntil: null, + }) + ) + }) + }) + it('records the failed embedding batch without exposing SQL, content or vectors', async () => { armProviderSource() dbChainMockFns.limit.mockResolvedValueOnce([{ id: 'document-1' }]) @@ -1726,6 +1875,38 @@ describe('in-process quota continuation dispatch', () => { expect(JSON.stringify(mockLogError.mock.calls)).not.toContain('private-') }) + it('records a transient database failure as failed in-process, with no retry to wait for', async () => { + mockGenerateEmbeddings.mockRejectedValue( + new DrizzleQueryError( + 'insert private-query', + ['private-parameter'], + Object.assign(new Error('canceling statement due to statement timeout'), { + code: '57014', + }) + ) + ) + + await processDocumentsWithQueue( + [queuedDocument], + 'knowledge-base-1', + {}, + 'request-1', + BILLING_ATTRIBUTION, + 'interactive' + ) + + expect(dbChainMockFns.set).toHaveBeenCalledWith( + expect.objectContaining({ processingStatus: 'failed', processingDeferredUntil: null }) + ) + expect( + dbChainMockFns.set.mock.calls.some( + ([value]) => + (value as Record).processingStatus === 'pending' && + (value as Record).processingDeferredUntil instanceof Date + ) + ).toBe(false) + }) + it('resumes an OCR-throttled regular KB from the durable outbox to a completed index', async () => { mockProcessDocument.mockRejectedValueOnce( new ProviderCapacityDeferredError('rate_limit', { retryAfterMs: 600_000 }) diff --git a/apps/sim/lib/knowledge/documents/processing-outbox-event.ts b/apps/sim/lib/knowledge/documents/processing-outbox-event.ts index ea8d2071ae1..03d632076a2 100644 --- a/apps/sim/lib/knowledge/documents/processing-outbox-event.ts +++ b/apps/sim/lib/knowledge/documents/processing-outbox-event.ts @@ -2,7 +2,9 @@ import type { db } from '@sim/db' import type { BillingAttributionSnapshot } from '@/lib/billing/core/billing-attribution' import { enqueueOutboxEvent } from '@/lib/core/outbox/service' import type { DocumentProcessingLane } from '@/lib/knowledge/documents/processing-payload' +import type { DocumentProcessingSnapshot } from '@/lib/knowledge/documents/processing-recovery-queue' import type { ProcessingOptions } from '@/lib/knowledge/documents/service' +import { QUEUED_DISPATCH_GRACE_MS } from '@/lib/knowledge/documents/types' export const KNOWLEDGE_DOCUMENT_PROCESSING_OUTBOX_EVENT = 'knowledge.document.processing.dispatch' @@ -25,3 +27,47 @@ export function enqueueKnowledgeDocumentProcessing( ): Promise { return enqueueOutboxEvent(executor, KNOWLEDGE_DOCUMENT_PROCESSING_OUTBOX_EVENT, payload) } + +export const KNOWLEDGE_DOCUMENT_DEFERRED_RETRY_CHECK_EVENT = + 'knowledge.document.deferred-retry-check' + +/** Thrown failures only: every wait in `checkDeferredDocumentRetry` defers without spending one. */ +export const DEFERRED_RETRY_CHECK_MAX_ATTEMPTS = 5 + +/** The one deferral a check is about: a later retry, reset, or deferral makes it a no-op. */ +export interface DeferredRetryCheckPayload { + knowledgeBaseId: string + documentId: string + processingQueueToken: string | null + processingQueuedAt: string | null + processingDeferredUntil: string +} + +type DeferredDocument = DocumentProcessingSnapshot & { + knowledgeBaseId: string + processingDeferredUntil: Date +} + +/** + * Schedules the watchdog for an uploaded document just written back to `pending` behind a + * database retry, in the transaction that wrote it. Connector documents have the recovery sweep; + * an uploaded one has no sweep, so without this a lost retry would leave it `pending` for good. + * The check becomes due once the retry is {@link QUEUED_DISPATCH_GRACE_MS} overdue, the same + * grace the retry API applies before it treats a queued generation as lost. + */ +export function enqueueDeferredRetryCheck( + executor: Pick, + deferred: DeferredDocument +): Promise { + const payload: DeferredRetryCheckPayload = { + knowledgeBaseId: deferred.knowledgeBaseId, + documentId: deferred.id, + processingQueueToken: deferred.processingQueueToken, + processingQueuedAt: deferred.processingQueuedAt?.toISOString() ?? null, + processingDeferredUntil: deferred.processingDeferredUntil.toISOString(), + } + return enqueueOutboxEvent(executor, KNOWLEDGE_DOCUMENT_DEFERRED_RETRY_CHECK_EVENT, payload, { + availableAt: new Date(deferred.processingDeferredUntil.getTime() + QUEUED_DISPATCH_GRACE_MS), + maxAttempts: DEFERRED_RETRY_CHECK_MAX_ATTEMPTS, + }) +} diff --git a/apps/sim/lib/knowledge/documents/processing-outbox-handler.ts b/apps/sim/lib/knowledge/documents/processing-outbox-handler.ts index 1989b5f160f..219f0dd6ac3 100644 --- a/apps/sim/lib/knowledge/documents/processing-outbox-handler.ts +++ b/apps/sim/lib/knowledge/documents/processing-outbox-handler.ts @@ -15,6 +15,7 @@ import { detachKnowledgeConnector, KNOWLEDGE_CONNECTOR_DETACH_EVENT, } from '@/lib/knowledge/connectors/detachment' +import { checkDeferredDocumentRetry } from '@/lib/knowledge/documents/deferred-retry-check' import { getOcrRequestRejection, isPermanentDocumentProcessingError, @@ -34,6 +35,7 @@ import { KNOWLEDGE_DOCUMENT_CONTINUATION_OUTBOX_EVENT, } from '@/lib/knowledge/documents/processing-continuation-dispatch' import { + KNOWLEDGE_DOCUMENT_DEFERRED_RETRY_CHECK_EVENT, KNOWLEDGE_DOCUMENT_PROCESSING_OUTBOX_EVENT, type KnowledgeDocumentProcessingOutboxPayload, } from '@/lib/knowledge/documents/processing-outbox-event' @@ -243,6 +245,7 @@ export const knowledgeDocumentProcessingOutboxHandlers = { [KNOWLEDGE_CONNECTOR_DETACH_EVENT]: detachKnowledgeConnector, [KNOWLEDGE_STORAGE_CLEANUP_EVENT]: cleanupKnowledgeStorage, [OCR_CHECKPOINT_CLEANUP_OUTBOX_EVENT]: cleanupOcrCheckpoint, + [KNOWLEDGE_DOCUMENT_DEFERRED_RETRY_CHECK_EVENT]: checkDeferredDocumentRetry, [EMBEDDING_CHECKPOINT_CLEANUP_EVENT]: cleanupEmbeddingCheckpoint, [KNOWLEDGE_DOCUMENT_PROCESSING_OUTBOX_EVENT]: withOutboxHandlerTimeout( processKnowledgeDocument, diff --git a/apps/sim/lib/knowledge/documents/processing-recovery-policy.ts b/apps/sim/lib/knowledge/documents/processing-recovery-policy.ts index f30246e20b6..d3f5cd8ef88 100644 --- a/apps/sim/lib/knowledge/documents/processing-recovery-policy.ts +++ b/apps/sim/lib/knowledge/documents/processing-recovery-policy.ts @@ -1,9 +1,11 @@ import { document, outboxEvent } from '@sim/db/schema' import { and, eq, gt, isNotNull, isNull, lt, lte, or, sql } from 'drizzle-orm' import { DOCUMENT_PROCESSING_STALE_THRESHOLD_MS } from '@/lib/knowledge/documents/processing-timeouts.server' -import { MAX_PROCESSING_ATTEMPTS, QUEUED_DISPATCH_GRACE_MS } from '@/lib/knowledge/documents/types' - -const RECOVERY_WINDOW_MS = 7 * 24 * 60 * 60 * 1000 +import { + MAX_PROCESSING_ATTEMPTS, + QUEUED_DISPATCH_GRACE_MS, + RECOVERY_WINDOW_MS, +} from '@/lib/knowledge/documents/types' /** One eligibility predicate is rechecked under the lifecycle locks before replacing a generation. */ export function documentProcessingRecoveryCondition( diff --git a/apps/sim/lib/knowledge/documents/service.ts b/apps/sim/lib/knowledge/documents/service.ts index 36732b8e036..ac927088c3b 100644 --- a/apps/sim/lib/knowledge/documents/service.ts +++ b/apps/sim/lib/knowledge/documents/service.ts @@ -104,7 +104,10 @@ import { } from '@/lib/knowledge/documents/processing-claim' import type { DocumentProcessingContinuation } from '@/lib/knowledge/documents/processing-continuation-dispatch' import { documentProcessingQueueOptions } from '@/lib/knowledge/documents/processing-lane' -import { enqueueKnowledgeDocumentProcessing } from '@/lib/knowledge/documents/processing-outbox-event' +import { + enqueueDeferredRetryCheck, + enqueueKnowledgeDocumentProcessing, +} from '@/lib/knowledge/documents/processing-outbox-event' import { assertDocumentProcessingBillingContext, createDocumentProcessingPayload, @@ -1393,10 +1396,24 @@ export interface DocumentProcessingAttemptContext extends DocumentProcessingExec readonly scheduleProviderContinuation?: ( error: ProviderCapacityDeferredError ) => Promise + /** + * Schedules the next attempt after a transient database failure and returns when it runs, or + * null when this failure is not retried that way. A scheduled document is left `pending` until + * then instead of `failed`, so a slow database window does not read as a bad document. + */ + readonly scheduleDatabaseRetry?: (error: unknown) => Date | null /** Signals that this invocation owns the persisted processing generation. */ readonly onClaimed?: () => void } +/** + * Processes documents in this process when no Trigger.dev worker can take them. + * + * Deliberately supplies no `scheduleDatabaseRetry`: nothing here can durably wait out a slow + * database window, since an in-memory timer dies with the process and would leave the document + * `pending` with no run behind it. A transient database failure is therefore recorded `failed`, + * which the document's Retry action and the retry API both accept. + */ async function dispatchInProcess( jobPayloads: DocumentProcessingPayload[], requestId: string, @@ -2175,10 +2192,13 @@ export async function processDocumentAsync( recordedError = continuationError } } - const deferredUntil = continuation?.deferredUntil ?? null + const databaseRetryAt = continuation?.deferredUntil + ? null + : (attemptContext?.scheduleDatabaseRetry?.(recordedError) ?? null) + const deferredUntil = continuation?.deferredUntil ?? databaseRetryAt const providerContinuationExhausted = recordedError instanceof ProviderCapacityContinuationExhaustedError - const quotaContinuationFailed = quotaContinuationAttempted && !deferredUntil + const quotaContinuationFailed = quotaContinuationAttempted && !continuation?.deferredUntil const failureDiagnostic = getConnectorFailureDiagnostic(recordedError) const errorMessage = byokCredentialRejected ? BYOK_EMBEDDING_CREDENTIAL_REJECTION_MESSAGE @@ -2220,43 +2240,73 @@ export async function processDocumentAsync( logger.error(logMessage, logContext) } - await db - .update(document) - .set({ - processingStatus: deferredUntil ? 'pending' : 'failed', - processingError: deferredUntil ? null : errorMessage, - processingStartedAt: deferredUntil ? null : processingStartedAt, - ...(continuation + const failureStatus = { + processingStatus: deferredUntil ? 'pending' : 'failed', + processingError: deferredUntil ? null : errorMessage, + processingStartedAt: deferredUntil ? null : processingStartedAt, + ...(continuation + ? { + processingQueuedAt: continuation.deferredUntil, + processingQueueToken: continuation.processingQueueToken, + } + : databaseRetryAt ? { - processingQueuedAt: continuation.deferredUntil, - processingQueueToken: continuation.processingQueueToken, + /** + * Stamps the queue like a continuation does, so a dispatch cannot claim the row as + * never-queued while its retry is scheduled. The retry itself claims by token, and + * an existing stamp is kept because a legacy retry claims by that stamp. + */ + processingQueuedAt: sql`COALESCE(${document.processingQueuedAt}, ${sql.param(databaseRetryAt, document.processingQueuedAt)})`, } : {}), - processingDeferredUntil: deferredUntil, - processingCompletedAt: deferredUntil ? null : new Date(), - ...(permanentError || - ocrRequestRejected || - byokCredentialRejected || - providerContinuationExhausted || - (embeddingQuotaExhausted && attemptContext?.quotaContinuationExhausted) - ? { processingAttempts: MAX_PROCESSING_ATTEMPTS } - : (embeddingQuotaExhausted || usageLimitExceeded || providerDeferral) && - attemptContext?.chargedAtDispatch - ? { processingAttempts: sql`GREATEST(${document.processingAttempts} - 1, 0)` } - : {}), + processingDeferredUntil: deferredUntil, + processingCompletedAt: deferredUntil ? null : new Date(), + ...(permanentError || + ocrRequestRejected || + byokCredentialRejected || + providerContinuationExhausted || + (embeddingQuotaExhausted && attemptContext?.quotaContinuationExhausted) + ? { processingAttempts: MAX_PROCESSING_ATTEMPTS } + : (embeddingQuotaExhausted || usageLimitExceeded || providerDeferral) && + attemptContext?.chargedAtDispatch + ? { processingAttempts: sql`GREATEST(${document.processingAttempts} - 1, 0)` } + : {}), + } + const failureStatusGuard = and( + eq(document.id, documentId), + eq(document.processingStatus, 'processing'), + eq(document.processingStartedAt, processingStartedAt), + ...queueGenerationConditions(attemptContext), + eq(document.userExcluded, false), + isNull(document.archivedAt), + isNull(document.deletedAt), + documentConnectorIsActive() + ) + if (databaseRetryAt) { + /** + * An uploaded document has no recovery sweep, so the deferral and the watchdog that fails it + * if this retry never runs commit together. + */ + await db.transaction(async (tx) => { + const [deferred] = await tx + .update(document) + .set(failureStatus) + .where(failureStatusGuard) + .returning({ + ...processingSnapshotColumns, + knowledgeBaseId: document.knowledgeBaseId, + connectorId: document.connectorId, + }) + if (deferred && deferred.connectorId === null && deferred.processingDeferredUntil) { + await enqueueDeferredRetryCheck(tx, { + ...deferred, + processingDeferredUntil: deferred.processingDeferredUntil, + }) + } }) - .where( - and( - eq(document.id, documentId), - eq(document.processingStatus, 'processing'), - eq(document.processingStartedAt, processingStartedAt), - ...queueGenerationConditions(attemptContext), - eq(document.userExcluded, false), - isNull(document.archivedAt), - isNull(document.deletedAt), - documentConnectorIsActive() - ) - ) + } else { + await db.update(document).set(failureStatus).where(failureStatusGuard) + } throw recordedError } diff --git a/apps/sim/lib/knowledge/documents/types.ts b/apps/sim/lib/knowledge/documents/types.ts index 4bf7435f690..8e2ab0bf46a 100644 --- a/apps/sim/lib/knowledge/documents/types.ts +++ b/apps/sim/lib/knowledge/documents/types.ts @@ -47,6 +47,9 @@ export const MAX_PROCESSING_ATTEMPTS = 5 */ export const QUEUED_DISPATCH_GRACE_MS = 240 * 60 * 1000 +/** How long after a document's upload, or a deferred retry, automatic recovery still acts on it. */ +export const RECOVERY_WINDOW_MS = 7 * 24 * 60 * 60 * 1000 + /** Worst-case wall clock for one processing run across its retry budget. */ export function worstCaseProcessingMinutes( maxDurationSeconds: number, diff --git a/apps/sim/lib/workspace-files/search/constants.ts b/apps/sim/lib/workspace-files/search/constants.ts index 823795f93a7..ecf3d6c41de 100644 --- a/apps/sim/lib/workspace-files/search/constants.ts +++ b/apps/sim/lib/workspace-files/search/constants.ts @@ -78,12 +78,13 @@ export const FILE_SEARCH_INDEX_TRANSACTION_LIMITS = { transactionTimeout: 30 * 1000, } as const -/** Attempts for failures other than database capacity cancellations. */ +/** Attempts for failures other than transient database failures. */ export const FILE_SEARCH_INDEX_MAX_ATTEMPTS = 3 /** * Attempts when PostgreSQL cancels an indexing statement on a timeout. One row's direct GIN insert * is not interruptible, so under storage saturation even a single ordinary chunk can outlive the - * statement deadline; smaller batches cannot help, only waiting out the slow window can. + * statement deadline; smaller batches cannot help, only waiting out the slow window can. Deadlocks, + * serialization failures, and dropped connections share the same budget and pacing. */ export const FILE_SEARCH_INDEX_CAPACITY_MAX_ATTEMPTS = 6 /** First capacity retry delay; later ones double up to the ceiling, about an hour in total. */ diff --git a/apps/sim/lib/workspace-files/search/indexing.test.ts b/apps/sim/lib/workspace-files/search/indexing.test.ts index ed6b6412551..59b80a18064 100644 --- a/apps/sim/lib/workspace-files/search/indexing.test.ts +++ b/apps/sim/lib/workspace-files/search/indexing.test.ts @@ -217,6 +217,11 @@ describe('indexing retry policy', () => { it.each([ ['statement timeout', 'canceling statement due to statement timeout', '57014'], ['lock timeout', 'canceling statement due to lock timeout', '55P03'], + ['transaction timeout', 'terminating connection due to transaction timeout', '25P04'], + ['deadlock', 'deadlock detected', '40P01'], + ['serialization failure', 'could not serialize access', '40001'], + ['dropped connection', 'write CONNECTION_CLOSED', 'CONNECTION_CLOSED'], + ['connection reset', 'read ECONNRESET', 'ECONNRESET'], ])('waits minutes, not seconds, after a %s', async (_label, message, code) => { const thrown = await thrownBy(statementTimeout(message, code)) const first = delayOf(getWorkspaceFileSearchRetry(thrown, 1, now)) @@ -252,6 +257,22 @@ describe('indexing retry policy', () => { ) }) + it('treats a reset outside the database as an ordinary failure', async () => { + vi.clearAllMocks() + mocks.begin.mockResolvedValue({ id: 'build', ...payload }) + mocks.file.mockResolvedValue({ + name: 'notes.txt', + size: 100, + contentUpdatedAt: new Date(payload.sourceContentUpdatedAt), + }) + mocks.load.mockRejectedValue( + Object.assign(new Error('read ECONNRESET'), { code: 'ECONNRESET' }) + ) + const thrown = await indexWorkspaceFileForSearch(payload, signal).catch((caught) => caught) + expect(thrown).toMatchObject({ code: 'ECONNRESET' }) + expect(getWorkspaceFileSearchRetry(thrown, 1, now)).toBeUndefined() + }) + it('treats a user cancellation as an ordinary failure', async () => { const thrown = await thrownBy(statementTimeout('canceling statement due to user request')) expect(getWorkspaceFileSearchRetry(thrown, 1, now)).toBeUndefined() diff --git a/apps/sim/lib/workspace-files/search/indexing.ts b/apps/sim/lib/workspace-files/search/indexing.ts index 1c04091bbdd..6002ab85d7c 100644 --- a/apps/sim/lib/workspace-files/search/indexing.ts +++ b/apps/sim/lib/workspace-files/search/indexing.ts @@ -1,7 +1,11 @@ import { Buffer } from 'node:buffer' import { createLogger } from '@sim/logger' -import { describeError, getPostgresCancellationReason } from '@sim/utils/errors' -import { backoffWithJitter } from '@sim/utils/retry' +import { describeError } from '@sim/utils/errors' +import { + type BackgroundRetryDecision, + type BackgroundRetryPolicy, + getBackgroundRetryDecision, +} from '@/lib/core/errors/background-retry' import { redactDatabaseQueryError } from '@/lib/core/errors/database-query-error' import { isPayloadSizeLimitError } from '@/lib/core/utils/stream-limits' import { getWorkspaceFile } from '@/lib/uploads/contexts/workspace' @@ -160,32 +164,26 @@ export async function markWorkspaceFileSearchIndexFailed( await failFileSearchRevision(parseRevision(payload), payload.dispatchToken) } -const CAPACITY_CANCELLATIONS = new Set(['statement_timeout', 'lock_timeout', 'transaction_timeout']) - -export type WorkspaceFileSearchRetryDecision = - | { retryAt: Date } - | { skipRetrying: true } - | undefined +const FILE_SEARCH_INDEX_RETRY_POLICY: BackgroundRetryPolicy = { + maxAttempts: FILE_SEARCH_INDEX_MAX_ATTEMPTS, + database: { + maxAttempts: FILE_SEARCH_INDEX_CAPACITY_MAX_ATTEMPTS, + baseDelayMs: FILE_SEARCH_INDEX_CAPACITY_RETRY_BASE_MS, + maxDelayMs: FILE_SEARCH_INDEX_CAPACITY_RETRY_MAX_MS, + }, +} /** - * Chooses the next attempt after `attempt` (1-based) failed. A statement, lock, or transaction - * timeout means the database had no capacity for this build right now, not that the file is bad: - * those back off for minutes so the retries outlast a slow window instead of all landing inside - * it. Anything else keeps the ordinary short retries. `undefined` keeps the runner's default delay. + * Chooses the next attempt after `attempt` (1-based) failed. A transient database failure (a + * statement, lock, or transaction timeout, a deadlock, or a dropped connection) means the database + * could not take this build right now, not that the file is bad: those back off for minutes so the + * retries outlast a slow window instead of all landing inside it. Anything else keeps the ordinary + * short retries. `undefined` keeps the runner's default delay. */ export function getWorkspaceFileSearchRetry( error: unknown, attempt: number, now = Date.now() -): WorkspaceFileSearchRetryDecision { - const reason = getPostgresCancellationReason(error) - if (reason && CAPACITY_CANCELLATIONS.has(reason)) { - if (attempt >= FILE_SEARCH_INDEX_CAPACITY_MAX_ATTEMPTS) return { skipRetrying: true } - const delayMs = backoffWithJitter(attempt, null, { - baseMs: FILE_SEARCH_INDEX_CAPACITY_RETRY_BASE_MS, - maxMs: FILE_SEARCH_INDEX_CAPACITY_RETRY_MAX_MS, - }) - return { retryAt: new Date(now + delayMs) } - } - return attempt >= FILE_SEARCH_INDEX_MAX_ATTEMPTS ? { skipRetrying: true } : undefined +): BackgroundRetryDecision { + return getBackgroundRetryDecision(error, attempt, FILE_SEARCH_INDEX_RETRY_POLICY, now) } diff --git a/packages/db/script-migrations/0021_embedding_search_connector.test.ts b/packages/db/script-migrations/0021_embedding_search_connector.test.ts index d2218247e7a..de39d4282b4 100644 --- a/packages/db/script-migrations/0021_embedding_search_connector.test.ts +++ b/packages/db/script-migrations/0021_embedding_search_connector.test.ts @@ -13,15 +13,27 @@ const untouched = { begin: vi.fn() } as unknown as Sql type PageRow = { scanned: number; filled: number; last_id: string | null } -/** The message the database pairs with each cancellation SQLSTATE. */ -const CANCELLATION_MESSAGES: Record = { +/** The message the database pairs with each SQLSTATE below. */ +const SERVER_MESSAGES: Record = { '55P03': 'canceling statement due to lock timeout', '57014': 'canceling statement due to statement timeout', + '40P01': 'deadlock detected', } -/** A driver error carrying a SQLSTATE, the shape `postgres` throws. */ -function postgresError(code: string, message = CANCELLATION_MESSAGES[code] ?? 'failed'): Error { - return Object.assign(new Error(`${message} (SQLSTATE ${code})`), { code }) +/** + * The error `postgres` throws for `code`: a SQLSTATE carries the server's message as is, and a + * lost connection is the driver's own connection error (`write `). + */ +function postgresError(code: string, message = SERVER_MESSAGES[code] ?? 'failed'): Error { + if (code.startsWith('CONNECTION_')) { + return Object.assign(new Error(`write ${code} localhost:5432`), { + code, + errno: code, + address: 'localhost', + port: 5432, + }) + } + return Object.assign(new Error(message), { code }) } /** @@ -65,8 +77,8 @@ describe('backfillProjectionSourceAcl', () => { vi.useRealTimers() }) - it.each(['55P03', '57014'])( - 'retries the page after a %s timeout and moves the cursor only once it commits', + it.each(['55P03', '57014', '40P01', 'CONNECTION_CLOSED'])( + 'retries the page after a %s failure and moves the cursor only once it commits', async (code) => { const { session, cursors } = sessionOf([ { scanned: 2, filled: 2, last_id: 'id-2' }, @@ -86,18 +98,18 @@ describe('backfillProjectionSourceAcl', () => { const { session, cursors } = sessionOf( Array.from({ length: PROJECTION_SOURCE_ACL_PAGE_RETRIES + 1 }, () => postgresError('55P03')) ) - await expect(backfillNow(session, 'embedding_keyword_tin', { pauseMs: 0 })).rejects.toThrow( - 'SQLSTATE 55P03' - ) + await expect( + backfillNow(session, 'embedding_keyword_tin', { pauseMs: 0 }) + ).rejects.toMatchObject({ code: '55P03' }) expect(cursors).toHaveLength(PROJECTION_SOURCE_ACL_PAGE_RETRIES + 1) expect(new Set(cursors)).toEqual(new Set([''])) }) it('propagates an error that is not a timeout without retrying', async () => { const { session, cursors } = sessionOf([postgresError('42P01')]) - await expect(backfillNow(session, 'embedding_search', { pauseMs: 0 })).rejects.toThrow( - 'SQLSTATE 42P01' - ) + await expect(backfillNow(session, 'embedding_search', { pauseMs: 0 })).rejects.toMatchObject({ + code: '42P01', + }) expect(cursors).toEqual(['']) }) diff --git a/packages/db/script-migrations/0021_embedding_search_connector.ts b/packages/db/script-migrations/0021_embedding_search_connector.ts index b5b6e3214ef..3fdfec7027d 100644 --- a/packages/db/script-migrations/0021_embedding_search_connector.ts +++ b/packages/db/script-migrations/0021_embedding_search_connector.ts @@ -1,4 +1,5 @@ import { createLogger } from '@sim/logger' +import { getPostgresErrorCode, getTransientDatabaseFailure } from '@sim/utils/errors' import { sleep } from '@sim/utils/helpers' import { backoffWithJitter } from '@sim/utils/retry' import postgres, { type Sql, type TransactionSql } from 'postgres' @@ -32,36 +33,6 @@ export const PROJECTION_SOURCE_ACL_PAGE_RETRIES = 12 /** Pause before a page is retried: 10 s, doubling to a 60 s base with up to 20% jitter (about 72 s). */ const PAGE_RETRY_PAUSE = { baseMs: 10_000, maxMs: 60_000 } as const -/** The SQLSTATE on a driver error, or on the error it wraps. */ -function postgresErrorCode(error: unknown): string | undefined { - if (typeof error !== 'object' || error === null) return undefined - const code = (error as { code?: unknown }).code - if (typeof code === 'string') return code - return postgresErrorCode((error as { cause?: unknown }).cause) -} - -/** The message on a driver error, or on the error it wraps. */ -function postgresErrorMessage(error: unknown): string | undefined { - if (typeof error !== 'object' || error === null) return undefined - const message = (error as { message?: unknown }).message - if (typeof message === 'string') return message - return postgresErrorMessage((error as { cause?: unknown }).cause) -} - -/** - * The two ways the database cancels a page: `lock_timeout` (55P03) while the page's index write - * waits on a lock the index's background maintenance holds, and `statement_timeout` (57014) when - * the page itself runs past {@link PROJECTION_SOURCE_ACL_PAGE_TIMEOUT_MS}. Both pass once the - * maintenance moves on, so both are retried the same way. 57014 is also what an explicit - * cancellation raises, and that is not retried: only the message tells the two apart. - */ -function isPageTimeout(error: unknown): boolean { - const code = postgresErrorCode(error) - if (code === '55P03') return true - if (code !== '57014') return false - return postgresErrorMessage(error)?.includes('statement timeout') ?? false -} - /** Pages between progress log lines. */ const PROGRESS_EVERY_PAGES = 100 @@ -244,15 +215,25 @@ export async function backfillProjectionSourceAcl( return row }) } catch (error) { - if (!isPageTimeout(error)) throw error - const code = postgresErrorCode(error) + /** + * The page is cancelled on `lock_timeout` (55P03) while its index write waits on a lock the + * index's background maintenance holds, and on `statement_timeout` (57014) when it runs past + * {@link PROJECTION_SOURCE_ACL_PAGE_TIMEOUT_MS}; both pass once the maintenance moves on. A + * deadlock, a serialization failure, or a dropped connection rolls the page back the same + * way, and a page only fills rows still unset, so all are retried in place. 57014 is also + * what an explicit cancellation raises, and that is not retried. + */ + const failure = getTransientDatabaseFailure(error) + if (!failure) throw error + const code = getPostgresErrorCode(error) timeouts += 1 if (timeouts > PROJECTION_SOURCE_ACL_PAGE_RETRIES) throw error if (Date.now() >= deadline) break const pauseMs = backoffWithJitter(timeouts, null, PAGE_RETRY_PAUSE) - logger.warn('Projection source and ACL backfill page timed out; retrying', { + logger.warn('Projection source and ACL backfill page failed transiently; retrying', { projection, afterId, + failure, code, attempt: timeouts, retryInMs: Math.round(pauseMs), diff --git a/packages/db/scripts/database-failure-classification.postgres.test.ts b/packages/db/scripts/database-failure-classification.postgres.test.ts new file mode 100644 index 00000000000..f5f43a5fda4 --- /dev/null +++ b/packages/db/scripts/database-failure-classification.postgres.test.ts @@ -0,0 +1,80 @@ +import { migrationTestDatabaseUrl } from '@sim/db/scripts/migration-fixture' +import { classifyDatabaseFailure } from '@sim/utils/errors' +import { sql as statement } from 'drizzle-orm' +import { drizzle } from 'drizzle-orm/postgres-js' +import postgres from 'postgres' +import { describe, expect, it } from 'vitest' + +/** A port on the database host with nothing listening, so a connection is refused at once. */ +function closedPortUrl(): string { + const url = new URL(migrationTestDatabaseUrl!) + url.port = '1' + return url.toString() +} + +async function rejectionOf(work: () => Promise): Promise { + try { + await work() + } catch (error) { + return error + } + throw new Error('Expected the work to fail') +} + +/** + * Classifies the errors postgres.js and Drizzle actually raise, rather than hand-built shapes: + * a transaction that loses its connection is rejected with the driver's bare connection error, + * with no query attached and no Drizzle wrapper, and must still read as a connection failure. + */ +describe.skipIf(!migrationTestDatabaseUrl)('database failure classification', () => { + it('reads a connection terminated mid-transaction as a connection failure', async () => { + const admin = postgres(migrationTestDatabaseUrl!, { max: 1 }) + const client = postgres(migrationTestDatabaseUrl!, { max: 1 }) + try { + const error = await rejectionOf(() => + drizzle(client).transaction(async (tx) => { + const [row] = await tx.execute<{ pid: number }>(statement`SELECT pg_backend_pid() AS pid`) + await admin`SELECT pg_terminate_backend(${row.pid})` + await tx.execute(statement`SELECT pg_sleep(0.2)`) + }) + ) + expect(classifyDatabaseFailure(error)).toBe('connection') + } finally { + await client.end({ timeout: 1 }).catch(() => {}) + await admin.end() + } + }) + + it('reads a refused connection as a connection failure, in a query and in a transaction', async () => { + const client = postgres(closedPortUrl(), { max: 1, connect_timeout: 2 }) + try { + const db = drizzle(client) + expect( + classifyDatabaseFailure(await rejectionOf(() => db.execute(statement`SELECT 1`))) + ).toBe('connection') + expect( + classifyDatabaseFailure( + await rejectionOf(() => db.transaction((tx) => tx.execute(statement`SELECT 1`))) + ) + ).toBe('connection') + } finally { + await client.end({ timeout: 1 }).catch(() => {}) + } + }) + + it('reads a transaction begun on an ending pool as a connection failure', async () => { + const client = postgres(migrationTestDatabaseUrl!, { max: 1 }) + await client`SELECT 1` + const ending = client.end({ timeout: 5 }) + const error = await rejectionOf(() => + drizzle(client).transaction((tx) => tx.execute(statement`SELECT 1`)) + ) + await ending + expect(classifyDatabaseFailure(error)).toBe('connection') + }) + + it('does not read a refused connection from another client as a database failure', async () => { + const error = await rejectionOf(() => fetch(`http://${new URL(closedPortUrl()).host}/`)) + expect(classifyDatabaseFailure(error)).toBe('permanent') + }) +}) diff --git a/packages/utils/src/errors.test.ts b/packages/utils/src/errors.test.ts index 0aa6b585c9e..af41c427b46 100644 --- a/packages/utils/src/errors.test.ts +++ b/packages/utils/src/errors.test.ts @@ -3,10 +3,12 @@ */ import { + classifyDatabaseFailure, describeError, findCause, getPostgresCancellationReason, getPostgresErrorCode, + getTransientDatabaseFailure, toError, } from '@sim/utils/errors' import { describe, expect, it } from 'vitest' @@ -66,6 +68,194 @@ describe('getPostgresCancellationReason', () => { }) }) +/** The driver error postgres.js throws, wrapped the way Drizzle wraps a failed query. */ +function failedQuery(code: string, message = 'private driver detail'): Error { + const driver = Object.assign(new Error(message), { code }) + return Object.assign(new Error('Failed query: private SQL\nparams: private'), { + query: 'private SQL', + params: ['private'], + cause: driver, + }) +} + +describe('classifyDatabaseFailure', () => { + it.each([ + ['57014', 'canceling statement due to statement timeout', 'capacity'], + ['55P03', 'canceling statement due to lock timeout', 'capacity'], + ['25P03', 'terminating connection due to idle-in-transaction timeout', 'capacity'], + ['25P04', 'terminating connection due to transaction timeout', 'capacity'], + ['53300', 'sorry, too many clients already', 'capacity'], + ['40P01', 'deadlock detected', 'conflict'], + ['40001', 'could not serialize access due to concurrent update', 'conflict'], + ['08000', 'connection exception', 'connection'], + ['08006', 'connection failure', 'connection'], + ['08P01', 'protocol violation', 'connection'], + ['57P01', 'terminating connection due to administrator command', 'connection'], + ['57P03', 'the database system is starting up', 'connection'], + ['CONNECTION_CLOSED', 'write CONNECTION_CLOSED', 'connection'], + ['CONNECTION_DESTROYED', 'write CONNECTION_DESTROYED', 'connection'], + ['CONNECTION_ENDED', 'write CONNECTION_ENDED', 'connection'], + ['CONNECT_TIMEOUT', 'write CONNECT_TIMEOUT', 'connection'], + ['ECONNRESET', 'read ECONNRESET', 'connection'], + ['EPIPE', 'write EPIPE', 'connection'], + ['ETIMEDOUT', 'connect ETIMEDOUT', 'connection'], + ['ECONNREFUSED', 'connect ECONNREFUSED', 'connection'], + ['EHOSTUNREACH', 'connect EHOSTUNREACH', 'connection'], + ['ENOTFOUND', 'getaddrinfo ENOTFOUND', 'connection'], + ['EAI_AGAIN', 'getaddrinfo EAI_AGAIN', 'connection'], + ['ENETDOWN', 'connect ENETDOWN', 'connection'], + ['ENETRESET', 'read ENETRESET', 'connection'], + ['ENETUNREACH', 'connect ENETUNREACH', 'connection'], + ])('classifies %s through a query wrapper', (code, message, expected) => { + const wrapped = failedQuery(code, message) + expect(classifyDatabaseFailure(wrapped)).toBe(expected) + expect(classifyDatabaseFailure(new Error('task wrapper', { cause: wrapped }))).toBe(expected) + expect(getTransientDatabaseFailure(wrapped)).toBe(expected) + }) + + it('treats an explicit cancellation as permanent although it shares the timeout SQLSTATE', () => { + const cancelled = failedQuery('57014', 'canceling statement due to user request') + expect(classifyDatabaseFailure(cancelled)).toBe('permanent') + expect(getTransientDatabaseFailure(cancelled)).toBeUndefined() + }) + + it('does not read a timeout into a 57014 with an unrecognized message', () => { + expect(classifyDatabaseFailure(failedQuery('57014', 'private-value'))).toBe('permanent') + }) + + it.each(['23505', '42P01', '22P02', 'XX000'])('treats %s as permanent', (code) => { + expect(classifyDatabaseFailure(failedQuery(code))).toBe('permanent') + }) + + it.each([ + 'ECONNRESET', + 'EPIPE', + 'ETIMEDOUT', + 'ECONNREFUSED', + 'EHOSTUNREACH', + 'ENOTFOUND', + 'EAI_AGAIN', + 'ENETDOWN', + 'ENETRESET', + 'ENETUNREACH', + ])('treats %s with no database query in its chain as permanent', (code) => { + const download = Object.assign(new Error(`socket ${code}`), { code }) + expect(classifyDatabaseFailure(download)).toBe('permanent') + expect(classifyDatabaseFailure(new Error('fetch failed', { cause: download }))).toBe( + 'permanent' + ) + }) + + /** The properties postgres.js defines on an error for a query in flight, non-enumerable. */ + function withDriverQuery(error: Error): Error { + return Object.defineProperties(error, { + query: { value: 'private SQL', enumerable: false }, + parameters: { value: ['private'], enumerable: false }, + args: { value: ['private'], enumerable: false }, + types: { value: undefined, enumerable: false }, + }) + } + + it('counts a socket error the driver raised with its query attached', () => { + const driver = withDriverQuery( + Object.assign(new Error('read ECONNRESET'), { code: 'ECONNRESET' }) + ) + expect(classifyDatabaseFailure(driver)).toBe('connection') + }) + + it.each([ + [ + 'a GraphQL client error with its query and variables', + { query: 'query Items { items { id } }', variables: { first: 50 } }, + ], + ['a query string beside a params array', { query: 'items', params: ['page'] }], + ['a query string beside a parameters array', { query: 'items', parameters: ['page'] }], + ])('treats a socket error under %s as permanent', (_label, fields) => { + const reset = Object.assign(new Error('read ECONNRESET'), { code: 'ECONNRESET' }) + const client = Object.assign(new Error('Request failed'), { ...fields, cause: reset }) + expect(classifyDatabaseFailure(client)).toBe('permanent') + expect(getTransientDatabaseFailure(client)).toBeUndefined() + }) + + /** A connection error as postgres.js builds one (`Errors.connection` in its source). */ + function driverConnectionError(code: string): Error { + return Object.assign(new Error(`write ${code} localhost:5432`), { + code, + errno: code, + address: 'localhost', + port: 5432, + }) + } + + it.each(['CONNECTION_CLOSED', 'CONNECTION_DESTROYED', 'CONNECTION_ENDED', 'CONNECT_TIMEOUT'])( + 'reads a bare driver-built %s as a connection failure, as a lost transaction raises it', + (code) => { + expect(classifyDatabaseFailure(driverConnectionError(code))).toBe('connection') + } + ) + + it.each([ + ['only the code', (code: string) => Object.assign(new Error('socket closed'), { code })], + [ + 'a different message', + (code: string) => Object.assign(driverConnectionError(code), { message: 'socket closed' }), + ], + [ + 'no errno', + (code: string) => + Object.assign(new Error(`write ${code} localhost:5432`), { code, address: 'localhost' }), + ], + [ + 'no address', + (code: string) => + Object.assign(new Error(`write ${code} localhost:5432`), { code, errno: code }), + ], + ])('does not read a CONNECTION_CLOSED error with %s as a database failure', (_label, build) => { + const foreign = build('CONNECTION_CLOSED') + expect(classifyDatabaseFailure(foreign)).toBe('permanent') + expect(classifyDatabaseFailure(new Error('provider request failed', { cause: foreign }))).toBe( + 'permanent' + ) + }) + + it('reads a foreign CONNECTION_CLOSED under a database query error as a connection failure', () => { + const foreign = Object.assign(new Error('socket closed'), { code: 'CONNECTION_CLOSED' }) + const wrapped = Object.assign(new Error('Failed query: private SQL\nparams: '), { + query: 'private SQL', + params: [], + cause: foreign, + }) + expect(classifyDatabaseFailure(wrapped)).toBe('connection') + }) + + it('reads a refused connection the driver took a query for, before it built the query', () => { + const refused = Object.defineProperties( + Object.assign(new AggregateError([], ''), { code: 'ECONNREFUSED' }), + { + query: { value: undefined, enumerable: false }, + parameters: { value: undefined, enumerable: false }, + args: { value: [], enumerable: false }, + types: { value: undefined, enumerable: false }, + } + ) + expect(classifyDatabaseFailure(refused)).toBe('connection') + }) + + it('treats failures without a code as permanent', () => { + expect(classifyDatabaseFailure(new Error('boom'))).toBe('permanent') + expect(classifyDatabaseFailure('boom')).toBe('permanent') + expect(classifyDatabaseFailure(undefined)).toBe('permanent') + }) + + it('classifies by the first code in the chain', () => { + const outer = Object.assign(new Error('unique'), { + code: '23505', + cause: failedQuery('40P01', 'deadlock detected'), + }) + expect(classifyDatabaseFailure(outer)).toBe('permanent') + }) +}) + describe('toError', () => { it('returns the same Error when given an Error', () => { const err = new Error('test') diff --git a/packages/utils/src/errors.ts b/packages/utils/src/errors.ts index d0b657b673e..14248fb4e25 100644 --- a/packages/utils/src/errors.ts +++ b/packages/utils/src/errors.ts @@ -62,6 +62,164 @@ export function getPostgresCancellationReason( return undefined } +/** + * How a failed database operation should be treated by a background job. + * + * - `capacity`: the database had no room for the work right now (a statement, lock, transaction, + * or idle-in-transaction timeout; too many connections). Waiting out the slow window helps. + * - `conflict`: the transaction lost to a concurrent one (deadlock, serialization failure). + * Running it again from the start succeeds. + * - `connection`: the connection to the database failed or was closed under the query. + * - `permanent`: anything else, including failures that did not come from the database at all. + * Whether to retry it is the caller's ordinary policy, not this classification's. + */ +export type DatabaseFailureClass = 'capacity' | 'conflict' | 'connection' | 'permanent' + +export type TransientDatabaseFailureClass = Exclude + +/** 57014 is handled apart; `25P04` is `transaction_timeout`, `25P03` `idle_in_transaction_session_timeout`. */ +const CAPACITY_CODES = new Set(['55P03', '25P03', '25P04', '53300']) +const CONFLICT_CODES = new Set(['40P01', '40001']) + +/** The server shutting down or not yet accepting connections, as during a restart or failover. */ +const SERVER_UNAVAILABLE_SQLSTATES = new Set(['57P01', '57P02', '57P03']) + +/** + * postgres.js's own codes for a lost or unavailable connection. They count only on an error the + * driver built (see {@link isDriverConnectionError}) or under a database query error, never on + * an arbitrary client error that happens to reuse the name. + */ +const DRIVER_CONNECTION_CODES = new Set([ + 'CONNECTION_CLOSED', + 'CONNECTION_DESTROYED', + 'CONNECTION_ENDED', + 'CONNECT_TIMEOUT', +]) + +/** + * Socket and name-resolution failures any client can raise; they count only when a database query + * carried them. + */ +const SOCKET_CONNECTION_CODES = new Set([ + 'ECONNRESET', + 'EPIPE', + 'ETIMEDOUT', + 'ECONNREFUSED', + 'EHOSTUNREACH', + 'ENOTFOUND', + 'EAI_AGAIN', + 'ENETDOWN', + 'ENETRESET', + 'ENETUNREACH', +]) + +const CONNECTION_EXCEPTION_SQLSTATE = /^08[0-9A-Z]{3}$/ + +/** + * Classifies a failure by the SQLSTATE or driver code in its `cause` chain. + * + * `57014` is both a statement timeout and an explicit cancellation, and only the message tells + * them apart: an explicit cancellation was asked for, so it is `permanent`. A socket error such as + * `ECONNRESET` is a database connection failure only when a database query error is in the chain + * (see {@link isDatabaseQueryError}), and a postgres.js connection code only on an error the driver + * built (see {@link isDriverConnectionError}) or under a query error; a file download or provider + * call raising the same code is not the database's to retry. + */ +export function classifyDatabaseFailure(error: unknown): DatabaseFailureClass { + const code = getPostgresErrorCode(error) + if (!code) return 'permanent' + if (code === '57014') { + return getPostgresCancellationReason(error) === 'statement_timeout' ? 'capacity' : 'permanent' + } + if (CAPACITY_CODES.has(code)) return 'capacity' + if (CONFLICT_CODES.has(code)) return 'conflict' + if (CONNECTION_EXCEPTION_SQLSTATE.test(code) || SERVER_UNAVAILABLE_SQLSTATES.has(code)) { + return 'connection' + } + if (DRIVER_CONNECTION_CODES.has(code)) { + return isDriverConnectionError(findCodedLink(error, code)) || carriesDatabaseQuery(error) + ? 'connection' + : 'permanent' + } + if (SOCKET_CONNECTION_CODES.has(code) && carriesDatabaseQuery(error)) return 'connection' + return 'permanent' +} + +/** The transient class of a database failure, or `undefined` when it is not one. */ +export function getTransientDatabaseFailure( + error: unknown +): TransientDatabaseFailureClass | undefined { + const failureClass = classifyDatabaseFailure(error) + return failureClass === 'permanent' ? undefined : failureClass +} + +/** + * Whether a link is one of the two errors a failed database query produces: + * + * - Drizzle's `DrizzleQueryError`, which sets no `name` of its own, so it is matched by shape: the + * SQL in `query`, bound values in a `params` array, and a message starting `Failed query: `. + * - A postgres.js error for a query it had taken on, onto which the driver defines `query`, + * `parameters`, `args`, and `types` as own properties. They are present even when the query + * never reached the server: a refused connection carries all four with `query` undefined. + * + * A `query` property alone is not enough: an HTTP or GraphQL client error carrying its own `query` + * would otherwise exempt a source failure from the connector breaker. + */ +function isDatabaseQueryError(value: Error): boolean { + const candidate = value as Error & { query?: unknown; params?: unknown } + if ( + typeof candidate.query === 'string' && + Array.isArray(candidate.params) && + candidate.message.startsWith('Failed query: ') + ) { + return true + } + return DRIVER_QUERY_PROPERTIES.every((property) => Object.hasOwn(value, property)) +} + +const DRIVER_QUERY_PROPERTIES = ['query', 'parameters', 'args', 'types'] as const + +/** + * Whether a link is a connection error postgres.js built itself. The driver makes each one the + * same way: `code` and `errno` both set to the code, a message `write `, + * and the target in `address`. A transaction that loses its connection is rejected with such an + * error directly, with no query attached and no Drizzle wrapper, so the query shapes above cannot + * be required of it. + */ +function isDriverConnectionError(value: unknown): boolean { + if (!(value instanceof Error)) return false + const candidate = value as Error & { code?: unknown; errno?: unknown } + return ( + typeof candidate.code === 'string' && + candidate.errno === candidate.code && + candidate.message.startsWith(`write ${candidate.code} `) && + Object.hasOwn(candidate, 'address') + ) +} + +/** The first link in the `cause` chain whose `code` is `code`, the one the classification read. */ +function findCodedLink(error: unknown, code: string): unknown { + const seen = new Set() + let current: unknown = error + while (current instanceof Error && !seen.has(current) && seen.size < 10) { + seen.add(current) + if ((current as Error & { code?: unknown }).code === code) return current + current = current.cause + } + return undefined +} + +function carriesDatabaseQuery(error: unknown): boolean { + const seen = new Set() + let current: unknown = error + while (current instanceof Error && !seen.has(current) && seen.size < 10) { + seen.add(current) + if (isDatabaseQueryError(current)) return true + current = current.cause + } + return false +} + /** * Returns the name of the PostgreSQL constraint that triggered the error (e.g. the unique index * name on a `23505`), when present on a thrown value. Mirrors the field populated by the