feat(knowledge): project document ACL and chunk changes asynchronously (#8199)

* feat(knowledge): project document ACL and chunk changes asynchronously

Document ACL/source changes and chunk writes mark their document in
knowledge_projection_dirty (one upsert with a generation bump) in the
writer's transaction. A knowledge projector converges every search
projection per document in short pages bounded by chunk rows, several
documents at once, and removes a mark only on the generation it read.

Writers that declare sim.projection_mode = 'async' skip the synchronous
embedding and document fan-out triggers. The processing commit and every
connector-lease ACL page declare it while the knowledge-async-projection
flag is on; every other writer, and every release before this one, keeps
writing projection rows itself. Marks are written in both modes so a
projector pass can never settle over a concurrent synchronous write.
Migrations 0021-0023 keep their trigger body; 0024 alone installs the
marking.

Search decides a marked document's rows on the document itself, so a
pending projection never admits a revoked grant. A document that moved
sources joins the new source's per-source ranking after its pass.

The projector runs as a Trigger.dev task, requested after writes and
swept every minute while there is work, or in-process without
Trigger.dev. A pass opens one worker per mark up to
KB_CONFIG_PROJECTION_CONCURRENCY. The separate source/ACL backfill is
folded into it behind knowledge-projection-fill.

* fix(knowledge): keep projection guards out of historical migrations

Restores 0016, 0019 and 0021 to their staging bodies; 0024 alone
re-creates the projection triggers with the mode guard, in the same
transaction that installs the marks. The fill starts only inside the
pass budget and reports what it marked as remaining. Passes dispatch
to Trigger.dev by the rule document processing uses, and the inline
coalescer clears its running flag in the same step it reads the owed
flag. Bumps the Helm chart for the projection cron job.

* test(knowledge): count projection requests per connector ACL case

Author: waleedlatif1

.github/workflows/test-build.yml

@@ -282,6 +282,8 @@ jobs:
           lib/knowledge/__integration__/kb-block-search.integration.ts
           lib/knowledge/__integration__/gitlab-workspace.integration.ts
           lib/knowledge/__integration__/unfilled-projection-source.integration.ts
+          lib/knowledge/__integration__/knowledge-projection.integration.ts
+          lib/knowledge/__integration__/async-projection-processing.integration.ts
           lib/knowledge/__integration__/purged-detach-reservation.integration.ts
           lib/core/outbox/service.integration.ts
           lib/knowledge/__integration__/connector-upload.integration.ts

apps/docs/content/docs/platform/self-hosting/background-jobs.mdx

@@ -47,6 +47,7 @@ Point cron at an **internal** address where possible (the in-cluster Service, or
 | Time pause/resume | `/api/resume/poll` | `*/1 * * * *` | Workflows paused on a timer |
 | Outbox processing | `/api/webhooks/outbox/process` | `*/1 * * * *` | Transactional-outbox retries for billing, membership, enterprise issuance, and workflow-deployment side effects |
 | Workspace file search dispatch | `/api/cron/workspace-file-search-dispatch` | `*/1 * * * *` | Dispatches indexing work for workspace file search |
+| Knowledge projection | `/api/cron/knowledge-projection` | `*/1 * * * *` | Brings knowledge base search up to date with document, permission, and chunk changes |
 | Connector sync | `/api/knowledge/connectors/sync` | `*/5 * * * *` | Knowledge base connector syncs |
 | Connector member sync | `/api/knowledge/connectors/member-sync` | `*/5 * * * *` | Per-member access sync for permission-aware connectors |
 | Connector directory sync | `/api/knowledge/connectors/directory-sync` | `*/5 * * * *` | Refreshes the directory groups administrator-mode connectors mirror, so a membership change takes effect without waiting for a content sync |

apps/sim/app/api/cron/knowledge-projection/route.test.ts

@@ -0,0 +1,86 @@
+/**
+ * @vitest-environment node
+ */
+import { createMockRequest } from '@sim/testing'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const mocks = vi.hoisted(() => ({
+  enqueueSweep: vi.fn(),
+  verifyCronAuth: vi.fn(),
+}))
+
+vi.mock('@/lib/auth/internal', () => ({ verifyCronAuth: mocks.verifyCronAuth }))
+vi.mock('@/lib/knowledge/projection/enqueue', () => ({
+  enqueueKnowledgeProjectionSweep: mocks.enqueueSweep,
+}))
+
+import { GET } from '@/app/api/cron/knowledge-projection/route'
+
+function request() {
+  return createMockRequest(
+    'GET',
+    undefined,
+    {},
+    'http://localhost:3000/api/cron/knowledge-projection'
+  )
+}
+
+describe('knowledge projection sweep route', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    mocks.verifyCronAuth.mockReturnValue(null)
+  })
+
+  it('returns as soon as Trigger.dev accepts the pass', async () => {
+    mocks.enqueueSweep.mockResolvedValue({
+      triggered: true,
+      backend: 'trigger-dev',
+      jobId: 'run-1',
+    })
+
+    const response = await GET(request())
+
+    expect(response.status).toBe(202)
+    await expect(response.json()).resolves.toEqual({
+      success: true,
+      triggered: true,
+      backend: 'trigger-dev',
+      jobId: 'run-1',
+    })
+  })
+
+  it('answers 200 without a pass when the projector has nothing to do', async () => {
+    mocks.enqueueSweep.mockResolvedValue({ triggered: false, backend: null, jobId: null })
+
+    const response = await GET(request())
+
+    expect(response.status).toBe(200)
+    await expect(response.json()).resolves.toEqual({
+      success: true,
+      triggered: false,
+      backend: null,
+      jobId: null,
+    })
+  })
+
+  it('returns the cron auth refusal without enqueueing', async () => {
+    mocks.verifyCronAuth.mockReturnValue(new Response(null, { status: 401 }))
+
+    const response = await GET(request())
+
+    expect(response.status).toBe(401)
+    expect(mocks.enqueueSweep).not.toHaveBeenCalled()
+  })
+
+  it('fails closed when Trigger.dev does not accept the pass', async () => {
+    mocks.enqueueSweep.mockRejectedValue(new Error('trigger unavailable'))
+
+    const response = await GET(request())
+
+    expect(response.status).toBe(500)
+    await expect(response.json()).resolves.toEqual({
+      success: false,
+      error: 'Sweep enqueue failed',
+    })
+  })
+})

