fix(knowledge): run the projection source/ACL backfill in the background instead of the deploy

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

@@ -0,0 +1,35 @@
+import { task, tasks } from '@trigger.dev/sdk'
+import { resolveTriggerRegion } from '@/lib/core/async-jobs/region'
+import {
+  PROJECTION_SOURCE_ACL_BACKFILL_TASK_ID,
+  type ProjectionSourceAclBackfillPayload,
+  runProjectionSourceAclBackfill,
+} from '@/lib/knowledge/search/projection-source-acl-backfill'
+
+/** One run's share of the backfill, inside the worker's run ceiling with room to end its page. */
+const RUN_BUDGET_MS = 60 * 60 * 1000
+
+/**
+ * Trigger.dev wrapper around `runProjectionSourceAclBackfill`. A run fills unset rows for up to
+ * {@link RUN_BUDGET_MS}, then triggers its continuation from the cursor it reached, so the whole
+ * projection is filled across as many bounded runs as it takes. Retry-safe: every run writes only
+ * rows still unset, so a retried or restarted run repeats no write. The queue admits one run at a
+ * time, so two starts never fill the same pages against each other.
+ */
+export const projectionSourceAclBackfillTask = task({
+  id: PROJECTION_SOURCE_ACL_BACKFILL_TASK_ID,
+  machine: 'small-1x',
+  retry: { maxAttempts: 3 },
+  queue: {
+    name: PROJECTION_SOURCE_ACL_BACKFILL_TASK_ID,
+    concurrencyLimit: 1,
+  },
+  run: async (payload: ProjectionSourceAclBackfillPayload) => {
+    const cursor = await runProjectionSourceAclBackfill(payload, { budgetMs: RUN_BUDGET_MS })
+    if (!cursor) return
+    const continuation: ProjectionSourceAclBackfillPayload = { ...payload, cursor }
+    await tasks.trigger(PROJECTION_SOURCE_ACL_BACKFILL_TASK_ID, continuation, {
+      region: await resolveTriggerRegion(),
+    })
+  },
+})

apps/sim/lib/knowledge/access/predicate.postgres.test.ts

