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.

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

@@ -710,9 +710,14 @@ describe('isStuckDocumentSweepEligible', () => {
 
   const candidate = (
     processingStatus: string,
-    overrides: { processingStartedAt?: Date | null; uploadedAt?: Date } = {}
+    overrides: {
+      processingQueuedAt?: Date | null
+      processingStartedAt?: Date | null
+      uploadedAt?: Date
+    } = {}
   ) => ({
     processingStatus,
+    processingQueuedAt: overrides.processingQueuedAt ?? null,
     processingStartedAt: overrides.processingStartedAt ?? null,
     uploadedAt: overrides.uploadedAt ?? minutesBefore(5),
   })
@@ -727,7 +732,7 @@ describe('isStuckDocumentSweepEligible', () => {
     expect(
       isStuckDocumentSweepEligible(
         candidate('pending', {
-          processingStartedAt: minutesBefore(90),
+          processingQueuedAt: minutesBefore(90),
           uploadedAt: minutesBefore(60 * 48),
         }),
         now
@@ -742,7 +747,7 @@ describe('isStuckDocumentSweepEligible', () => {
     expect(
       isStuckDocumentSweepEligible(
         candidate('pending', {
-          processingStartedAt: minutesBefore(241),
+          processingQueuedAt: minutesBefore(241),
           uploadedAt: minutesBefore(60 * 48),
         }),
         now
@@ -785,6 +790,19 @@ describe('isStuckDocumentSweepEligible', () => {
     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('never reclaims a completed document', () => {
     expect(
       isStuckDocumentSweepEligible(

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

@@ -82,6 +82,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
 }
@@ -98,16 +99,12 @@ export interface StuckDocumentSweepCandidate {
  * pass, so queued documents get {@link QUEUED_DISPATCH_GRACE_MINUTES} before
  * they are considered lost.
  *
- * No column records dispatch time. `uploadedAt` is the row's creation time,
- * which for a document dispatched by the sync that created it is within that
- * sync's own runtime — an over-estimate of the wait bounded by the one-hour
- * sync ceiling, and the best available proxy. A document the sweep itself
- * re-dispatched has no such proxy at all, its `uploadedAt` being arbitrarily
- * old, so the sweep stamps `processingStartedAt` with the re-dispatch time and
- * this reads it back. That is an overload of the column, but a safe one: every
- * other reader of `processingStartedAt` gates on `processingStatus` being
- * `processing` first, and a worker overwrites the stamp with its own the moment
- * it starts.
+ * 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
@@ -132,7 +129,7 @@ export function isStuckDocumentSweepEligible(doc: StuckDocumentSweepCandidate, n
     case 'failed':
       return true
     case 'pending': {
-      const queuedAt = doc.processingStartedAt ?? doc.uploadedAt
+      const queuedAt = doc.processingQueuedAt ?? doc.uploadedAt
       return now.getTime() - queuedAt.getTime() > QUEUED_DISPATCH_GRACE_MINUTES * 60 * 1000
     }
     case 'processing': {
@@ -1474,6 +1471,7 @@ export async function executeSync(
         fileSize: document.fileSize,
         mimeType: document.mimeType,
         processingStatus: document.processingStatus,
+        processingQueuedAt: document.processingQueuedAt,
         processingStartedAt: document.processingStartedAt,
         uploadedAt: document.uploadedAt,
       })
@@ -1542,7 +1540,8 @@ export async function executeSync(
                  * tell a document still waiting for a worker from one whose
                  * dispatch was lost. See {@link isStuckDocumentSweepEligible}.
                  */
-                processingStartedAt: sweepEvaluatedAt,
+                processingQueuedAt: sweepEvaluatedAt,
+                processingStartedAt: null,
                 processingCompletedAt: null,
                 processingError: null,
                 chunkCount: 0,

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

@@ -42,18 +42,24 @@ describe('retryDocumentProcessing requeue stamp', () => {
     resetDbChainMock()
   })
 
-  it('stamps the requeue time instead of clearing it', async () => {
+  it('stamps the requeue time on the dispatch column', async () => {
     const before = Date.now()
     const values = await captureRequeueValues()
     const after = Date.now()
 
-    expect(values.processingStartedAt).toBeInstanceOf(Date)
-    const stamp = values.processingStartedAt as Date
+    expect(values.processingQueuedAt).toBeInstanceOf(Date)
+    const stamp = values.processingQueuedAt as Date
     expect(stamp.getTime()).toBeGreaterThanOrEqual(before)
     expect(stamp.getTime()).toBeLessThanOrEqual(after)
     expect(values.processingCompletedAt).toBeNull()
   })
 
+  it('leaves processingStartedAt null so the API reports no start time', async () => {
+    const values = await captureRequeueValues()
+
+    expect(values.processingStartedAt).toBeNull()
+  })
+
   it('leaves the requeued document outside the reach of the next connector sync', async () => {
     const values = await captureRequeueValues()
     const uploadedAt = new Date(Date.now() - 30 * 24 * 60 * 60 * 1000)
@@ -63,6 +69,7 @@ describe('retryDocumentProcessing requeue stamp', () => {
       isStuckDocumentSweepEligible(
         {
           processingStatus: values.processingStatus as string,
+          processingQueuedAt: values.processingQueuedAt as Date | null,
           processingStartedAt: values.processingStartedAt as Date | null,
           uploadedAt,
         },
@@ -72,7 +79,12 @@ describe('retryDocumentProcessing requeue stamp', () => {
 
     expect(
       isStuckDocumentSweepEligible(
-        { processingStatus: 'pending', processingStartedAt: null, uploadedAt },
+        {
+          processingStatus: 'pending',
+          processingQueuedAt: null,
+          processingStartedAt: null,
+          uploadedAt,
+        },
         sweptAt
       )
     ).toBe(true)

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

@@ -2498,17 +2498,12 @@ export async function retryDocumentProcessing(
    *
    * The document sits at `pending` until a worker claims it, and for a
    * connector-owned document the connector sweep
-   * (`isStuckDocumentSweepEligible`) reads `processingStartedAt ?? uploadedAt`
-   * to judge how long it has been queued. Clearing the column would fall the
-   * sweep back on `uploadedAt`, which for a document synced days ago is
-   * arbitrarily old — so the next sync would reclaim the document out from
+   * (`isStuckDocumentSweepEligible`) measures queue wait from
+   * `processingQueuedAt`, falling back to `uploadedAt`. Leaving it unset would
+   * fall the sweep back on `uploadedAt`, which for a document synced days ago
+   * is arbitrarily old — so the next sync would reclaim the document out from
    * under this very retry, duplicating its work and billing a second indexing
-   * pass. Stamping the requeue time puts it back inside the grace period.
-   *
-   * Safe against the processing claim: `processDocumentAsync` writes its own
-   * `processingStartedAt` unconditionally when it starts and compares every
-   * later write against that value, so it never requires the column to be null
-   * to claim a document.
+   * pass.
    */
   const requeuedAt = new Date()
   await db.transaction(async (tx) => {
@@ -2518,7 +2513,8 @@ export async function retryDocumentProcessing(
       .update(document)
       .set({
         processingStatus: 'pending',
-        processingStartedAt: requeuedAt,
+        processingQueuedAt: requeuedAt,
+        processingStartedAt: null,
         processingCompletedAt: null,
         processingError: null,
         chunkCount: 0,

packages/db/migrations/0297_eminent_sinister_six.sql

@@ -0,0 +1 @@
+ALTER TABLE "document" ADD COLUMN "processing_queued_at" timestamp;