fix(knowledge): stop the stuck-document sweep reclaiming still-queued documents (#6921)

* fix(knowledge): stop the stuck-document sweep reclaiming still-queued documents

The sweep gave 'processing' a staleness cutoff but gave 'pending' and 'failed'
none, and the only cross-run guard was uploadedAt < syncStartedAt, which scopes
within a sync but not across them.

That was harmless while document processing awaited inline: documents were
terminal by the time a sync ended. Since dispatch became asynchronous they sit
'pending' until a worker picks them up, so the next sync deleted their
embeddings, reset them, and re-dispatched while the original runs were still
executing — and each re-dispatch mints a fresh pass id, so it billed again.

Queued documents now get a grace period derived from queue drain time rather
than run duration: the processing queue's concurrency is 20 and global across
workspaces, so a 2,600-document corpus takes roughly two hours to drain. The
existing 45-minute threshold bounds a run, not a wait, and would still reclaim
live documents on any corpus over about 900.

No column records dispatch time, so the signal is processingStartedAt falling
back to uploadedAt. The sweep now stamps that column when it re-dispatches, and
the user-triggered retry stamps it instead of clearing it — without both, only a
document's first dispatch was protected and the sweep churned once per sync
forever on anything that kept waiting.

The grace narrows the duplicate-billing window but cannot close it: a
re-dispatch mints a new request id and the Trigger idempotency key is scoped per
dispatch by design. Closing it durably needs a document-scoped key or a
dispatch-generation column, recorded in TSDoc and deliberately not built here.

* fix(knowledge): record dispatch time in its own column instead of overloading the start time

Two review findings on this PR traced to the same root: the queue grace was
reading processingStartedAt, a column that means something else.

Externally, a pending row carrying a dispatch timestamp reported a processing
start time for work that had not started. Internally, updateDocument sets a
document pending and refreshes uploadedAt while leaving the previous run's
processingStartedAt in place, and completion never clears it — so the sweep aged
a re-dispatched document from a leftover stamp rather than its actual dispatch,
and could reclaim it inside the grace window while the earlier queue entry was
still live. That reopens the duplicate-billing race this PR closes.

processing_queued_at is written on every re-dispatch and read by the sweep;
processingStartedAt goes back to meaning what its name says, so its external
contract is byte-identical to before this PR.

Not stamped on first dispatch: the row is created and dispatched inside the same
sync run, so uploadedAt is already accurate there and a guarded write per upload
would buy no behavior.

The internal document route returns a full table row through a passthrough
schema, so the new column would have shipped as an undeclared raw Date. Declared
and serialized explicitly instead.

* fix(knowledge): stamp the queue time in the dispatch funnel, not at each call site

Giving dispatch its own column was not enough. The connector content-update path
updates a row in place with pending status and a fresh uploadedAt while leaving
every prior-run processing column intact, so a document dispatched once already
carried a stale queue stamp, the refreshed uploadedAt was never consulted, and
the row aged from a dead run's timestamp — the same bug one level down.

"Written on dispatch and only on dispatch" is now structural rather than a
convention three call sites remember: the stamp lives in processDocumentsWithQueue,
which every one of the nine dispatch paths already goes through, so none can
forget it. The sweep's and the retry's own stamps are kept because they land
inside the same transaction as their reset, so a document keeps its grace even
if the dispatch that follows throws.

The funnel also clears processingStartedAt, guarded on the row still being
pending so it cannot disturb a worker's compare-and-set. That closes the
API-facing half at its source: a pending row can no longer report a start time
from any path, including the content-update path no serializer would have seen.

Author: waleedlatif1

apps/sim/lib/api/contracts/knowledge/documents.ts

@@ -267,6 +267,8 @@ export const documentDataSchema = z
     tokenCount: z.number(),
     characterCount: z.number(),
     processingStatus: z.enum(['pending', 'processing', 'completed', 'failed']),
+    /** When indexing was last dispatched to a worker, which precedes a worker starting it. */
+    processingQueuedAt: nullableWireDateSchema.optional(),
     processingStartedAt: nullableWireDateSchema.optional(),
     processingCompletedAt: nullableWireDateSchema.optional(),
     processingError: z.string().nullable().optional(),

apps/sim/lib/knowledge/api/internal-route.ts

@@ -60,6 +60,7 @@ function serializeNullableDate(date: Date | string | null): string | null {
 export function toInternalKnowledgeDocument<
   T extends {
     uploadedAt: Date | string
+    processingQueuedAt?: Date | string | null
     processingStartedAt?: Date | string | null
     processingCompletedAt?: Date | string | null
     date1?: Date | string | null
@@ -69,6 +70,7 @@ export function toInternalKnowledgeDocument<
   return documentDataSchema.parse({
     ...document,
     uploadedAt: serializeDate(document.uploadedAt),
+    processingQueuedAt: serializeNullableDate(document.processingQueuedAt ?? null),
     processingStartedAt: serializeNullableDate(document.processingStartedAt ?? null),
     processingCompletedAt: serializeNullableDate(document.processingCompletedAt ?? null),
     date1: serializeNullableDate(document.date1 ?? null),

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

@@ -7,6 +7,7 @@ import { beforeEach, describe, expect, it, vi } from 'vitest'
 import {
   classifySuspectListing,
   evaluateListingSafety,
+  isStuckDocumentSweepEligible,
   mergeHydratedDocument,
   type PreviousListingObservation,
 } from '@/lib/knowledge/connectors/sync-engine'
@@ -702,3 +703,125 @@ describe('mergeHydratedDocument', () => {
     expect(merged.sourceUrl).toBe('https://example.com/a')
   })
 })
+
+describe('isStuckDocumentSweepEligible', () => {
+  const now = new Date('2026-08-20T12:00:00.000Z')
+  const minutesBefore = (minutes: number) => new Date(now.getTime() - minutes * 60 * 1000)
+
+  const candidate = (
+    processingStatus: string,
+    overrides: {
+      processingQueuedAt?: Date | null
+      processingStartedAt?: Date | null
+      uploadedAt?: Date
+    } = {}
+  ) => ({
+    processingStatus,
+    processingQueuedAt: overrides.processingQueuedAt ?? null,
+    processingStartedAt: overrides.processingStartedAt ?? null,
+    uploadedAt: overrides.uploadedAt ?? minutesBefore(5),
+  })
+
+  it('leaves a document dispatched by the previous sync and still queued alone', () => {
+    expect(
+      isStuckDocumentSweepEligible(candidate('pending', { uploadedAt: minutesBefore(90) }), now)
+    ).toBe(false)
+  })
+
+  it('leaves a document the sweep itself re-dispatched alone while it waits', () => {
+    expect(
+      isStuckDocumentSweepEligible(
+        candidate('pending', {
+          processingQueuedAt: minutesBefore(90),
+          uploadedAt: minutesBefore(60 * 48),
+        }),
+        now
+      )
+    ).toBe(false)
+  })
+
+  it('reclaims a queued document once the grace period has passed', () => {
+    expect(
+      isStuckDocumentSweepEligible(candidate('pending', { uploadedAt: minutesBefore(241) }), now)
+    ).toBe(true)
+    expect(
+      isStuckDocumentSweepEligible(
+        candidate('pending', {
+          processingQueuedAt: minutesBefore(241),
+          uploadedAt: minutesBefore(60 * 48),
+        }),
+        now
+      )
+    ).toBe(true)
+  })
+
+  it('holds a queued document at the grace boundary', () => {
+    expect(
+      isStuckDocumentSweepEligible(candidate('pending', { uploadedAt: minutesBefore(240) }), now)
+    ).toBe(false)
+  })
+
+  it('reclaims a failed document with no grace', () => {
+    expect(isStuckDocumentSweepEligible(candidate('failed'), now)).toBe(true)
+    expect(
+      isStuckDocumentSweepEligible(
+        candidate('failed', { processingStartedAt: minutesBefore(1) }),
+        now
+      )
+    ).toBe(true)
+  })
+
+  it('reclaims a processing document only once its run is stale', () => {
+    expect(
+      isStuckDocumentSweepEligible(
+        candidate('processing', { processingStartedAt: minutesBefore(44) }),
+        now
+      )
+    ).toBe(false)
+    expect(
+      isStuckDocumentSweepEligible(
+        candidate('processing', { processingStartedAt: minutesBefore(46) }),
+        now
+      )
+    ).toBe(true)
+  })
+
+  it('reclaims a processing document with no start time', () => {
+    expect(isStuckDocumentSweepEligible(candidate('processing'), now)).toBe(true)
+  })
+
+  it('ignores a start time a worker left on a document that was requeued', () => {
+    expect(
+      isStuckDocumentSweepEligible(
+        candidate('pending', {
+          processingQueuedAt: minutesBefore(90),
+          processingStartedAt: minutesBefore(60 * 48),
+          uploadedAt: minutesBefore(60 * 72),
+        }),
+        now
+      )
+    ).toBe(false)
+  })
+
+  it('gives a document whose content was just updated the full grace period', () => {
+    expect(
+      isStuckDocumentSweepEligible(
+        candidate('pending', {
+          processingQueuedAt: null,
+          processingStartedAt: minutesBefore(60 * 48),
+          uploadedAt: minutesBefore(5),
+        }),
+        now
+      )
+    ).toBe(false)
+  })
+
+  it('never reclaims a completed document', () => {
+    expect(
+      isStuckDocumentSweepEligible(
+        candidate('completed', { uploadedAt: minutesBefore(60 * 48) }),
+        now
+      )
+    ).toBe(false)
+  })
+})

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

@@ -10,7 +10,7 @@ import { createLogger } from '@sim/logger'
 import { getErrorMessage, toError } from '@sim/utils/errors'
 import { generateId } from '@sim/utils/id'
 import { randomInt } from '@sim/utils/random'
-import { and, desc, eq, gt, inArray, isNotNull, isNull, lt, ne, or, sql } from 'drizzle-orm'
+import { and, desc, eq, gt, inArray, isNotNull, isNull, lt, ne, sql } from 'drizzle-orm'
 import { decryptApiKey } from '@/lib/api-key/crypto'
 import {
   assertBillingAttributionSnapshot,
@@ -58,9 +58,91 @@ const CONTENT_INFLIGHT_BUDGET_BYTES = 64 * 1024 * 1024
 const MAX_PAGES = 500
 const MAX_SAFE_TITLE_LENGTH = 200
 const STALE_PROCESSING_MINUTES = 45
+/**
+ * Grace period a document that is merely *queued* 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.
+ */
+const QUEUED_DISPATCH_GRACE_MINUTES = 240
 const RETRY_WINDOW_DAYS = 7
 const MAX_CONSECUTIVE_FAILURES = 10
 
+/** The processing state the stuck-document sweep decides on, one row at a time. */
+export interface StuckDocumentSweepCandidate {
+  processingStatus: string
+  processingQueuedAt: Date | null
+  processingStartedAt: Date | null
+  uploadedAt: Date
+}
+
+/**
+ * Decides whether the sweep may reclaim one document — delete its embeddings,
+ * reset it, and dispatch it again.
+ *
+ * Since document processing is dispatched to `knowledge-process-document`
+ * rather than awaited inline, a document sits at `pending` from dispatch until
+ * a worker claims it; `processing` is only written once a worker has actually
+ * started. Reclaiming a `pending` document therefore risks racing a run that is
+ * still queued, which both duplicates its work and bills a second indexing
+ * 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.
+ *
+ * `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.
+ *
+ * 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:
+ * a document queued for longer than the grace is still reclaimed while its
+ * original run waits, and Trigger.dev will not deduplicate the second dispatch
+ * because the idempotency key `processDocumentsWithQueue` uses is
+ * `doc-process-<documentId>-<requestId>` with a fresh `requestId` per dispatch
+ * — scoped per dispatch by design, so it blocks intra-dispatch retries and
+ * nothing else. Each duplicate run mints its own indexing pass and bills for
+ * it, which is the double-billing `451d2ccbde` closed for the inline path.
+ * Closing it durably needs state, not timing: a document-scoped idempotency
+ * key, or a dispatch-generation column the worker carries and checks before
+ * indexing, so a superseded run declines to bill. That is a larger change than
+ * this hotfix.
+ */
+export function isStuckDocumentSweepEligible(doc: StuckDocumentSweepCandidate, now: Date): boolean {
+  switch (doc.processingStatus) {
+    case 'failed':
+      return true
+    case 'pending': {
+      const queuedAt = doc.processingQueuedAt ?? doc.uploadedAt
+      return now.getTime() - queuedAt.getTime() > QUEUED_DISPATCH_GRACE_MINUTES * 60 * 1000
+    }
+    case 'processing': {
+      if (!doc.processingStartedAt) return true
+      return (
+        now.getTime() - doc.processingStartedAt.getTime() > STALE_PROCESSING_MINUTES * 60 * 1000
+      )
+    }
+    default:
+      return false
+  }
+}
+
 /** Sanitizes a document title for use in S3 storage keys. */
 function sanitizeStorageTitle(title: string): string {
   return title.replace(/[^a-zA-Z0-9.-]/g, '_').slice(0, MAX_SAFE_TITLE_LENGTH)
@@ -1368,45 +1450,47 @@ export async function executeSync(
       throw new Error(`Knowledge base ${connector.knowledgeBaseId} was deleted during sync`)
     }
 
-    // Retry stuck documents that failed, never started, or were abandoned mid-processing.
-    // Only retry docs uploaded BEFORE this sync — docs added in the current sync
-    // are still processing asynchronously and would cause a duplicate processing race.
-    // Documents stuck in 'processing' beyond STALE_PROCESSING_MINUTES are considered
-    // abandoned (e.g. the Trigger.dev task process exited before processing completed).
-    // Documents uploaded more than RETRY_WINDOW_DAYS ago are not retried.
-    const staleProcessingCutoff = new Date(Date.now() - STALE_PROCESSING_MINUTES * 60 * 1000)
-    const stuckDocs = await db
+    /**
+     * Reclaims documents this connector left unfinished: a terminated attempt, a
+     * dispatch that never produced a run, or a run abandoned mid-processing.
+     *
+     * The query narrows to this connector's non-terminal documents inside the
+     * `RETRY_WINDOW_DAYS` window and excludes anything created by this sync;
+     * {@link isStuckDocumentSweepEligible} then makes the per-document decision,
+     * so the age rules live in one place rather than being split between SQL and
+     * TypeScript. Skipped (oversized) documents are recorded as content-less
+     * `failed` rows with no storage key and can never be reprocessed, so they are
+     * excluded outright.
+     */
+    const sweepEvaluatedAt = new Date()
+    const sweepCandidates = await db
       .select({
         id: document.id,
         fileUrl: document.fileUrl,
         filename: document.filename,
         fileSize: document.fileSize,
         mimeType: document.mimeType,
+        processingStatus: document.processingStatus,
+        processingQueuedAt: document.processingQueuedAt,
+        processingStartedAt: document.processingStartedAt,
+        uploadedAt: document.uploadedAt,
       })
       .from(document)
       .where(
         and(
           eq(document.connectorId, connectorId),
-          or(
-            inArray(document.processingStatus, ['pending', 'failed']),
-            and(
-              eq(document.processingStatus, 'processing'),
-              or(
-                isNull(document.processingStartedAt),
-                lt(document.processingStartedAt, staleProcessingCutoff)
-              )
-            )
-          ),
+          inArray(document.processingStatus, ['pending', 'failed', 'processing']),
           lt(document.uploadedAt, syncStartedAt),
           gt(document.uploadedAt, retryCutoff),
           eq(document.userExcluded, false),
-          // Skipped (oversized) docs are recorded as content-less failed rows with no
-          // storage key; they cannot be reprocessed, so exclude them from retry.
           isNotNull(document.storageKey),
           isNull(document.archivedAt),
           isNull(document.deletedAt)
         )
       )
+    const stuckDocs = sweepCandidates.filter((doc) =>
+      isStuckDocumentSweepEligible(doc, sweepEvaluatedAt)
+    )
 
     if (stuckDocs.length > 0) {
       logger.info(`Retrying ${stuckDocs.length} stuck documents`, { connectorId })
@@ -1451,6 +1535,12 @@ export async function executeSync(
               .update(document)
               .set({
                 processingStatus: 'pending',
+                /**
+                 * Records when this re-dispatch was queued, so a later sweep can
+                 * tell a document still waiting for a worker from one whose
+                 * dispatch was lost. See {@link isStuckDocumentSweepEligible}.
+                 */
+                processingQueuedAt: sweepEvaluatedAt,
                 processingStartedAt: null,
                 processingCompletedAt: null,
                 processingError: null,