@@ -622,6 +622,24 @@ describe.runIf(Boolean(databaseUrl))('knowledge ACLs in PostgreSQL', () => {
     )
     expect(await admits(perRow, 'members-other')).toBe(false)
     expect(await onRowAdmits('members-other')).toBe(false)
+    /**
+     * A chunk the backfill has not reached carries no source or ACL yet and is decided on its
+     * document, as every candidate was before the columns existed: search must not depend on the
+     * backfill, must not lose a document to it, and must not rank an unreadable one because of it.
+     */
+    await connection.unsafe(
+      "INSERT INTO document(id, connector_id, acl, acl_verified_at) VALUES ('unfilled-other', 'members', ARRAY['s:slack:-:bob'], statement_timestamp())"
+    )
+    await connection.unsafe(`INSERT INTO embedding_search(id, document_id) VALUES
+      ('unfilled-other-chunk', 'unfilled-other'), ('unfilled-admin-chunk', 'admin-current'),
+      ('unfilled-members-chunk', 'members-current'), ('unfilled-gone-chunk', 'deleted-connector')`)
+    await connection.unsafe(
+      "DELETE FROM embedding_search WHERE id IN ('admin-current-chunk', 'members-current-chunk', 'deleted-connector-chunk')"
+    )
+    expect(await onRowAdmits('unfilled-other')).toBe(false)
+    expect(await onRowAdmits('admin-current')).toBe(true)
+    expect(await onRowAdmits('members-current')).toBe(true)
+    expect(await onRowAdmits('deleted-connector')).toBe(false)
     /**
      * Candidate ranking defers the live source proof, as the per-row candidate predicate does: a
      * caller holds those grants only after authorization, so applying the clause during ranking

apps/sim/lib/knowledge/access/predicate.test.ts

@@ -17,13 +17,56 @@ vi.unmock('@sim/db/schema')
 process.env.DATABASE_URL ??= 'postgresql://user:pass@localhost:5432/test'
 
 const { PgDialect } = await import('drizzle-orm/pg-core')
-const { knowledgeAccessCondition } = await import('@/lib/knowledge/access/predicate')
+const { embeddingSearch } = await import('@sim/db/schema')
+const { knowledgeAccessCondition, projectionCandidateAccessCondition } = await import(
+  '@/lib/knowledge/access/predicate'
+)
 const { SYSTEM_ACCESS_SCOPE } = await import('@/lib/knowledge/access/types')
 
 function render(condition: ReturnType<typeof knowledgeAccessCondition>) {
   return new PgDialect().sqlToQuery(condition)
 }
 
+describe('projectionCandidateAccessCondition', () => {
+  const plan = {
+    connectors: { workspace: ['ws-src'], admin: [], members: [], liveProofRequired: [] },
+    observers: { confirmed: [], observed: [] },
+    memberSources: [],
+  }
+
+  it('decides a filled row on its mirrored columns and an unfilled row on its document', () => {
+    const { sql, params } = render(
+      projectionCandidateAccessCondition(
+        embeddingSearch,
+        { kind: 'user', userId: 'user-1', tokens: ['ws', 'u:alice'] },
+        plan
+      )
+    )
+    expect(sql).toContain(
+      '("embedding_search"."acl" IS NULL AND EXISTS (\n    SELECT 1 FROM "document"\n    WHERE "document"."id" = "embedding_search"."document_id"\n      AND ('
+    )
+    expect(sql).toContain('"document"."acl" && ARRAY[$1, $2]::text[]')
+    expect(sql).toMatch(
+      /OR \("embedding_search"\."acl" && ARRAY\[\$\d+, \$\d+\]::text\[\]\n {4}AND \("embedding_search"\."connector_id" IS NULL OR "embedding_search"\."connector_id" = ANY\(ARRAY\[\$\d+\]::text\[\]\)\)\)\)$/
+    )
+    expect(params.slice(0, 2)).toEqual(['ws', 'u:alice'])
+    expect(params.slice(-3)).toEqual(['ws', 'u:alice', 'ws-src'])
+    for (const param of params) expect(Array.isArray(param)).toBe(false)
+  })
+
+  it('still denies everything for an empty token set', () => {
+    expect(
+      render(
+        projectionCandidateAccessCondition(
+          embeddingSearch,
+          { kind: 'user', userId: 'user-1', tokens: [] },
+          plan
+        )
+      ).sql
+    ).toBe('false')
+  })
+})
+
 describe('knowledgeAccessCondition', () => {
   it('overlaps the ACL with the tokens as a literal array of scalar binds', () => {
     const { sql, params } = render(

apps/sim/lib/knowledge/access/predicate.ts

@@ -312,9 +312,19 @@ export function knowledgeCandidateAccessConditionForConnectors(
  * observation's freshness, and requirement clauses live on the document. Both are refused there,
  * under the full predicate, before content is returned — this predicate only decides what is worth
  * ranking.
+ *
+ * A row the backfill has not reached yet carries no ACL (`acl IS NULL`) and is decided on its
+ * document instead, under {@link knowledgeCandidateAccessConditionForConnectors} — the join per
+ * candidate that every row paid before the columns existed. The backfill runs in the background,
+ * so search never waits on it, never loses a document to it, and never ranks an unreadable one
+ * into a bounded candidate pool because of it.
  */
 export function projectionCandidateAccessCondition(
-  projection: { connectorId: AnyPgColumn | SQL; acl: AnyPgColumn | SQL },
+  projection: {
+    connectorId: AnyPgColumn | SQL
+    acl: AnyPgColumn | SQL
+    documentId: AnyPgColumn | SQL
+  },
   scope: KnowledgeAccessScope | SystemAccessScope,
   plan: SearchAccessPlan
 ): SQL {
@@ -330,8 +340,13 @@ export function projectionCandidateAccessCondition(
     ...plan.connectors.admin,
     ...plan.connectors.members,
   ]
-  return sql`(${projection.acl} && ${tokens}
-    AND (${projection.connectorId} IS NULL OR ${inSources(mirrored)}))`
+  const unfilled = sql`(${projection.acl} IS NULL AND EXISTS (
+    SELECT 1 FROM ${document}
+    WHERE ${document.id} = ${projection.documentId}
+      AND ${knowledgeCandidateAccessConditionForConnectors(scope, plan)}
+  ))`
+  return sql`(${unfilled} OR (${projection.acl} && ${tokens}
+    AND (${projection.connectorId} IS NULL OR ${inSources(mirrored)})))`
 }
 
 /**

apps/sim/lib/knowledge/search/projection-source-acl-backfill.test.ts

@@ -0,0 +1,123 @@
+/**
+ * @vitest-environment node
+ */
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const { mockBackfill, mockEnd, mockPostgres, mockTasksTrigger } = vi.hoisted(() => ({
+  mockBackfill: vi.fn(),
+  mockEnd: vi.fn(async () => undefined),
+  mockPostgres: vi.fn(),
+  mockTasksTrigger: vi.fn(async () => ({ id: 'run-1' })),
+}))
+
+vi.mock('@sim/db', () => ({ resolveDbUrl: () => 'postgres://localhost:5432/sim' }))
+vi.mock('@sim/db/script-migrations/0021_embedding_search_connector', () => ({
+  PROJECTION_SOURCE_ACL_TABLES: ['embedding_search', 'embedding_keyword_tin'],
+  backfillProjectionSourceAcl: mockBackfill,
+}))
+vi.mock('postgres', () => ({ default: mockPostgres }))
+vi.mock('@trigger.dev/sdk', () => ({ tasks: { trigger: mockTasksTrigger } }))
+vi.mock('@/lib/core/async-jobs/region', () => ({ resolveTriggerRegion: async () => 'us-east-1' }))
+vi.mock('@/lib/core/utils/background', () => ({
+  runDetached: (_label: string, work: () => Promise<unknown>) => {
+    void work()
+  },
+}))
+
+import {
+  enqueueProjectionSourceAclBackfill,
+  runProjectionSourceAclBackfill,
+} from '@/lib/knowledge/search/projection-source-acl-backfill'
+
+const connection = { end: mockEnd }
+
+describe('runProjectionSourceAclBackfill', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    mockPostgres.mockReturnValue(connection)
+    mockBackfill.mockImplementation(async (_sql, projection, options) => ({
+      projection,
+      scanned: 0,
+      written: 0,
+      afterId: options?.afterId ?? '',
+      done: true,
+    }))
+  })
+
+  it('fills both projections in order on its own connection and closes it', async () => {
+    await expect(runProjectionSourceAclBackfill({ pageSize: 50, pauseMs: 10 })).resolves.toBeNull()
+    expect(mockBackfill.mock.calls.map(([, projection]) => projection)).toEqual([
+      'embedding_search',
+      'embedding_keyword_tin',
+    ])
+    for (const [sql, , options] of mockBackfill.mock.calls) {
+      expect(sql).toBe(connection)
+      expect(options).toMatchObject({ afterId: undefined, pageSize: 50, pauseMs: 10 })
+    }
+    expect(mockEnd).toHaveBeenCalledTimes(1)
+  })
+
+  it('resumes after the cursor in its projection and from the start of the next', async () => {
+    await runProjectionSourceAclBackfill({
+      cursor: { projection: 'embedding_keyword_tin', afterId: 'chunk-9' },
+    })
+    expect(mockBackfill).toHaveBeenCalledTimes(1)
+    expect(mockBackfill.mock.calls[0][1]).toBe('embedding_keyword_tin')
+    expect(mockBackfill.mock.calls[0][2]).toMatchObject({ afterId: 'chunk-9' })
+  })
+
+  it('returns where a budgeted run stopped so the next run can carry on', async () => {
+    mockBackfill.mockResolvedValueOnce({
+      projection: 'embedding_search',
+      scanned: 100,
+      written: 100,
+      afterId: 'chunk-100',
+      done: false,
+    })
+    await expect(runProjectionSourceAclBackfill({}, { budgetMs: 1000 })).resolves.toEqual({
+      projection: 'embedding_search',
+      afterId: 'chunk-100',
+    })
+    expect(mockBackfill).toHaveBeenCalledTimes(1)
+    expect(mockBackfill.mock.calls[0][2].budgetMs).toBeLessThanOrEqual(1000)
+    expect(mockEnd).toHaveBeenCalledTimes(1)
+  })
+
+  it('closes the connection when a page fails', async () => {
+    mockBackfill.mockRejectedValueOnce(new Error('canceling statement due to statement timeout'))
+    await expect(runProjectionSourceAclBackfill({})).rejects.toThrow('statement timeout')
+    expect(mockEnd).toHaveBeenCalledTimes(1)
+  })
+})
+
+describe('enqueueProjectionSourceAclBackfill', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    mockPostgres.mockReturnValue(connection)
+    mockBackfill.mockResolvedValue({
+      projection: 'embedding_search',
+      scanned: 0,
+      written: 0,
+      afterId: '',
+      done: true,
+    })
+  })
+
+  it('hands the backfill to the Trigger.dev worker when one is configured', async () => {
+    await expect(enqueueProjectionSourceAclBackfill({ pageSize: 25 }, true)).resolves.toEqual({
+      runId: 'run-1',
+    })
+    expect(mockTasksTrigger).toHaveBeenCalledWith(
+      'projection-source-acl-backfill',
+      { pageSize: 25 },
+      { region: 'us-east-1' }
+    )
+    expect(mockBackfill).not.toHaveBeenCalled()
+  })
+
+  it('fills the projections detached in this process without one', async () => {
+    await expect(enqueueProjectionSourceAclBackfill({}, false)).resolves.toBeNull()
+    expect(mockTasksTrigger).not.toHaveBeenCalled()
+    await vi.waitFor(() => expect(mockBackfill).toHaveBeenCalledTimes(2))
+  })
+})

