fix(knowledge): decide unfilled projection rows on their document and start the backfill from the outbox

An unfilled row no longer passes the on-row candidate predicate outright:
it is decided on its document under the resolved candidate predicate, the
join per candidate every row paid before the columns existed, so the
bounded candidate pools and the exact slice hold only rows that hydration
will keep. A partial index on the unfilled rows keeps that branch, and the
backfill's keyset pages, an index probe.

The migration also leaves one outbox event whose handler starts the
backfill task, so it runs after every deploy without an operator.

Claude-Session: https://claude.ai/code/session_01XU6c7pKRpa5CMoMHKDdqxX

Author: waleedlatif1

apps/sim/lib/core/outbox/processor.ts

@@ -18,6 +18,7 @@ import { slackSearchOutboxHandlers } from '@/lib/knowledge/application/slack-sea
 import { getConnectorFailureDiagnostic } from '@/lib/knowledge/connectors/connector-error'
 import { knowledgeDocumentProcessingOutboxHandlers } from '@/lib/knowledge/documents/processing-outbox-handler'
 import { recoverKnowledgeDocumentProcessing } from '@/lib/knowledge/documents/processing-recovery'
+import { projectionSourceAclBackfillOutboxHandlers } from '@/lib/knowledge/search/projection-source-acl-backfill'
 import { inboxCleanupOutboxHandlers } from '@/lib/mothership/inbox/cleanup-outbox'
 import { organizationResourceCleanupOutboxHandlers } from '@/lib/organizations/resource-cleanup'
 import { workspaceFileLiveDocOutboxHandlers } from '@/lib/uploads/contexts/workspace/workspace-file-live-doc-outbox'
