fix(knowledge): give document processing per-tenant queue lanes (#7937)

Document processing ran through one Trigger.dev queue with a global
concurrency limit and no concurrency key, so a single connector backfill
could hold every slot and leave every other tenant's uploads queued
behind it for hours.

Dispatch now names a lane — interactive for work a person is waiting on,
backfill for connector-driven ingestion — and keys each run by the
entity that owns the knowledge base, so the limit applies per tenant
copy of the queue rather than fleet-wide.

- Key on the owning entity, not the workspace: an organization-scoped
  knowledge base carries a null workspace id and would otherwise share
  one bucket with every other such tenant. Owned scopes reuse
  resourceScopeKey so tenant identity keeps one spelling.
- Stamp the lane on the payload so quota and capacity continuations
  resume in the lane they were admitted against instead of promoting
  themselves out of the backfill ceiling.
- Parse an absent or unrecognized lane as backfill rather than throwing,
  so payloads written before the lanes existed and payloads stamped by a
  newer version mid-rollout cannot burn a run's retry budget.
- Keep interactive work on the pre-existing queue name; a queue the
  running worker has not registered parks its runs in PENDING_VERSION,
  and the app deploys separately from the worker.
- Stop absorbing a rejected backfill chunk in-process: those documents
  are reported failed and reclaimed by the stuck-document sweep instead
  of outlasting the connector lease they run under.

Author: waleedlatif1

apps/sim/app/api/v1/knowledge/[id]/documents/route.test.ts

@@ -245,7 +245,8 @@ describe('v1 knowledge document upload route', () => {
       'kb-1',
       {},
       'req-1',
-      SYSTEM_BILLING_ATTRIBUTION
+      SYSTEM_BILLING_ATTRIBUTION,
+      'interactive'
     )
   })
 })

apps/sim/background/knowledge-processing.test.ts

@@ -7,18 +7,24 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 const {
   mockAssertBillingAttributionSnapshot,
   mockProcessDocumentAsync,
+  mockQueue,
   mockResolveTriggerRegion,
   mockTask,
   mockTrigger,
 } = vi.hoisted(() => ({
   mockAssertBillingAttributionSnapshot: vi.fn(),
   mockProcessDocumentAsync: vi.fn(),
   mockResolveTriggerRegion: vi.fn(),
+  mockQueue: vi.fn((config) => config),
   mockTask: vi.fn((config) => config),
   mockTrigger: vi.fn(),
 }))
 
