fix(knowledge): start the projection backfill the way the table backfill is started

The outbox event and its slice-and-defer handler are gone; nothing else in
the repo starts background work that way. The fill is started from the app
side as the table backfill is: tasks.trigger on the Trigger.dev worker when
one is configured, runDetached in-process otherwise, with the manual script
as the operator path after a deploy.

The registered migration is now 0022_projection_source_acl_backfill, which
supersedes 0021 so a database that already recorded the synchronous shape
still gains the unfilled indexes. The page statement timeout is exported so
a caller bounding a run can leave it as headroom.

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

Author: waleedlatif1

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

@@ -33,9 +33,6 @@ vi.mock('@/lib/knowledge/application/slack-search/outbox', () => ({
 vi.mock('@/lib/knowledge/documents/processing-outbox-handler', () => ({
   knowledgeDocumentProcessingOutboxHandlers: {},
 }))
-vi.mock('@/lib/knowledge/search/projection-source-acl-backfill', () => ({
-  projectionSourceAclBackfillOutboxHandlers: {},
-}))
 vi.mock('@/lib/mothership/inbox/cleanup-outbox', () => ({ inboxCleanupOutboxHandlers: {} }))
 vi.mock('@/lib/organizations/resource-cleanup', () => ({
   organizationResourceCleanupOutboxHandlers: {},

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

@@ -18,7 +18,6 @@ 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'
@@ -43,7 +42,6 @@ const handlers = {
   ...invitationMigrationOutboxHandlers,
   ...directGrantOutboxHandlers,
   ...knowledgeDocumentProcessingOutboxHandlers,
-  ...projectionSourceAclBackfillOutboxHandlers,
   ...organizationResourceCleanupOutboxHandlers,
   ...inboxCleanupOutboxHandlers,
   ...permissionAccessRequestOutboxHandlers,

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

@@ -3,55 +3,33 @@
  */
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 
-const { mockBackfill, mockEnd, mockPostgres, mockTasksTrigger, envState } = vi.hoisted(() => ({
+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' })),
-  envState: { triggerEnabled: false, secret: undefined as string | undefined },
 }))
 
 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/config/env', () => ({
-  env: {
-    get TRIGGER_SECRET_KEY() {
-      return envState.secret
-    },
+vi.mock('@/lib/core/utils/background', () => ({
+  runDetached: (_label: string, work: () => Promise<unknown>) => {
+    void work()
   },
 }))
-vi.mock('@/lib/core/config/env-flags', () => ({
-  get isTriggerDevEnabled() {
-    return envState.triggerEnabled
-  },
-}))
-vi.mock('@/lib/core/outbox/service', () => ({
-  continueOutboxHandler: (reason: string) => ({
-    outcome: 'deferred',
-    reason,
-    consumeAttempt: false,
-  }),
-  withOutboxHandlerTimeout: (handler: unknown, timeoutMs: number) =>
-    Object.assign(handler as object, { timeoutMs }),
-}))
 
 import {
   enqueueProjectionSourceAclBackfill,
-  projectionSourceAclBackfillOutboxHandlers,
   runProjectionSourceAclBackfill,
 } from '@/lib/knowledge/search/projection-source-acl-backfill'
 
 const connection = { end: mockEnd }
-const handler =
-  projectionSourceAclBackfillOutboxHandlers['knowledge.projection.source_acl.backfill']
-const context = { eventId: 'e', eventType: 'knowledge.projection.source_acl.backfill' }
 
 describe('runProjectionSourceAclBackfill', () => {
   beforeEach(() => {
@@ -112,7 +90,7 @@ describe('runProjectionSourceAclBackfill', () => {
   })
 })
 
-describe('the outbox event the migration leaves behind', () => {
+describe('enqueueProjectionSourceAclBackfill', () => {
   beforeEach(() => {
     vi.clearAllMocks()
     mockPostgres.mockReturnValue(connection)
@@ -125,40 +103,21 @@ describe('the outbox event the migration leaves behind', () => {
     })
   })
 
-  it('hands the backfill to the Trigger.dev worker when there is one', async () => {
-    envState.triggerEnabled = true
-    envState.secret = 'tr_secret'
-    await expect(enqueueProjectionSourceAclBackfill({ pageSize: 25 })).resolves.toEqual({
+  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' }
     )
-    await expect(handler({}, context as never)).resolves.toBeUndefined()
-    expect(mockTasksTrigger).toHaveBeenCalledTimes(2)
     expect(mockBackfill).not.toHaveBeenCalled()
   })
 
-  it('runs a bounded slice per outbox run without one, and stays pending until it is filled', async () => {
-    envState.triggerEnabled = false
-    envState.secret = undefined
-    mockBackfill.mockResolvedValueOnce({
-      projection: 'embedding_search',
-      scanned: 100,
-      written: 100,
-      afterId: 'chunk-100',
-      done: false,
-    })
-    await expect(handler({}, context as never)).resolves.toMatchObject({
-      outcome: 'deferred',
-      consumeAttempt: false,
-    })
+  it('fills the projections detached in this process without one', async () => {
+    await expect(enqueueProjectionSourceAclBackfill({}, false)).resolves.toBeNull()
     expect(mockTasksTrigger).not.toHaveBeenCalled()
-    expect(mockBackfill).toHaveBeenCalledTimes(1)
-    expect(mockBackfill.mock.calls[0][2].budgetMs).toBeLessThanOrEqual(handler.timeoutMs!)
-    await expect(handler({}, context as never)).resolves.toBeUndefined()
-    expect(mockBackfill).toHaveBeenCalledTimes(3)
+    await vi.waitFor(() => expect(mockBackfill).toHaveBeenCalledTimes(2))
   })
 })

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

@@ -1,7 +1,6 @@
 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'
@@ -10,13 +9,7 @@ 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 {
-  continueOutboxHandler,
-  type DeferredOutboxHandlerResult,
-  type OutboxEventContext,
-  type OutboxHandlerRegistry,
-  withOutboxHandlerTimeout,
-} from '@/lib/core/outbox/service'
+import { runDetached } from '@/lib/core/utils/background'
 
 const logger = createLogger('ProjectionSourceAclBackfill')
 
@@ -82,56 +75,24 @@ export async function runProjectionSourceAclBackfill(
   }
 }
 
-/** Whether the deployment has a Trigger.dev worker to hand the backfill to. */
-export function projectionSourceAclBackfillUsesTrigger(): boolean {
-  return Boolean(isTriggerDevEnabled && env.TRIGGER_SECRET_KEY)
-}
-
 /**
  * Starts the backfill the way the table backfill is started: on the deployment's Trigger.dev
- * worker, where bounded runs chain until both projections are filled. Safe to call again at any
- * time — a run only fills rows still unset.
+ * 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 = {}
-): Promise<{ runId: string }> {
-  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 }
-}
-
-/** One outbox run's share of the backfill on a deployment without a worker, inside its window. */
-const OUTBOX_RUN_BUDGET_MS = 4 * 60 * 1000
-const OUTBOX_HANDLER_TIMEOUT_MS = 5 * 60 * 1000
-
-/**
- * Handles the event script migration `0021_embedding_search_connector` leaves behind, so the
- * backfill starts once the app that ships this handler is up, on every deployment. With a
- * Trigger.dev worker the event is done once the task is enqueued: the task owns its own retries
- * and continuations. Without one the backfill runs here, one bounded slice per outbox run, and the
- * event stays pending until both projections are filled — every slice's writes are durable, and a
- * slice that starts over after a restart skips the filled rows through the unfilled index, so an
- * interrupted deployment loses nothing but the time of one slice.
- */
-export const projectionSourceAclBackfillOutboxHandlers: OutboxHandlerRegistry = {
-  [PROJECTION_SOURCE_ACL_BACKFILL_OUTBOX_EVENT]: withOutboxHandlerTimeout(
-    async (
-      _payload: unknown,
-      _context: OutboxEventContext
-    ): Promise<undefined | DeferredOutboxHandlerResult> => {
-      if (projectionSourceAclBackfillUsesTrigger()) {
-        await enqueueProjectionSourceAclBackfill()
-        return undefined
-      }
-      const cursor = await runProjectionSourceAclBackfill({}, { budgetMs: OUTBOX_RUN_BUDGET_MS })
-      if (!cursor) return undefined
-      return continueOutboxHandler(
-        `projection source and ACL backfill paused at ${cursor.projection} after ${cursor.afterId}`
-      )
-    },
-    OUTBOX_HANDLER_TIMEOUT_MS
-  ),
+  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/scripts/backfill-projection-source-acl.ts

@@ -2,30 +2,31 @@
 
 /**
  * Starts the projection source and ACL backfill that script migration
- * `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.
+ * `0022_projection_source_acl_backfill` 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.
  *
  * Usage:
  *   bun apps/sim/scripts/backfill-projection-source-acl.ts
  */
 
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
+import { env } from '@/lib/core/config/env'
+import { isTriggerDevEnabled } from '@/lib/core/config/env-flags'
 import {
   enqueueProjectionSourceAclBackfill,
-  projectionSourceAclBackfillUsesTrigger,
   runProjectionSourceAclBackfill,
 } from '@/lib/knowledge/search/projection-source-acl-backfill'
 
 const logger = createLogger('BackfillProjectionSourceAcl')
 
+/** A script has no long-lived process to detach into, so without a worker it fills inline. */
 async function main(): Promise<void> {
-  if (projectionSourceAclBackfillUsesTrigger()) {
-    const handle = await enqueueProjectionSourceAclBackfill()
-    logger.info('Backfill enqueued on the Trigger.dev worker', handle)
+  if (isTriggerDevEnabled && env.TRIGGER_SECRET_KEY) {
+    const handle = await enqueueProjectionSourceAclBackfill({}, true)
+    logger.info('Backfill enqueued on the Trigger.dev worker', handle ?? {})
     return
   }
   await runProjectionSourceAclBackfill({})

packages/db/script-migrations-paused-billing-attribution.test.ts

@@ -453,7 +453,7 @@ describe('script migration registry', () => {
       '0017_index_search_documents',
       '0018_repair_workspace_file_content_revision',
       '0019_tin_keyword_projection',
-      '0021_embedding_search_connector',
+      '0022_projection_source_acl_backfill',
     ])
   })
 })

packages/db/script-migrations/0016_backfill_search_vectors.postgres.test.ts

@@ -392,6 +392,7 @@ describe.runIf(Boolean(databaseUrl))('search projection upgrade in PostgreSQL',
       { name: '0018_repair_workspace_file_content_revision' },
       { name: '0019_tin_keyword_projection' },
       { name: '0021_embedding_search_connector' },
+      { name: '0022_projection_source_acl_backfill' },
     ])
     const [{ complete }] = await sql`SELECT count(*)::int AS complete FROM embedding e
       JOIN embedding_search s ON s.id = e.id JOIN embedding_keyword_search k ON k.id = e.id

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

@@ -1,7 +1,5 @@
-import {
-  backfillProjectionSourceAcl,
-  embeddingSearchConnectorMigration,
-} from '@sim/db/script-migrations/0021_embedding_search_connector'
+import { backfillProjectionSourceAcl } from '@sim/db/script-migrations/0021_embedding_search_connector'
+import { projectionSourceAclBackfillMigration as embeddingSearchConnectorMigration } from '@sim/db/script-migrations/0022_projection_source_acl_backfill'
 import { generateId } from '@sim/utils/id'
 import postgres, { type Sql } from 'postgres'
 import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest'
@@ -40,10 +38,6 @@ 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,
@@ -65,7 +59,7 @@ 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, indexes and start event again without failing, so a cut-short deploy completes', async () => {
+  it('installs its triggers and indexes 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`
@@ -78,14 +72,6 @@ describe.runIf(Boolean(databaseUrl))('projection source and ACL backfill in Post
         '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 () => {