fix(knowledge): stop recovery from re-dispatching documents whose runs are still queued

Author: waleedlatif1

apps/sim/lib/knowledge/__integration__/connector-lifecycle-locks.integration.ts

@@ -259,4 +259,37 @@ describe('source lifecycle KB guards', () => {
       .where(eq(document.id, retryDocumentId))
     expect(unchanged.status).toBe('failed')
   })
+
+  it('moves an expired queued generation charge to its replacement instead of adding one', async () => {
+    const stampedAt = new Date(Date.now() - 24 * 60 * 60_000)
+    await db
+      .update(document)
+      .set({
+        processingStatus: 'pending',
+        processingAttempts: 2,
+        processingQueuedAt: stampedAt,
+        processingQueueToken: 'expired-generation',
+        processingCompletedAt: null,
+      })
+      .where(eq(document.id, retryDocumentId))
+    await run('recover')
+    const [row] = await db
+      .select({
+        status: document.processingStatus,
+        attempts: document.processingAttempts,
+        token: document.processingQueueToken,
+      })
+      .from(document)
+      .where(eq(document.id, retryDocumentId))
+    expect(row).toEqual({ status: 'pending', attempts: 1, token: null })
+  })
+
+  it('keeps the charge of an attempt that reached a worker', async () => {
+    await run('recover')
+    const [row] = await db
+      .select({ attempts: document.processingAttempts })
+      .from(document)
+      .where(eq(document.id, retryDocumentId))
+    expect(row.attempts).toBe(1)
+  })
 })

apps/sim/lib/knowledge/__integration__/stored-document-recovery.integration.ts

@@ -244,6 +244,49 @@ describe('independent recovery of retained connector documents', () => {
     }
   })
 