apps/sim/lib/knowledge/search/projection-source-acl-backfill.ts

@@ -0,0 +1,98 @@
+import { resolveDbUrl } from '@sim/db'
+import {
+  backfillProjectionSourceAcl,
+  PROJECTION_SOURCE_ACL_TABLES,
+  type ProjectionSourceAclTable,
+} from '@sim/db/script-migrations/0021_embedding_search_connector'
+import { createLogger } from '@sim/logger'
+import postgres from 'postgres'
+import { resolveTriggerRegion } from '@/lib/core/async-jobs/region'
+import { env } from '@/lib/core/config/env'
+import { isTriggerDevEnabled } from '@/lib/core/config/env-flags'
+import { runDetached } from '@/lib/core/utils/background'
+
+const logger = createLogger('ProjectionSourceAclBackfill')
+
+export const PROJECTION_SOURCE_ACL_BACKFILL_TASK_ID = 'projection-source-acl-backfill'
+
+/** Where a run stopped, so the next one carries on from there instead of rescanning. */
+export interface ProjectionSourceAclBackfillCursor {
+  projection: ProjectionSourceAclTable
+  afterId: string
+}
+
+export interface ProjectionSourceAclBackfillPayload {
+  /** The first projection's first page when absent. */
+  cursor?: ProjectionSourceAclBackfillCursor
+  pageSize?: number
+  pauseMs?: number
+}
+
+export interface ProjectionSourceAclBackfillRunOptions {
+  /** Stop once this much time has passed and return where to resume; unbounded otherwise. */
+  budgetMs?: number
+}
+
+/**
+ * Fills the ranking projections' source and ACL columns from the cursor onwards, one projection
+ * after the other, on a connection of its own: the page statement binds the keyset cursor as a
+ * scalar and needs no array parameter, so the pool's options would serve, but a run this long
+ * should not hold one of the worker's pooled connections. Returns the cursor to continue from when
+ * the budget ran out, `null` once both projections are filled.
+ */
+export async function runProjectionSourceAclBackfill(
+  payload: ProjectionSourceAclBackfillPayload,
+  options: ProjectionSourceAclBackfillRunOptions = {}
+): Promise<ProjectionSourceAclBackfillCursor | null> {
+  const url = resolveDbUrl('DATABASE_URL', process.env.SIM_DB_ROLE?.trim() || 'web')
+  if (!url) throw new Error('DATABASE_URL is required to backfill the projection source and ACL')
+  const sql = postgres(url, { max: 1, max_lifetime: null, onnotice: () => undefined })
+  const startedAt = Date.now()
+  try {
+    const start = payload.cursor
+      ? PROJECTION_SOURCE_ACL_TABLES.indexOf(payload.cursor.projection)
+      : 0
+    if (start < 0) throw new Error(`Unknown projection ${payload.cursor?.projection}`)
+    for (const projection of PROJECTION_SOURCE_ACL_TABLES.slice(start)) {
+      const budgetMs =
+        options.budgetMs === undefined
+          ? undefined
+          : Math.max(0, options.budgetMs - (Date.now() - startedAt))
+      const progress = await backfillProjectionSourceAcl(sql, projection, {
+        afterId: payload.cursor?.projection === projection ? payload.cursor.afterId : undefined,
+        pageSize: payload.pageSize,
+        pauseMs: payload.pauseMs,
+        budgetMs,
+      })
+      if (!progress.done) return { projection, afterId: progress.afterId }
+    }
+    logger.info('Projection source and ACL backfill complete', {
+      elapsedMs: Date.now() - startedAt,
+    })
+    return null
+  } finally {
+    await sql.end()
+  }
+}
+
+/**
+ * Starts the backfill the way the table backfill is started: on the deployment's Trigger.dev
+ * worker when one is configured, where bounded runs chain until both projections are filled, and
+ * detached in this process otherwise. Safe to call again at any time — a run only fills rows still
+ * unset.
+ */
+export async function enqueueProjectionSourceAclBackfill(
+  payload: ProjectionSourceAclBackfillPayload = {},
+  useTrigger = Boolean(isTriggerDevEnabled && env.TRIGGER_SECRET_KEY)
+): Promise<{ runId: string } | null> {
+  if (useTrigger) {
+    const { tasks } = await import('@trigger.dev/sdk')
+    const handle = await tasks.trigger(PROJECTION_SOURCE_ACL_BACKFILL_TASK_ID, payload, {
+      region: await resolveTriggerRegion(),
+    })
+    logger.info('Projection source and ACL backfill enqueued', { runId: handle.id })
+    return { runId: handle.id }
+  }
+  runDetached(PROJECTION_SOURCE_ACL_BACKFILL_TASK_ID, () => runProjectionSourceAclBackfill(payload))
+  return null
+}

apps/sim/lib/knowledge/search/queries.test.ts

@@ -1527,6 +1527,9 @@ describe('permitted-document planner', () => {
       /** The ranked CTE carries the mirrored source and ACL the predicate tests. */
       expect(statement).toContain('AS connector_id')
       expect(statement).toContain('ranked_tin_chunks.acl')
+      /** A row the backfill has not filled (`acl IS NULL`) is decided on its document instead. */
+      expect(statement).toContain('IS NULL AND EXISTS (')
+      expect(statement).toContain('ranked_tin_chunks.document_id')
     })
 
     it('widens the window for a broad resolved scope whose first page came back short', async () => {