fix(knowledge): keep the deferred-retry watchdog alive through database failures

A transient database failure while checking postpones the check without spending an attempt, paced by how overdue it is and capped at the recheck interval, until a terminal bound past the recovery window; other errors, and any error past that bound, spend attempts as before.

Author: waleedlatif1

apps/sim/lib/knowledge/documents/deferred-retry-check.test.ts

@@ -25,6 +25,7 @@ vi.mock('@/lib/knowledge/documents/processing-recovery-queue', () => ({
 
 import {
   checkDeferredDocumentRetry,
+  DEFERRED_RETRY_CHECK_TERMINAL_MS,
   DEFERRED_RETRY_LOST_ERROR,
   DEFERRED_RETRY_RECHECK_MS,
 } from '@/lib/knowledge/documents/deferred-retry-check'
@@ -175,6 +176,58 @@ describe('checkDeferredDocumentRetry', () => {
     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'

apps/sim/lib/knowledge/documents/deferred-retry-check.ts

@@ -2,7 +2,9 @@ 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,
@@ -23,6 +25,17 @@ 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.'
 
@@ -78,13 +91,43 @@ function isSameDeferral(
  * 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<unknown> = async (
   rawPayload,
   context
 ): Promise<DeferredOutboxHandlerResult | undefined> => {
   const payload = parsePayload(rawPayload)
-  context.signal.throwIfAborted()
+  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<DeferredOutboxHandlerResult | undefined> {
+  signal.throwIfAborted()
   const [row] = await db
     .select({
       ...processingSnapshotColumns,
@@ -109,7 +152,7 @@ export const checkDeferredDocumentRetry: OutboxHandler<unknown> = async (
     return deferOutboxHandler('Deferred retry is not overdue yet', dueAt - now, false)
   }
   if (now < deferredUntil + RECOVERY_WINDOW_MS) {
-    const { abandoned } = await inspectDocumentProcessingLiveness([row], context.signal)
+    const { abandoned } = await inspectDocumentProcessingLiveness([row], signal)
     if (abandoned.length === 0) {
       return deferOutboxHandler(
         'Deferred retry may still be running',
@@ -118,7 +161,7 @@ export const checkDeferredDocumentRetry: OutboxHandler<unknown> = async (
       )
     }
   }
-  context.signal.throwIfAborted()
+  signal.throwIfAborted()
 
   const failed = await db
     .update(document)