-vi.mock('@trigger.dev/sdk', () => ({ task: mockTask, tasks: { trigger: mockTrigger } }))
+vi.mock('@trigger.dev/sdk', () => ({
+  queue: mockQueue,
+  task: mockTask,
+  tasks: { trigger: mockTrigger },
+}))
 vi.mock('@/lib/core/async-jobs/region', () => ({ resolveTriggerRegion: mockResolveTriggerRegion }))
 vi.mock('@/lib/billing/core/billing-attribution', () => ({
   assertBillingAttributionSnapshot: mockAssertBillingAttributionSnapshot,

apps/sim/background/knowledge-processing.ts

@@ -1,5 +1,5 @@
 import { createLogger } from '@sim/logger'
-import { task } from '@trigger.dev/sdk'
+import { queue, task } from '@trigger.dev/sdk'
 import { env, envNumber } from '@/lib/core/config/env'
 import {
   BYOK_EMBEDDING_CREDENTIAL_REJECTION_MESSAGE,
@@ -13,6 +13,10 @@ import {
   isPermanentDocumentProcessingError,
   isUsageLimitDocumentProcessingError,
 } from '@/lib/knowledge/documents/document-processing-error'
+import {
+  BACKFILL_PROCESSING_QUEUE_NAME,
+  INTERACTIVE_PROCESSING_QUEUE_NAME,
+} from '@/lib/knowledge/documents/processing-lane'
 import {
   assertDocumentProcessingBillingContext,
   assertDocumentProcessingPayload,
@@ -205,6 +209,34 @@ export async function runDocumentProcessing(
   }
 }
 
+/**
+ * Both lanes are keyed by tenant at dispatch, so `concurrencyLimit` is the
+ * ceiling one tenant may hold in that lane, not a ceiling for the fleet. The
+ * shared bound is the Trigger.dev environment concurrency limit, which is where
+ * a global ceiling belongs; observed peak there is ~97 across every task.
+ *
+ * Both default to the limit the single shared queue carried, which is what
+ * keeps this split from ever draining slower than the queue it replaces: the
+ * busiest case it has to beat is one tenant alone, and one tenant alone still
+ * gets the same slots it used to get for backfill plus a separate allowance for
+ * work someone is waiting on. Any second tenant is pure gain, because under the
+ * shared queue it got whatever the first one left.
+ *
+ * Splitting the two into separate variables is for operating them, not for
+ * sizing them: backfill is the one to lower when the environment ceiling is the
+ * binding constraint, and lowering it must not slow down a person's upload.
+ */
+export const interactiveProcessingQueue = queue({
+  name: INTERACTIVE_PROCESSING_QUEUE_NAME,
+  concurrencyLimit: envNumber(env.KB_CONFIG_CONCURRENCY_LIMIT, 20),
+})
+
+/** Referenced by no dispatch site: named per trigger, declared here so the deploy registers it. */
+export const backfillProcessingQueue = queue({
+  name: BACKFILL_PROCESSING_QUEUE_NAME,
+  concurrencyLimit: envNumber(env.KB_CONFIG_BACKFILL_CONCURRENCY_LIMIT, 20),
+})
+
 export const processDocument = task({
   id: 'knowledge-process-document',
   maxDuration: envNumber(env.KB_CONFIG_MAX_DURATION, 600),
@@ -227,10 +259,12 @@ export const processDocument = task({
      */
     outOfMemory: { machine: 'large-2x' },
   },
-  queue: {
-    concurrencyLimit: envNumber(env.KB_CONFIG_CONCURRENCY_LIMIT, 20),
-    name: 'document-processing-queue',
-  },
+  /**
+   * The lane every dispatch names explicitly. Declared here as well so a
+   * trigger that somehow omits the option still lands on a registered queue
+   * rather than waiting in `PENDING_VERSION` for one that does not exist.
+   */
+  queue: interactiveProcessingQueue,
   run: (payload: DocumentProcessingPayload, { ctx }) =>
     runDocumentProcessing(payload, ctx.attempt.number),
 })

apps/sim/lib/core/config/env.ts

@@ -459,7 +459,8 @@ export const env = createEnv({
     KB_CONFIG_RETRY_FACTOR:                z.number().optional().default(2),       // Retry backoff factor
     KB_CONFIG_MIN_TIMEOUT:                 z.number().optional().default(1000),    // Min timeout in ms
     KB_CONFIG_MAX_TIMEOUT:                 z.number().optional().default(10000),   // Max timeout in ms
-    KB_CONFIG_CONCURRENCY_LIMIT:           z.number().optional().default(20),      // Concurrent document-processing task runs (Trigger.dev queue depth)
+    KB_CONFIG_CONCURRENCY_LIMIT:           z.number().optional().default(20),      // Per-tenant concurrent document-processing runs in the interactive lane
+    KB_CONFIG_BACKFILL_CONCURRENCY_LIMIT:  z.number().optional().default(20),      // Per-tenant concurrent document-processing runs in the connector-backfill lane
     KB_CONFIG_EMBEDDING_CONCURRENCY:       z.number().optional().default(8),       // Concurrent embedding API requests within one embed call
     /** Deployment operating budgets shared by every caller using the same provider credential. */
     KB_CONFIG_EMBEDDING_REQUESTS_PER_MINUTE: z.number().positive().optional().default(600),

apps/sim/lib/embeddings/client.ts

@@ -70,11 +70,20 @@ const logger = createLogger('EmbeddingClient')
  * Embedding requests issued concurrently within a single embed call.
  *
  * A provider's rate limit is per API key, so this multiplies with however many
- * documents are being processed at once: the document-processing queue admits
- * {@link env.KB_CONFIG_CONCURRENCY_LIMIT} task runs, each reaching here. It was
- * previously read from that same variable, so one knob set both factors and the
- * product reached four figures of in-flight requests against one key — enough to
- * hold a provider at its limit indefinitely, which no retry policy can absorb.
+ * documents are being processed at once. That document count is no longer a
+ * single number: the processing queues admit
+ * {@link env.KB_CONFIG_CONCURRENCY_LIMIT} interactive and
+ * {@link env.KB_CONFIG_BACKFILL_CONCURRENCY_LIMIT} backfill runs *per tenant*,
+ * bounded in aggregate by the Trigger.dev environment concurrency limit, and
+ * each run reaches here. The product is held down instead by the durable
+ * per-credential token bucket in `waitForProviderAdmission`, which every one of
+ * those runs shares. This factor was previously read from the same variable as
+ * the queue depth, so one knob set both and the product reached four figures of
+ * in-flight requests against one key — enough to hold a provider at its limit
+ * indefinitely, which no retry policy can absorb.
+ *
+ * The `bulk` parameter below is a different axis: it marks document indexing as
+ * opposed to query-time embedding, and is true for an interactive upload too.
  */
 const DEFAULT_CONCURRENT_BATCHES = 8
 const MAX_ALLOWED_CONCURRENT_BATCHES = 16

apps/sim/lib/knowledge/__integration__/embedding-processing-recovery.integration.ts

@@ -174,7 +174,14 @@ describe('embedding progress survives a processing slice', () => {
       })
     })
     expect(
-      await processDocumentsWithQueue([file], ids.knowledgeBaseId, {}, requestId, billing)
+      await processDocumentsWithQueue(
+        [file],
+        ids.knowledgeBaseId,
+        {},
+        requestId,
+        billing,
+        'interactive'
+      )
     ).toMatchObject({ accepted: 1, failed: 0 })
     const [deferred] = await db.select().from(document).where(eq(document.id, file.documentId))
     expect(deferred).toMatchObject({

apps/sim/lib/knowledge/__integration__/ocr-input-failures.integration.ts

@@ -168,7 +168,14 @@ describe('OCR input failures stop without partial indexing or futile retries', (
         workspaceId: ids.workspaceId,
       })
       expect(
-        await processDocumentsWithQueue([file], ids.knowledgeBaseId, {}, generateId(), billing)
+        await processDocumentsWithQueue(
+          [file],
+          ids.knowledgeBaseId,
+          {},
+          generateId(),
+          billing,
+          'interactive'
+        )
       ).toMatchObject({ accepted: 1, failed: 0 })
       const [failed] = await db.select().from(document).where(eq(document.id, file.documentId))
       expect(failed).toMatchObject({

apps/sim/lib/knowledge/__integration__/provider-processing-recovery.integration.ts

@@ -292,7 +292,8 @@ describe('provider throttling resumes the shared indexing pipeline', () => {
           ids.knowledgeBaseId,
           {},
           requestId,
-          billing
+          billing,
+          'interactive'
         )
         if (holdParentHandoff) {
           await Promise.race([

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

@@ -383,6 +383,7 @@ describe('processDocOps dispatch buffering', () => {
       {},
       expect.any(String),
       input.billingAttribution,
+      'backfill',
       { connectorId: 'connector', stillHeld: input.lease.stillHeld }
     )
     expect(input.state.result).toMatchObject({ docsAdded: 1, docsUpdated: 1 })
@@ -405,6 +406,7 @@ describe('processDocOps dispatch buffering', () => {
       {},
       expect.any(String),
       input.billingAttribution,
+      'backfill',
       { connectorId: 'connector', stillHeld: input.lease.stillHeld }
     )
     expect(input.state.result.processingDispatch).toEqual({ requested: 3, accepted: 0, failed: 0 })

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

@@ -11,7 +11,6 @@ import { toError } from '@sim/utils/errors'
 import { generateId } from '@sim/utils/id'
 import { and, asc, desc, eq, inArray, isNotNull, isNull, lt, ne, sql } from 'drizzle-orm'
 import type { BillingAttributionSnapshot } from '@/lib/billing/core/billing-attribution'
-import { env, envNumber } from '@/lib/core/config/env'
 import { ProviderCapacityDeferredError } from '@/lib/core/rate-limiter/provider-capacity-error'
 import { withDatabaseReadRetry } from '@/lib/db/read-retry'
 import type { ConnectorAccessMode } from '@/lib/knowledge/connectors/access-modes'
@@ -81,20 +80,12 @@ export const CONNECTOR_SYNC_MAX_SOURCE_PAYLOAD_BYTES = 256 * 1024 * 1024
 const PROCESSING_DISPATCH_BATCH_SIZE = 25
 
 /**
- * Bounds each sync's contribution to the shared processing queue. Oldest eligible
- * documents drain first; the remaining backlog stays eligible for subsequent syncs.
+ * Bounds each sync's contribution to this tenant's bulk processing queue. Oldest
+ * eligible documents drain first; the remaining backlog stays eligible for
+ * subsequent syncs.
  */
 export const STUCK_RETRY_MAX_CANDIDATES_PER_SYNC = 200
 
-/**
- * Concurrent `knowledge-process-document` runs, shared by every workspace.
- *
- * Read from the same env var the task itself is configured with rather than
- * restated, so the drain estimate below cannot describe a queue depth the
- * deployment does not actually run.
- */
-const PROCESSING_QUEUE_CONCURRENCY = envNumber(env.KB_CONFIG_CONCURRENCY_LIMIT, 20)
-
 export class ConnectorSyncCapacityError extends Error {}
 
 export function sourcePageFitsSyncWorkingSet(rowsAlreadyLoaded: number, pageRows: number): boolean {
@@ -1012,6 +1003,7 @@ export async function processDocOps(input: ProcessDocOpsInput): Promise<boolean>
         {},
         generateId(),
         billingAttribution,
+        'backfill',
         { connectorId, stillHeld: input.lease.stillHeld }
       )
       result.processingDispatch.accepted += dispatch.accepted
@@ -1510,6 +1502,7 @@ export async function sweepStuckDocuments(input: SweepStuckDocumentsInput): Prom
         {},
         generateId(),
         billingAttribution,
+        'backfill',
         { connectorId, stillHeld: input.lease.stillHeld }
       )
       result.processingDispatch.accepted += dispatch.accepted