apps/sim/app/api/cron/knowledge-projection/route.ts

@@ -0,0 +1,29 @@
+import { createLogger } from '@sim/logger'
+import { getErrorMessage } from '@sim/utils/errors'
+import { type NextRequest, NextResponse } from 'next/server'
+import { verifyCronAuth } from '@/lib/auth/internal'
+import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
+import { enqueueKnowledgeProjectionSweep } from '@/lib/knowledge/projection/enqueue'
+
+const logger = createLogger('KnowledgeProjectionSweepRoute')
+
+export const dynamic = 'force-dynamic'
+export const maxDuration = 60
+
+/**
+ * The knowledge projector's periodic sweep: enqueues one pass per window while there is work, and
+ * returns once Trigger.dev accepts it. Writers ask for passes as they commit; this converges
+ * whatever those requests missed.
+ */
+export const GET = withRouteHandler(async (request: NextRequest) => {
+  const authError = verifyCronAuth(request, 'Knowledge projection sweep')
+  if (authError) return authError
+
+  try {
+    const result = await enqueueKnowledgeProjectionSweep()
+    return NextResponse.json({ success: true, ...result }, { status: result.triggered ? 202 : 200 })
+  } catch (error) {
+    logger.error('Knowledge projection sweep enqueue failed', { error: getErrorMessage(error) })
+    return NextResponse.json({ success: false, error: 'Sweep enqueue failed' }, { status: 500 })
+  }
+})

apps/sim/background/knowledge-projection.ts