@@ -42,6 +43,7 @@ const handlers = {
   ...invitationMigrationOutboxHandlers,
   ...directGrantOutboxHandlers,
   ...knowledgeDocumentProcessingOutboxHandlers,
+  ...projectionSourceAclBackfillOutboxHandlers,
   ...organizationResourceCleanupOutboxHandlers,
   ...inboxCleanupOutboxHandlers,
   ...permissionAccessRequestOutboxHandlers,

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

@@ -623,19 +623,23 @@ 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. It is ranked, and decided
-     * at hydration under the full predicate, exactly as every candidate was before the columns
-     * existed — search must not depend on the backfill, and must not lose a document to it.
+     * 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(
-      "INSERT INTO embedding_search(id, document_id) VALUES ('unfilled-other-chunk', 'unfilled-other'), ('admin-current-unfilled', 'admin-current')"
+      "DELETE FROM embedding_search WHERE id IN ('admin-current-chunk', 'members-current-chunk', 'deleted-connector-chunk')"
     )
-    expect(await onRowAdmits('unfilled-other')).toBe(true)
-    expect(await admits(perRow, 'unfilled-other')).toBe(false)
+    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

@@ -34,19 +34,24 @@ describe('projectionCandidateAccessCondition', () => {
     memberSources: [],
   }
 
-  it('admits a row the backfill has not filled, and decides a filled row on its mirrored columns', () => {
+  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).toBe(
-      '("embedding_search"."acl" IS NULL OR ("embedding_search"."acl" && ARRAY[$1, $2]::text[]\n' +
-        '    AND ("embedding_search"."connector_id" IS NULL OR "embedding_search"."connector_id" = ANY(ARRAY[$3]::text[]))))'
+    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(params).toEqual(['ws', 'u:alice', 'ws-src'])
+    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', () => {

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

@@ -313,13 +313,18 @@ export function knowledgeCandidateAccessConditionForConnectors(
  * 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 passes: it is decided at
- * hydration under the full document predicate, exactly as every candidate was before the columns
- * existed. The backfill runs in the background, so search never waits on it and never loses a
- * document to it.
+ * 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 {
@@ -335,7 +340,12 @@ export function projectionCandidateAccessCondition(
     ...plan.connectors.admin,
     ...plan.connectors.members,
   ]
-  return sql`(${projection.acl} IS NULL OR (${projection.acl} && ${tokens}
+  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

@@ -13,14 +13,21 @@ const { mockBackfill, mockEnd, mockPostgres, mockTasksTrigger } = vi.hoisted(()
 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'],
+  PROJECTION_SOURCE_ACL_BACKFILL_OUTBOX_EVENT: 'knowledge.projection.source_acl.backfill',
   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,
+  projectionSourceAclBackfillOutboxHandlers,
   runProjectionSourceAclBackfill,
 } from '@/lib/knowledge/search/projection-source-acl-backfill'
 
@@ -110,9 +117,15 @@ describe('enqueueProjectionSourceAclBackfill', () => {
     expect(mockBackfill).not.toHaveBeenCalled()
   })
 
-  it('fills the projections inline without one', async () => {
+  it('fills the projections detached in this process without one', async () => {
     await expect(enqueueProjectionSourceAclBackfill({}, false)).resolves.toBeNull()
     expect(mockTasksTrigger).not.toHaveBeenCalled()
-    expect(mockBackfill).toHaveBeenCalledTimes(2)
+    await vi.waitFor(() => expect(mockBackfill).toHaveBeenCalledTimes(2))
+  })
+
+  it('starts from the outbox event the migration leaves behind', () => {
+    expect(Object.keys(projectionSourceAclBackfillOutboxHandlers)).toEqual([
+      'knowledge.projection.source_acl.backfill',
+    ])
   })
 })

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

@@ -1,6 +1,7 @@
 import { resolveDbUrl } from '@sim/db'
 import {
   backfillProjectionSourceAcl,
+  PROJECTION_SOURCE_ACL_BACKFILL_OUTBOX_EVENT,
   PROJECTION_SOURCE_ACL_TABLES,
   type ProjectionSourceAclTable,
 } from '@sim/db/script-migrations/0021_embedding_search_connector'
@@ -9,6 +10,8 @@ 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 type { OutboxHandlerRegistry } from '@/lib/core/outbox/service'
+import { runDetached } from '@/lib/core/utils/background'
 
 const logger = createLogger('ProjectionSourceAclBackfill')
 
@@ -76,8 +79,9 @@ export async function runProjectionSourceAclBackfill(
 
 /**
  * Starts the backfill the way the table backfill is started: on the deployment's Trigger.dev
- * worker when there is one, where bounded runs chain until both projections are filled, and inline
- * otherwise. Safe to call again at any time — a run only fills rows still unset.
+ * worker when there is one, 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 = {},
@@ -91,6 +95,16 @@ export async function enqueueProjectionSourceAclBackfill(
     logger.info('Projection source and ACL backfill enqueued', { runId: handle.id })
     return { runId: handle.id }
   }
-  await runProjectionSourceAclBackfill(payload)
+  runDetached(PROJECTION_SOURCE_ACL_BACKFILL_TASK_ID, () => runProjectionSourceAclBackfill(payload))
   return null
 }
+
+/**
+ * The event script migration `0021_embedding_search_connector` leaves behind, so the backfill it
+ * hands off starts once the app that ships this handler is up, on every deployment.
+ */
+export const projectionSourceAclBackfillOutboxHandlers: OutboxHandlerRegistry = {
+  [PROJECTION_SOURCE_ACL_BACKFILL_OUTBOX_EVENT]: async () => {
+    await enqueueProjectionSourceAclBackfill()
+  },
+}

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

@@ -1527,8 +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 ranked and decided at hydration. */
-      expect(statement).toContain('" IS NULL OR ("')
+      /** 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 () => {

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

@@ -1819,7 +1819,11 @@ export async function executeKeywordSearch(params: KeywordSearchParams): Promise
     const onRowKeywordVisibility = (excludedSources: readonly string[]) =>
       and(
         projectionCandidateAccessCondition(
-          { connectorId: sql`ranked_tin_chunks.connector_id`, acl: sql`ranked_tin_chunks.acl` },
+          {
+            connectorId: sql`ranked_tin_chunks.connector_id`,
+            acl: sql`ranked_tin_chunks.acl`,
+            documentId: sql`ranked_tin_chunks.document_id`,
+          },
           access,
           accessPlan!
         ),

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

@@ -2,30 +2,40 @@
 
 /**
  * Starts the projection source and ACL backfill that script migration
- * `0021_embedding_search_connector` leaves to the background: on a deployment with Trigger.dev it
- * enqueues the `projection-source-acl-backfill` task, which chains bounded runs until both ranking
- * projections are filled; without one it fills them here, paced the same way. Safe to run again at
- * any time — a run only fills rows still unset.
+ * `0021_embedding_search_connector` leaves to the background, by hand. The migration's outbox
+ * event already starts it after a deploy; this is for starting it again — on a deployment with
+ * Trigger.dev it enqueues the `projection-source-acl-backfill` task, which chains bounded runs
+ * until both ranking projections are filled; without one it fills them here, paced the same way.
+ * Safe to run at any time — a run only fills rows still unset.
  *
  * Usage:
  *   bun apps/sim/scripts/backfill-projection-source-acl.ts
  */
 
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
-import { enqueueProjectionSourceAclBackfill } from '@/lib/knowledge/search/projection-source-acl-backfill'
+import { env } from '@/lib/core/config/env'
+import { isTriggerDevEnabled } from '@/lib/core/config/env-flags'
+import {
+  enqueueProjectionSourceAclBackfill,
+  runProjectionSourceAclBackfill,
+} from '@/lib/knowledge/search/projection-source-acl-backfill'
 
 const logger = createLogger('BackfillProjectionSourceAcl')
 
+async function main(): Promise<void> {
+  if (isTriggerDevEnabled && env.TRIGGER_SECRET_KEY) {
+    const handle = await enqueueProjectionSourceAclBackfill({}, true)
+    logger.info('Backfill enqueued on the Trigger.dev worker', handle ?? {})
+    return
+  }
+  await runProjectionSourceAclBackfill({})
+  logger.info('Backfill complete')
+}
+
 if (import.meta.main) {
-  enqueueProjectionSourceAclBackfill().then(
-    (handle) => {
-      logger.info(
-        handle ? 'Backfill enqueued on the Trigger.dev worker' : 'Backfill complete',
-        handle ?? {}
-      )
-      process.exit(0)
-    },
+  main().then(
+    () => process.exit(0),
     (error) => {
       logger.error('Backfill failed', toError(error))
       process.exit(1)

packages/db/script-migrations/0021_embedding_search_connector.postgres.test.ts

@@ -40,6 +40,10 @@ describe.runIf(Boolean(databaseUrl))('projection source and ACL backfill in Post
     await sql`CREATE TABLE document (
       id text PRIMARY KEY, connector_id text, acl text[] NOT NULL DEFAULT '{ws}'
     )`
+    await sql`CREATE TABLE outbox_event (
+      id text PRIMARY KEY, event_type text NOT NULL, payload json NOT NULL,
+      status text NOT NULL DEFAULT 'pending'
+    )`
     for (const projection of ['embedding_search', 'embedding_keyword_tin']) {
       await sql`CREATE TABLE ${sql(projection)} (
         id text PRIMARY KEY, document_id text NOT NULL, enabled boolean NOT NULL DEFAULT true,
@@ -61,17 +65,27 @@ describe.runIf(Boolean(databaseUrl))('projection source and ACL backfill in Post
     await sql`ALTER TABLE embedding_keyword_tin DISABLE TRIGGER embedding_keyword_tin_source_acl_set`
   })
 
-  it('installs its triggers and indexes again without failing, so a cut-short deploy completes', async () => {
+  it('installs its triggers, indexes and start event again without failing, so a cut-short deploy completes', async () => {
     await expect(embeddingSearchConnectorMigration.up(sql)).resolves.toBeUndefined()
     const indexes = await sql<{ indexname: string }[]>`
       SELECT indexname FROM pg_indexes WHERE schemaname = ${schemaName} ORDER BY indexname`
     expect(indexes.map((row) => row.indexname)).toEqual(
       expect.arrayContaining([
         'embedding_keyword_tin_acl_gin_idx',
+        'embedding_keyword_tin_acl_unfilled_idx',
         'embedding_search_acl_gin_idx',
+        'embedding_search_acl_unfilled_idx',
         'embedding_search_source_idx',
       ])
     )
+    /** Two runs of the migration leave the app one event, so the backfill is started once. */
+    expect(await sql`SELECT id, event_type, status FROM outbox_event`).toEqual([
+      {
+        id: 'projection-source-acl-backfill:0021',
+        event_type: 'knowledge.projection.source_acl.backfill',
+        status: 'pending',
+      },
+    ])
   })
 
   it('fills only the rows still unset, in pages, and leaves a chunk that changed documents to its trigger', async () => {