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

The sweep treats failed as terminal — "the run that produced it has ended" — and
that was true when processing ran inline. Since dispatch became asynchronous the
worker writes failed and then rethrows, and the processing task retries up to
three times, so failed is a resting state between attempts rather than a final
one. The sweep gave it no grace at all, so it reclaimed a document mid-retry
chain, deleted its embeddings, and re-dispatched with a fresh pass id: two live
runs, two indexing passes, two bills, through the one state deliberately left
unguarded.

Aged from processingCompletedAt, which every failure write stamps, so it reads
exactly when the last attempt ended rather than when the document was dispatched.

Sized at the queue grace rather than a run-duration bound. What is being waited
on between attempts is a re-queue behind the same global concurrency limit, not
another run, so on a large backlog the next attempt starts hours after the
previous one ended and a run-duration bound would still reclaim live work.

The grace itself is now an expression rather than a constant — corpus over queue
concurrency, times occupancy, times a contention factor — with each input
documented where it can be re-measured. The previous 240 was derived from a
2,600-document corpus and sat below the drain time of the 7,730-document
connector it was written for. Rounding is up: too small silently re-bills live
work, too large only delays recovery of documents nothing is processing.

Also withdraws the queue stamp when every dispatch in a batch fails, so a
provably-undispatched document keeps next-sync recovery instead of waiting out a
grace it never earned. Scoped to the batch's own ids, to rows still pending, and
by compare-and-set on the exact stamp that call wrote, so a concurrent batch that
re-stamped a document keeps its grace. Best-effort by design, while the stamp
write still throws: a failed stamp means the grace cannot be promised and
dispatching anyway is the unsafe direction, whereas a failed withdrawal only
delays recovery and must not mask the dispatch error underneath it.

Author: waleedlatif1

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`)
   }
 }