@@ -0,0 +1,45 @@
+import { task } from '@trigger.dev/sdk'
+import {
+  type BackgroundRetryPolicy,
+  backgroundRetryAttemptCeiling,
+  getBackgroundRetryDecision,
+} from '@/lib/core/errors/background-retry'
+import {
+  KNOWLEDGE_PROJECTION_PASS_BUDGET_MS,
+  KNOWLEDGE_PROJECTION_TASK_ID,
+  requestKnowledgeProjection,
+} from '@/lib/knowledge/projection/enqueue'
+import { runKnowledgeProjectionPass } from '@/lib/knowledge/projection/run'
+
+/**
+ * A pass gives a single document up on a lock or statement timeout without failing, so a failed
+ * pass lost its connection or its database. Those back off for minutes; the sweep starts a fresh
+ * pass every minute regardless, so a few attempts are enough.
+ */
+export const KNOWLEDGE_PROJECTION_RETRY_POLICY: BackgroundRetryPolicy = {
+  maxAttempts: 2,
+  database: { maxAttempts: 3, baseDelayMs: 60 * 1000, maxDelayMs: 5 * 60 * 1000 },
+}
+
+/**
+ * Runs one knowledge projector pass. One pass runs at a time and projects several documents at
+ * once itself; the prompt requests and the sweep collapse into whichever pass is queued. A pass
+ * that ran out of budget with marks left asks for the next one. Retry-safe: a pass writes only rows
+ * that differ from their source and removes a mark only on the generation it read.
+ */
+export const knowledgeProjectionTask = task({
+  id: KNOWLEDGE_PROJECTION_TASK_ID,
+  machine: 'small-1x',
+  maxDuration: 15 * 60,
+  retry: { maxAttempts: backgroundRetryAttemptCeiling(KNOWLEDGE_PROJECTION_RETRY_POLICY) },
+  queue: { name: KNOWLEDGE_PROJECTION_TASK_ID, concurrencyLimit: 1 },
+  catchError: async ({ error, ctx }) =>
+    getBackgroundRetryDecision(error, ctx.attempt.number, KNOWLEDGE_PROJECTION_RETRY_POLICY),
+  run: async () => {
+    const result = await runKnowledgeProjectionPass({
+      budgetMs: KNOWLEDGE_PROJECTION_PASS_BUDGET_MS,
+    })
+    if (result.remaining) await requestKnowledgeProjection()
+    return result
+  },
+})

apps/sim/background/projection-source-acl-backfill.ts

