fix(knowledge): give failed documents a grace while their retries are still scheduled

apps/sim/lib/knowledge/connectors/sync-engine.test.ts

@@ -713,26 +713,38 @@ describe('isStuckDocumentSweepEligible', () => {
     overrides: {
       processingQueuedAt?: Date | null
       processingStartedAt?: Date | null
+      processingCompletedAt?: Date | null
       uploadedAt?: Date
     } = {}
   ) => ({
     processingStatus,
     processingQueuedAt: overrides.processingQueuedAt ?? null,
     processingStartedAt: overrides.processingStartedAt ?? null,
+    processingCompletedAt: overrides.processingCompletedAt ?? null,
     uploadedAt: overrides.uploadedAt ?? minutesBefore(5),
   })
 
+  /**
+   * Pinned to the derivation in sync-engine (corpus 7,730 / concurrency 20 x
+   * 1 minute occupancy x 2 contention). A change to any input should fail here
+   * so it is re-checked deliberately rather than absorbed silently.
+   */
+  const GRACE_MINUTES = 773
+
   it('leaves a document dispatched by the previous sync and still queued alone', () => {
     expect(
-      isStuckDocumentSweepEligible(candidate('pending', { uploadedAt: minutesBefore(90) }), now)
+      isStuckDocumentSweepEligible(
+        candidate('pending', { uploadedAt: minutesBefore(GRACE_MINUTES - 1) }),
+        now
+      )
     ).toBe(false)
   })
 
   it('leaves a document the sweep itself re-dispatched alone while it waits', () => {
     expect(
       isStuckDocumentSweepEligible(
         candidate('pending', {
-          processingQueuedAt: minutesBefore(90),
+          processingQueuedAt: minutesBefore(GRACE_MINUTES - 1),
           uploadedAt: minutesBefore(60 * 48),
         }),
         now
@@ -742,12 +754,15 @@ describe('isStuckDocumentSweepEligible', () => {
 
   it('reclaims a queued document once the grace period has passed', () => {
     expect(
-      isStuckDocumentSweepEligible(candidate('pending', { uploadedAt: minutesBefore(241) }), now)
+      isStuckDocumentSweepEligible(
+        candidate('pending', { uploadedAt: minutesBefore(GRACE_MINUTES + 1) }),
+        now
+      )
     ).toBe(true)
     expect(
       isStuckDocumentSweepEligible(
         candidate('pending', {
-          processingQueuedAt: minutesBefore(241),
+          processingQueuedAt: minutesBefore(GRACE_MINUTES + 1),
           uploadedAt: minutesBefore(60 * 48),
         }),
         now
@@ -757,15 +772,63 @@ describe('isStuckDocumentSweepEligible', () => {
 
   it('holds a queued document at the grace boundary', () => {
     expect(
-      isStuckDocumentSweepEligible(candidate('pending', { uploadedAt: minutesBefore(240) }), now)
+      isStuckDocumentSweepEligible(
+        candidate('pending', { uploadedAt: minutesBefore(GRACE_MINUTES) }),
+        now
+      )
+    ).toBe(false)
+  })
+
+  it('leaves a failed document alone while its Trigger retries may still run', () => {
+    expect(
+      isStuckDocumentSweepEligible(
+        candidate('failed', { processingCompletedAt: minutesBefore(1) }),
+        now
+      )
     ).toBe(false)
   })
 
-  it('reclaims a failed document with no grace', () => {
-    expect(isStuckDocumentSweepEligible(candidate('failed'), now)).toBe(true)
+  it('ages a failed document from its last attempt, not from its dispatch', () => {
+    expect(
+      isStuckDocumentSweepEligible(
+        candidate('failed', {
+          processingQueuedAt: minutesBefore(60 * 48),
+          processingCompletedAt: minutesBefore(1),
+          uploadedAt: minutesBefore(60 * 72),
+        }),
+        now
+      )
+    ).toBe(false)
+  })
+
+  it('reclaims a failed document once no retry of it can still be live', () => {
+    expect(
+      isStuckDocumentSweepEligible(
+        candidate('failed', { processingCompletedAt: minutesBefore(GRACE_MINUTES + 1) }),
+        now
+      )
+    ).toBe(true)
+  })
+
+  it('holds a failed document at the grace boundary', () => {
+    expect(
+      isStuckDocumentSweepEligible(
+        candidate('failed', { processingCompletedAt: minutesBefore(GRACE_MINUTES) }),
+        now
+      )
+    ).toBe(false)
+  })
+
+  it('falls back to the dispatch stamp when a failed row never recorded completion', () => {
+    expect(
+      isStuckDocumentSweepEligible(
+        candidate('failed', { processingQueuedAt: minutesBefore(1), uploadedAt: minutesBefore(1) }),
+        now
+      )
+    ).toBe(false)
     expect(
       isStuckDocumentSweepEligible(
-        candidate('failed', { processingStartedAt: minutesBefore(1) }),
+        candidate('failed', { processingQueuedAt: minutesBefore(GRACE_MINUTES + 1) }),
         now
       )
     ).toBe(true)
@@ -794,7 +857,7 @@ describe('isStuckDocumentSweepEligible', () => {
     expect(
       isStuckDocumentSweepEligible(
         candidate('pending', {
-          processingQueuedAt: minutesBefore(90),
+          processingQueuedAt: minutesBefore(GRACE_MINUTES - 1),
           processingStartedAt: minutesBefore(60 * 48),
           uploadedAt: minutesBefore(60 * 72),
         }),

apps/sim/lib/knowledge/connectors/sync-engine.ts

@@ -58,24 +58,37 @@ const CONTENT_INFLIGHT_BUDGET_BYTES = 64 * 1024 * 1024
 const MAX_PAGES = 500
 const MAX_SAFE_TITLE_LENGTH = 200
 const STALE_PROCESSING_MINUTES = 45
+/** Largest connector corpus observed in production, which sets the queue drain to beat. */
+const LARGEST_OBSERVED_CORPUS_DOCUMENTS = 7_730
+/** `document-processing-queue`'s `concurrencyLimit` — global, shared by every workspace. */
+const PROCESSING_QUEUE_CONCURRENCY = 20
+/** Wall time a typical document occupies a queue slot, parse through embedding. */
+const TYPICAL_DOCUMENT_OCCUPANCY_MINUTES = 1
+/** Headroom for the queue being shared: another tenant's backlog cuts our share of it. */
+const QUEUE_CONTENTION_FACTOR = 2
 /**
- * Grace period a document that is merely *queued* gets before the stuck-document
- * sweep may reclaim it.
+ * Grace period a document waiting on the processing queue gets before the
+ * stuck-document sweep may reclaim it.
  *
- * `STALE_PROCESSING_MINUTES` bounds a run that has already begun — `maxDuration`
- * (10m) x `maxAttempts` (3) on `knowledge-process-document` is 30 minutes, so 45
- * covers it with headroom. Queue *wait* is a different quantity: it is
- * backlog / concurrency, not run duration. `document-processing-queue` has a
- * global `concurrencyLimit` of 20 shared by every workspace, and a single large
- * connector corpus is on the order of the 2,600-document site
- * `CONNECTOR_SYNC_MAX_DURATION_SECONDS` was raised for — 130 waves of 20, which
- * at roughly a minute of occupancy per document drains in about two hours, and
- * longer while other workspaces hold slots. Four hours is about twice that
- * drain, an order of magnitude above the one-hour sync ceiling, and still well
- * under the 1,440-minute default sync interval, so a default-configured
- * connector waits no longer for recovery than it already did.
+ * Derived rather than chosen, so the next person can re-run the arithmetic with
+ * their own numbers instead of trusting this one: the sweep must not reclaim a
+ * document that a full queue drain has simply not reached yet, so the grace is
+ * that drain — corpus / concurrency x per-document occupancy — times a
+ * contention factor for the queue being shared across workspaces. At the values
+ * above that is 7,730 / 20 x 1 x 2 = 773 minutes, just under thirteen hours.
+ *
+ * Each input is measurable and should be re-measured when it moves: corpus size
+ * from the largest connector in production, concurrency from
+ * `knowledge-process-document`'s queue config, occupancy from run durations.
+ * Note the failure mode is asymmetric — too small silently re-bills live work,
+ * too large only delays recovery of documents nothing is processing — so round
+ * up, never down.
  */
-const QUEUED_DISPATCH_GRACE_MINUTES = 240
+const QUEUED_DISPATCH_GRACE_MINUTES = Math.ceil(
+  (LARGEST_OBSERVED_CORPUS_DOCUMENTS / PROCESSING_QUEUE_CONCURRENCY) *
+    TYPICAL_DOCUMENT_OCCUPANCY_MINUTES *
+    QUEUE_CONTENTION_FACTOR
+)
 const RETRY_WINDOW_DAYS = 7
 const MAX_CONSECUTIVE_FAILURES = 10
 
@@ -84,6 +97,7 @@ export interface StuckDocumentSweepCandidate {
   processingStatus: string
   processingQueuedAt: Date | null
   processingStartedAt: Date | null
+  processingCompletedAt: Date | null
   uploadedAt: Date
 }
 
@@ -99,16 +113,28 @@ export interface StuckDocumentSweepCandidate {
  * pass, so queued documents get {@link QUEUED_DISPATCH_GRACE_MINUTES} before
  * they are considered lost.
  *
- * Queue wait is measured from `processingQueuedAt`, written by every path that
- * re-dispatches an existing document — this sweep and the user-facing retry.
- * It falls back to `uploadedAt` when NULL, which covers a document dispatched
- * by the sync that created it (`uploadedAt` then sits within that sync's own
- * runtime, an over-estimate bounded by the one-hour sync ceiling) and rows
- * written before the column existed.
+ * Queue wait is measured from `processingQueuedAt`, stamped by
+ * `processDocumentsWithQueue` — the funnel every dispatch passes through — so
+ * no caller can dispatch without recording when. It falls back to `uploadedAt`
+ * when NULL, which covers rows written before the column existed.
+ *
+ * `failed` is not a terminal state and gets the same grace. `processDocumentAsync`
+ * records the failure and then rethrows, so `knowledge-process-document` retries
+ * it up to `maxAttempts` (3): between attempts the row reads `failed` while a
+ * live run is scheduled to pick it up again. The gap between attempts is bounded
+ * by the queue, not by run duration — a retried run re-enters the same queue
+ * behind the same global concurrency limit — so `maxDuration` x `maxAttempts`
+ * (30 minutes) and `STALE_PROCESSING_MINUTES` are both far too short to be safe
+ * here: on the very backlog this grace exists for, the next attempt starts hours
+ * after the last one ended. `failed` is therefore aged from
+ * `processingCompletedAt`, the instant the last attempt ended, which every
+ * failure write stamps.
  *
- * `failed` gets no grace. It is a terminal state: the run that produced it has
- * ended, so re-dispatching cannot duplicate live work, and it is the state the
- * user-facing retry path also operates from.
+ * A document whose retries genuinely exhaust is still recovered: its final
+ * failure stops moving `processingCompletedAt`, so one grace later it becomes
+ * eligible and the next sync re-dispatches it. Recovery is delayed by the grace,
+ * never lost. The user-facing retry stays immediate — it writes `pending` and
+ * dispatches without consulting the sweep at all.
  *
  * This narrows the duplicate-dispatch window; it cannot close it. A grace
  * period is a timing guarantee, and no timing guarantee is a correctness one:
@@ -126,8 +152,13 @@ export interface StuckDocumentSweepCandidate {
  */
 export function isStuckDocumentSweepEligible(doc: StuckDocumentSweepCandidate, now: Date): boolean {
   switch (doc.processingStatus) {
-    case 'failed':
-      return true
+    case 'failed': {
+      const lastAttemptEndedAt =
+        doc.processingCompletedAt ?? doc.processingQueuedAt ?? doc.uploadedAt
+      return (
+        now.getTime() - lastAttemptEndedAt.getTime() > QUEUED_DISPATCH_GRACE_MINUTES * 60 * 1000
+      )
+    }
     case 'pending': {
       const queuedAt = doc.processingQueuedAt ?? doc.uploadedAt
       return now.getTime() - queuedAt.getTime() > QUEUED_DISPATCH_GRACE_MINUTES * 60 * 1000
@@ -1473,6 +1504,7 @@ export async function executeSync(
         processingStatus: document.processingStatus,
         processingQueuedAt: document.processingQueuedAt,
         processingStartedAt: document.processingStartedAt,
+        processingCompletedAt: document.processingCompletedAt,
         uploadedAt: document.uploadedAt,
       })
       .from(document)

apps/sim/lib/knowledge/documents/retry-processing-grace.test.ts

@@ -1,7 +1,7 @@
 /**
  * @vitest-environment node
  */
-import { dbChainMock, dbChainMockFns, resetDbChainMock } from '@sim/testing'
+import { dbChainMock, dbChainMockFns, flattenMockConditions, resetDbChainMock } from '@sim/testing'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 
 vi.mock('@sim/db', () => dbChainMock)
@@ -74,6 +74,7 @@ describe('retryDocumentProcessing requeue stamp', () => {
           processingStatus: values.processingStatus as string,
           processingQueuedAt: values.processingQueuedAt as Date | null,
           processingStartedAt: values.processingStartedAt as Date | null,
+          processingCompletedAt: values.processingCompletedAt as Date | null,
           uploadedAt,
         },
         sweptAt
@@ -86,6 +87,7 @@ describe('retryDocumentProcessing requeue stamp', () => {
           processingStatus: 'pending',
           processingQueuedAt: null,
           processingStartedAt: null,
+          processingCompletedAt: null,
           uploadedAt,
         },
         sweptAt
@@ -121,7 +123,7 @@ describe('processDocumentsWithQueue dispatch stamp', () => {
     const after = Date.now()
 
     const stampCall = dbChainMockFns.set.mock.calls.find(
-      (call) => (call[0] as Record<string, unknown> | undefined)?.processingQueuedAt !== undefined
+      (call) => (call[0] as Record<string, unknown> | undefined)?.processingQueuedAt instanceof Date
     )
     expect(stampCall).toBeDefined()
     const values = stampCall?.[0] as Record<string, unknown>
@@ -132,4 +134,32 @@ describe('processDocumentsWithQueue dispatch stamp', () => {
     expect(stamp.getTime()).toBeLessThanOrEqual(after)
     expect(values.processingStartedAt).toBeNull()
   })
+
+  it('withdraws the stamp when every dispatch fails', async () => {
+    await dispatch()
+
+    const withdrawal = dbChainMockFns.set.mock.calls.find(
+      (call) => (call[0] as Record<string, unknown> | undefined)?.processingQueuedAt === null
+    )
+    expect(withdrawal).toBeDefined()
+  })
+
+  it('scopes the withdrawal to this batch, to pending rows, and to its own stamp', async () => {
+    await dispatch()
+
+    const stampCall = dbChainMockFns.set.mock.calls.find(
+      (call) => (call[0] as Record<string, unknown> | undefined)?.processingQueuedAt instanceof Date
+    )
+    const stamp = (stampCall?.[0] as Record<string, unknown>).processingQueuedAt as Date
+
+    const scoped = dbChainMockFns.where.mock.calls.some((call) => {
+      const nodes = flattenMockConditions(call[0])
+      return (
+        nodes.some((node) => node.type === 'inArray' && Array.isArray(node.values)) &&
+        nodes.some((node) => node.type === 'eq' && node.right === 'pending') &&
+        nodes.some((node) => node.type === 'eq' && node.right === stamp)
+      )
+    })
+    expect(scoped).toBe(true)
+  })
 })

apps/sim/lib/knowledge/documents/service.ts

@@ -706,6 +706,36 @@ async function markDocumentsQueued(documentIds: string[], queuedAt: Date): Promi
     .where(and(inArray(document.id, documentIds), eq(document.processingStatus, 'pending')))
 }
 
+/**
+ * Withdraws a queue stamp whose dispatch provably never happened.
+ *
+ * {@link markDocumentsQueued} runs before dispatch on purpose — `batchTrigger`
+ * chunks, so a batch can half-succeed, and stamping afterwards would leave the
+ * runs that did start with no stamp and no grace. The cost of that ordering is
+ * that a batch where *every* dispatch failed still carries a fresh stamp, and
+ * recovery sweeps would honour a grace period the documents did not earn. Total
+ * failure is the one case where nothing was dispatched, so the stamp can be
+ * taken back and the next sweep is free to reclaim them immediately.
+ *
+ * Scoped three ways so it can only ever undo its own write: to the ids in this
+ * batch, to rows still `pending` (a worker that has since claimed one keeps its
+ * timestamps — see {@link markDocumentsQueued}), and to the exact stamp this
+ * call wrote, so a concurrent dispatch that has already re-stamped a document
+ * is left alone.
+ */
+async function clearDocumentsQueued(documentIds: string[], queuedAt: Date): Promise<void> {
+  await db
+    .update(document)
+    .set({ processingQueuedAt: null })
+    .where(
+      and(
+        inArray(document.id, documentIds),
+        eq(document.processingStatus, 'pending'),
+        eq(document.processingQueuedAt, queuedAt)
+      )
+    )
+}
+
 /**
  * Dispatches document processing jobs via Trigger.dev's `batchTrigger` when
  * available, or in-process otherwise. Throws only when every dispatch fails;
@@ -728,10 +758,9 @@ export async function processDocumentsWithQueue(
     buildJobPayload(doc, knowledgeBaseId, processingOptions, requestId, billingContext)
   )
 
-  await markDocumentsQueued(
-    createdDocuments.map((doc) => doc.documentId),
-    new Date()
-  )
+  const documentIds = createdDocuments.map((doc) => doc.documentId)
+  const queuedAt = new Date()
+  await markDocumentsQueued(documentIds, queuedAt)
 
   const useTrigger = isTriggerAvailable()
   logger.info(
@@ -748,6 +777,19 @@ export async function processDocumentsWithQueue(
   )
 
   if (dispatched === 0) {
+    /**
+     * Best-effort, unlike the stamp itself: failing to write the stamp means the
+     * grace cannot be promised and dispatching anyway is the unsafe direction, so
+     * that write throws. Failing to withdraw one only delays recovery by a grace
+     * period, so it must not mask the dispatch failure that is the real error.
+     */
+    try {
+      await clearDocumentsQueued(documentIds, queuedAt)
+    } catch (error) {
+      logger.warn(`[${requestId}] Failed to withdraw the queue stamp after a failed dispatch`, {
+        error: getErrorMessage(error),
+      })
+    }
     throw new Error(`All ${jobPayloads.length} document processing dispatches failed`)
   }
 }