+  it('replaces expired queued generations without spending the budget, then indexes once', async () => {
+    const ids = await seed()
+    const file = await failedFile(ids)
+    const expire = (token: string) =>
+      db
+        .update(document)
+        .set({
+          processingStatus: 'pending',
+          processingCompletedAt: null,
+          processingError: null,
+          processingQueuedAt: old(),
+          processingQueueToken: token,
+        })
+        .where(eq(document.id, file.documentId))
+    let generation = 'old-fixture-generation'
+    await expire(generation)
+    for (let cycle = 0; cycle <= MAX_PROCESSING_ATTEMPTS; cycle++) {
+      expect(await recoverKnowledgeDocumentProcessing()).toBe(1)
+      const [row] = await db.select().from(document).where(eq(document.id, file.documentId))
+      expect(row.processingAttempts).toBe(1)
+      expect(row.processingQueueToken).not.toBe(generation)
+      generation = row.processingQueueToken!
+      await expire(generation)
+    }
+    await db
+      .update(document)
+      .set({ processingQueuedAt: new Date() })
+      .where(eq(document.id, file.documentId))
+    expect(await recoverKnowledgeDocumentProcessing()).toBe(0)
+
+    const events = await eventsFor(ids)
+    expect(events).toHaveLength(MAX_PROCESSING_ATTEMPTS + 1)
+    const before = fixture.embeddingCalls
+    for (const event of events) {
+      expect(
+        await outbox.processOutboxEventById(event.id, knowledgeDocumentProcessingOutboxHandlers)
+      ).toBe('completed')
+    }
+    expect(fixture.embeddingCalls).toBe(before + 1)
+    const [indexed] = await db.select().from(document).where(eq(document.id, file.documentId))
+    expect(indexed.processingStatus, indexed.processingError ?? undefined).toBe('completed')
+  })
+
   it('keeps permanent, excluded, paused, fresh, deferred, and expired work out of automatic recovery', async () => {
     const ids = await seed()
     const file = await failedFile(ids)

apps/sim/lib/knowledge/connectors/sync-content-pass.test.ts

@@ -1,6 +1,8 @@
 /** @vitest-environment node */
 import {
   dbChainMockFns,
+  hasMockCondition,
+  type MockCondition,
   queueTableRows,
   resetDbChainMock as resetDatabaseMock,
   schemaMock,
@@ -643,6 +645,28 @@ describe('content pass checkpoint intent', () => {
   )
 })
 
+describe('resurrecting verified listed documents', () => {
+  it('only writes documents that are actually tombstoned', async () => {
+    sourceBody = { value: '<p>Current content</p>' }
+    await runPass({ access: 'admin' })
+    const resurrect = dbChainMockFns.set.mock.invocationCallOrder.find((_order, index) => {
+      const [values] = dbChainMockFns.set.mock.calls[index]
+      return Object.keys(values).length === 1 && values.deletedAt === null
+    })
+    expect(resurrect).toBeDefined()
+    const whereIndex = dbChainMockFns.where.mock.invocationCallOrder.findIndex(
+      (order) => order > resurrect!
+    )
+    expect(
+      hasMockCondition(
+        dbChainMockFns.where.mock.calls[whereIndex][0],
+        (node: MockCondition) =>
+          node.type === 'isNotNull' && node.column === schemaMock.document.deletedAt
+      )
+    ).toBe(true)
+  })
+})
+
 describe('permission refresh through the shared content pass', () => {
   const current: StoredPage = {
     ...EXISTING,

apps/sim/lib/knowledge/connectors/sync-content-pass.ts

@@ -290,6 +290,7 @@ export async function runConnectorContentPass(input: ContentPassInput) {
               and(
                 eq(document.connectorId, input.connectorId),
                 inArray(document.externalId, verified.slice(offset, offset + 500)),
+                isNotNull(document.deletedAt),
                 isNotNull(document.contentHash),
                 isNull(document.archivedAt)
               )

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

@@ -27,7 +27,10 @@ import {
   persistSkippedRetryHashes,
   updateDocument,
 } from '@/lib/knowledge/connectors/sync-persistence'
-import { documentProcessingRecoveryCondition } from '@/lib/knowledge/documents/processing-recovery-policy'
+import {
+  documentProcessingRecoveryCondition,
+  releaseUnclaimedDispatchAttempt,
+} from '@/lib/knowledge/documents/processing-recovery-policy'
 import { DOCUMENT_PROCESSING_STALE_THRESHOLD_MS } from '@/lib/knowledge/documents/processing-timeouts.server'
 import type { DocumentData } from '@/lib/knowledge/documents/service'
 import { isTriggerAvailable, processDocumentsWithQueue } from '@/lib/knowledge/documents/service'
@@ -1446,6 +1449,7 @@ export async function sweepStuckDocuments(input: SweepStuckDocumentsInput): Prom
             processingDeferredUntil: null,
             processingCompletedAt: null,
             processingError: null,
+            processingAttempts: releaseUnclaimedDispatchAttempt(sweepEvaluatedAt),
             chunkCount: 0,
             tokenCount: 0,
             characterCount: 0,

apps/sim/lib/knowledge/documents/processing-continuation-dispatch.test.ts

@@ -19,6 +19,7 @@ import type {
   DocumentProcessingLane,
   DocumentProcessingPayload,
 } from '@/lib/knowledge/documents/processing-payload'
+import { QUEUED_DISPATCH_GRACE_MS } from '@/lib/knowledge/documents/types'
 
 const BILLING_ATTRIBUTION = {
   actorUserId: 'user-1',
@@ -73,4 +74,20 @@ describe('dispatchDocumentProcessingContinuation', () => {
       concurrencyKey: 'organization:org-1',
     })
   })
+
+  /** Trigger.dev starts a delayed run's TTL when the delay ends; the deadline follows the stamp. */
+  it('expires a continuation unstarted before recovery may replace it', async () => {
+    const deferredUntil = new Date(Date.now() + 15 * 60_000)
+    await dispatchDocumentProcessingContinuation(
+      { ...payload('backfill'), processingQueuedAt: deferredUntil.toISOString() },
+      deferredUntil,
+      'continuation-key',
+      true
+    )
+
+    const { ttl } = mockTrigger.mock.calls[0][2]
+    expect(mockTrigger.mock.calls[0][2]).toMatchObject({ delay: deferredUntil })
+    expect(ttl).toBeGreaterThan(0)
+    expect(ttl * 1000).toBeLessThan(QUEUED_DISPATCH_GRACE_MS)
+  })
 })

apps/sim/lib/knowledge/documents/processing-continuation-dispatch.ts

@@ -5,7 +5,7 @@ import { resolveTriggerRegion } from '@/lib/core/async-jobs/region'
 import { env } from '@/lib/core/config/env'
 import { isTriggerDevEnabled } from '@/lib/core/config/env-flags'
 import { isInsideTriggerRun } from '@/lib/core/config/trigger-runtime'
-import { documentProcessingQueueOptions } from '@/lib/knowledge/documents/processing-lane'
+import { documentProcessingRunOptions } from '@/lib/knowledge/documents/processing-lane'
 import type { DocumentProcessingPayload } from '@/lib/knowledge/documents/processing-payload'
 
 export interface DocumentProcessingContinuation {
@@ -37,7 +37,7 @@ export async function dispatchDocumentProcessingContinuation(
        * deferred on quota resume as interactive work and escape the tenant's
        * bulk ceiling — the retry path would become the way around the limit.
        */
-      ...documentProcessingQueueOptions(payload),
+      ...documentProcessingRunOptions(payload, deferredUntil),
       region,
     })
     return

apps/sim/lib/knowledge/documents/processing-lane.test.ts

@@ -1,14 +1,19 @@
 /**
  * @vitest-environment node
  */
-import { describe, expect, it } from 'vitest'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 import {
   BACKFILL_PROCESSING_QUEUE_NAME,
   documentProcessingQueueOptions,
+  documentProcessingRunExpiry,
   INTERACTIVE_PROCESSING_QUEUE_NAME,
 } from '@/lib/knowledge/documents/processing-lane'
 import type { DocumentProcessingPayload } from '@/lib/knowledge/documents/processing-payload'
 import { resolveDocumentProcessingLane } from '@/lib/knowledge/documents/processing-payload'
+import {
+  QUEUED_DISPATCH_GRACE_MS,
+  QUEUED_DISPATCH_START_DEADLINE_MS,
+} from '@/lib/knowledge/documents/types'
 
 const BILLING_ATTRIBUTION = {
   actorUserId: 'user-1',
@@ -130,3 +135,51 @@ describe('documentProcessingQueueOptions', () => {
     )
   })
 })
+
+describe('documentProcessingRunExpiry', () => {
+  const NOW = new Date('2026-09-18T12:00:00.000Z')
+
+  beforeEach(() => {
+    vi.useFakeTimers()
+    vi.setSystemTime(NOW)
+  })
+  afterEach(() => {
+    vi.useRealTimers()
+  })
+
+  it('expires an undelayed run at the start deadline, before recovery may replace it', () => {
+    expect(QUEUED_DISPATCH_START_DEADLINE_MS).toBeLessThan(QUEUED_DISPATCH_GRACE_MS)
+    expect(documentProcessingRunExpiry({ processingQueuedAt: NOW.toISOString() })).toEqual({
+      ttl: QUEUED_DISPATCH_START_DEADLINE_MS / 1000,
+    })
+  })
+
+  it('gives a late relay only the time its generation has left', () => {
+    const stampedAt = new Date(NOW.getTime() - 60 * 60_000)
+    expect(documentProcessingRunExpiry({ processingQueuedAt: stampedAt.toISOString() })).toEqual({
+      ttl: (QUEUED_DISPATCH_START_DEADLINE_MS - 60 * 60_000) / 1000,
+    })
+  })
+
+  it('expires a generation already past its deadline at the minimum', () => {
+    const stampedAt = new Date(NOW.getTime() - QUEUED_DISPATCH_GRACE_MS)
+    expect(documentProcessingRunExpiry({ processingQueuedAt: stampedAt.toISOString() })).toEqual({
+      ttl: 1,
+    })
+  })
+
+  /** Trigger.dev starts a delayed run's TTL when its delay ends, not when it is triggered. */
+  it('counts a delayed run from the end of its delay', () => {
+    const deferredUntil = new Date(NOW.getTime() + 30 * 60_000)
+    expect(
+      documentProcessingRunExpiry(
+        { processingQueuedAt: deferredUntil.toISOString() },
+        deferredUntil
+      )
+    ).toEqual({ ttl: QUEUED_DISPATCH_START_DEADLINE_MS / 1000 })
+  })
+
+  it('leaves a payload without a queue stamp on the queue default', () => {
+    expect(documentProcessingRunExpiry({})).toEqual({})
+  })
+})

apps/sim/lib/knowledge/documents/processing-lane.ts

@@ -3,6 +3,7 @@ import type {
   DocumentProcessingBillingContext,
   DocumentProcessingPayload,
 } from '@/lib/knowledge/documents/processing-payload'
+import { QUEUED_DISPATCH_START_DEADLINE_MS } from '@/lib/knowledge/documents/types'
 
 /**
  * Queue backing the interactive lane.
@@ -55,3 +56,35 @@ export function documentProcessingQueueOptions(payload: DocumentProcessingPayloa
     concurrencyKey: documentProcessingTenantKey(payload),
   }
 }
+
+/**
+ * Trigger.dev `ttl` expiring a run unstarted at its generation's start deadline
+ * (queue stamp + {@link QUEUED_DISPATCH_START_DEADLINE_MS}). Trigger.dev counts `ttl`
+ * from enqueue, which for a delayed run is `notBefore`. A generation already past its
+ * deadline gets the one-second minimum; one without a stamp keeps the queue default.
+ */
+export function documentProcessingRunExpiry(
+  payload: Pick<DocumentProcessingPayload, 'processingQueuedAt'>,
+  notBefore?: Date
+): { ttl?: number } {
+  if (!payload.processingQueuedAt) return {}
+  const deadline =
+    new Date(payload.processingQueuedAt).getTime() + QUEUED_DISPATCH_START_DEADLINE_MS
+  const enqueuedAt = Math.max(Date.now(), notBefore?.getTime() ?? 0)
+  return { ttl: Math.max(1, Math.floor((deadline - enqueuedAt) / 1000)) }
+}
+
+/**
+ * Every Trigger.dev option a `knowledge-process-document` dispatch needs: its lane's
+ * per-tenant queue and its start deadline. Dispatch sites use this rather than the
+ * parts so none can enqueue a run that outlives recovery's grace.
+ */
+export function documentProcessingRunOptions(
+  payload: DocumentProcessingPayload,
+  notBefore?: Date
+): { queue: string; concurrencyKey: string; ttl?: number } {
+  return {
+    ...documentProcessingQueueOptions(payload),
+    ...documentProcessingRunExpiry(payload, notBefore),
+  }
+}

apps/sim/lib/knowledge/documents/processing-queue.test.ts

@@ -368,6 +368,23 @@ describe('processDocumentsWithQueue dispatch backend', () => {
     })
   })
 
+  it('expires each queued run before recovery may replace its generation', async () => {
+    markInsideTriggerRun()
+
+    await processDocumentsWithQueue(
+      [DOCUMENT],
+      'knowledge-base-1',
+      {},
+      'request-1',
+      BILLING_ATTRIBUTION,
+      'backfill'
+    )
+
+    const [item] = mockBatchTrigger.mock.calls[0][1]
+    expect(item.options.ttl).toBeGreaterThan(0)
+    expect(item.options.ttl * 1000).toBeLessThan(QUEUED_DISPATCH_GRACE_MS)
+  })
+
   /**
    * The starvation this split exists to prevent: connector backfill must not be
    * able to occupy the queue a person's upload is admitted through.