fix(knowledge): keep the outbox event pending until a deployment without a worker fills the projections

Without a Trigger.dev worker the outbox handler no longer detaches the
backfill and completes the event; it runs one bounded slice per outbox run
and yields with continueOutboxHandler until both projections are filled, so
a restart loses at most one slice and the event is never marked done ahead
of the work. The index builds run on one reserved connection, so the
session-scoped lock timeout covers every build and its reset.

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

Author: waleedlatif1

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

@@ -33,6 +33,9 @@ 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/knowledge/search/projection-source-acl-backfill.test.ts

@@ -3,11 +3,12 @@
  */
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 
-const { mockBackfill, mockEnd, mockPostgres, mockTasksTrigger } = vi.hoisted(() => ({
+const { mockBackfill, mockEnd, mockPostgres, mockTasksTrigger, envState } = 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' }))
@@ -19,11 +20,27 @@ vi.mock('@sim/db/script-migrations/0021_embedding_search_connector', () => ({
 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()
+vi.mock('@/lib/core/config/env', () => ({
+  env: {
+    get TRIGGER_SECRET_KEY() {
+      return envState.secret
+    },
   },
 }))
+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,
@@ -32,6 +49,9 @@ import {
 } 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(() => {
@@ -92,7 +112,7 @@ describe('runProjectionSourceAclBackfill', () => {
   })
 })
 
-describe('enqueueProjectionSourceAclBackfill', () => {
+describe('the outbox event the migration leaves behind', () => {
   beforeEach(() => {
     vi.clearAllMocks()
     mockPostgres.mockReturnValue(connection)
@@ -106,26 +126,39 @@ describe('enqueueProjectionSourceAclBackfill', () => {
   })
 
   it('hands the backfill to the Trigger.dev worker when there is one', async () => {
-    await expect(enqueueProjectionSourceAclBackfill({ pageSize: 25 }, true)).resolves.toEqual({
+    envState.triggerEnabled = true
+    envState.secret = 'tr_secret'
+    await expect(enqueueProjectionSourceAclBackfill({ pageSize: 25 })).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('fills the projections detached in this process without one', async () => {
-    await expect(enqueueProjectionSourceAclBackfill({}, false)).resolves.toBeNull()
+  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,
+    })
     expect(mockTasksTrigger).not.toHaveBeenCalled()
-    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',
-    ])
+    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)
   })
 })

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

@@ -10,8 +10,13 @@ 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'
+import {
+  continueOutboxHandler,
+  type DeferredOutboxHandlerResult,
+  type OutboxEventContext,
+  type OutboxHandlerRegistry,
+  withOutboxHandlerTimeout,
+} from '@/lib/core/outbox/service'
 
 const logger = createLogger('ProjectionSourceAclBackfill')
 
@@ -77,34 +82,56 @@ 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 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.
+ * worker, where bounded runs chain until both projections are filled. 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
+  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
+
 /**
- * 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.
+ * 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]: async () => {
-    await enqueueProjectionSourceAclBackfill()
-  },
+  [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
+  ),
 }

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

@@ -14,19 +14,18 @@
 
 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')
 
 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 ?? {})
+  if (projectionSourceAclBackfillUsesTrigger()) {
+    const handle = await enqueueProjectionSourceAclBackfill()
+    logger.info('Backfill enqueued on the Trigger.dev worker', handle)
     return
   }
   await runProjectionSourceAclBackfill({})

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

@@ -201,14 +201,17 @@ export async function backfillProjectionSourceAcl(
  * vector projection the source index that lets the planner lead with a few sources when the
  * caller's tokens alone would match most of the index. Built concurrently, so the triggers and the
  * backfill keep writing. `CONCURRENTLY` cannot run in a transaction, and the pool's lock timeout
- * would cancel a build that merely waits for a long transaction to finish.
+ * would cancel a build that merely waits for a long transaction to finish. The timeout is a
+ * session setting, so one connection is reserved for it, the builds, and the reset — a pool
+ * would otherwise hand the builds to connections that never saw the setting.
  *
  * The unfilled index on each projection lists the rows the backfill has not reached: each page
  * reads its rows from it instead of walking past every filled one, and the on-row predicate's
  * unfilled branch, an `OR` beside the ACL overlap, stays an index probe for the planner — once the
  * projection is filled, a probe of an empty index.
  */
-export async function indexProjectionAcl(sql: Sql): Promise<void> {
+export async function indexProjectionAcl(pool: Sql): Promise<void> {
+  const sql = await pool.reserve()
   const [{ timeout }] = await sql`SELECT current_setting('lock_timeout') AS timeout`
   await sql.unsafe('SET lock_timeout = 0')
   try {
@@ -237,14 +240,15 @@ export async function indexProjectionAcl(sql: Sql): Promise<void> {
     for (const projection of PROJECTION_SOURCE_ACL_TABLES) await sql.unsafe(`ANALYZE ${projection}`)
   } finally {
     await sql`SELECT set_config('lock_timeout', ${timeout}, false)`
+    sql.release()
   }
 }
 
 /**
  * Leaves the app one outbox event to start the backfill from. The outbox processor runs on every
  * deployment, so the backfill starts on its own once the app that ships the handler is up —
- * on the Trigger.dev worker where there is one, detached in the app otherwise — without an
- * operator remembering to. Idempotent under its fixed id.
+ * enqueued on the Trigger.dev worker where there is one, run in bounded slices by the outbox
+ * itself otherwise — without an operator remembering to. Idempotent under its fixed id.
  */
 export async function enqueueProjectionSourceAclBackfillEvent(sql: Sql): Promise<void> {
   await sql`