@@ -464,6 +464,7 @@ export const env = createEnv({
     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
+    KB_CONFIG_PROJECTION_CONCURRENCY:      z.number().optional().default(8),       // Most documents one knowledge projector pass projects at once, each on its own connection
     /** Deployment operating budgets shared by every caller using the same provider credential. */
     KB_CONFIG_EMBEDDING_REQUESTS_PER_MINUTE: z.number().positive().optional().default(600),
     KB_CONFIG_EMBEDDING_TOKENS_PER_MINUTE:   z.number().positive().optional().default(600000),
@@ -634,6 +635,8 @@ export const env = createEnv({
     CREDENTIAL_GROUPS:                    z.boolean().optional(),                 // Enable enterprise Credential Groups globally
     KNOWLEDGE_MEMBER_ACCESS:              z.boolean().optional(),                 // Enable per-member knowledge connectors and hybrid-by-default retrieval globally
     KNOWLEDGE_TIN_KEYWORD:                z.boolean().optional(),                 // Rank large-scope keyword retrieval through the Tin text index where it exists
+    KNOWLEDGE_ASYNC_PROJECTION:           z.boolean().optional(),                 // Knowledge writers leave search projection rows to the background projector
+    KNOWLEDGE_PROJECTION_FILL:            z.boolean().optional(),                 // The knowledge projector fills projection rows written before they carried a source and ACL
 
     // Organizations - for self-hosted deployments
     ORGANIZATIONS_ENABLED:                 z.boolean().optional(),                 // Enable organizations on self-hosted (bypasses plan requirements)

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

@@ -10,6 +10,8 @@ const { mockFetch, mockIsPlatformAdmin, envRef } = vi.hoisted(() => ({
   mockIsPlatformAdmin: vi.fn(),
   envRef: {
     APPCONFIG_APPLICATION: 'sim-staging' as string | undefined,
+    KNOWLEDGE_PROJECTION_FILL: undefined as boolean | undefined,
+    KNOWLEDGE_ASYNC_PROJECTION: undefined as boolean | undefined,
     APPCONFIG_ENVIRONMENT: 'staging' as string | undefined,
     TABLES_V2_API: undefined as boolean | undefined,
     TABLE_ROW_TTL: undefined as boolean | undefined,
@@ -148,6 +150,8 @@ describe('isFeatureEnabled', () => {
     envRef.CREDENTIAL_GROUPS = undefined
     envRef.KNOWLEDGE_MEMBER_ACCESS = undefined
     envRef.KNOWLEDGE_TIN_KEYWORD = undefined
+    envRef.KNOWLEDGE_ASYNC_PROJECTION = undefined
+    envRef.KNOWLEDGE_PROJECTION_FILL = undefined
     envRef.SLACK_SEARCH_SHARED_APP = undefined
   })
 
@@ -197,6 +201,32 @@ describe('isFeatureEnabled', () => {
     })
   })
 
+  describe('knowledge-async-projection flag', () => {
+    it('is a global switch', async () => {
+      expect(await isFeatureEnabled('knowledge-async-projection')).toBe(false)
+      envRef.KNOWLEDGE_ASYNC_PROJECTION = true
+      expect(await isFeatureEnabled('knowledge-async-projection')).toBe(true)
+    })
+
+    it('follows an AppConfig global rule', async () => {
+      withAppConfig({ 'knowledge-async-projection': { enabled: true } })
+      expect(await isFeatureEnabled('knowledge-async-projection')).toBe(true)
+    })
+  })
+
+  describe('knowledge-projection-fill flag', () => {
+    it('is a global switch', async () => {
+      expect(await isFeatureEnabled('knowledge-projection-fill')).toBe(false)
+      envRef.KNOWLEDGE_PROJECTION_FILL = true
+      expect(await isFeatureEnabled('knowledge-projection-fill')).toBe(true)
+    })
+
+    it('follows an AppConfig global rule', async () => {
+      withAppConfig({ 'knowledge-projection-fill': { enabled: true } })
+      expect(await isFeatureEnabled('knowledge-projection-fill')).toBe(true)
+    })
+  })
+
   describe('knowledge-member-access flag', () => {
     it('uses a global fallback switch off AppConfig', async () => {
       expect(await isFeatureEnabled('knowledge-member-access')).toBe(false)

apps/sim/lib/core/config/feature-flags.test.ts

@@ -105,6 +105,23 @@ const FEATURE_FLAGS = {
       'invalid. Off-AppConfig falls back to KNOWLEDGE_TIN_KEYWORD.',
     fallback: 'KNOWLEDGE_TIN_KEYWORD',
   },
+  'knowledge-async-projection': {
+    description:
+      'Knowledge writers (document processing and connector ACL writes) leave search projection ' +
+      'rows to the background knowledge projector instead of rewriting them in their own ' +
+      'transaction. Global on/off only; turn it on only once no release older than the ' +
+      'projector serves search. Off-AppConfig falls back to KNOWLEDGE_ASYNC_PROJECTION.',
+    fallback: 'KNOWLEDGE_ASYNC_PROJECTION',
+  },
+  'knowledge-projection-fill': {
+    description:
+      'The knowledge projector also fills search projection rows written before they carried ' +
+      "their document's source and ACL, marking at most 100 documents at once so fresh writes " +
+      'never wait behind much of it. Global on/off only; off pauses the fill, and search keeps ' +
+      'deciding unfilled rows on their document. Off-AppConfig falls back to ' +
+      'KNOWLEDGE_PROJECTION_FILL.',
+    fallback: 'KNOWLEDGE_PROJECTION_FILL',
+  },
 } satisfies Record<string, FeatureFlagDefinition>
 
 /**