fix(file-search): preserve compatible dispatch cleanup

Author: icecrasher321

.github/workflows/test-build.yml

@@ -134,6 +134,23 @@ jobs:
           BILLING_USAGE_TEST_DATABASE_URL: postgresql://postgres:postgres@127.0.0.1:5433/sim_billing_test
         run: bunx vitest run lib/billing/core/usage-log.postgres.test.ts
 
+      - name: Verify file search dispatch deadlines on PostgreSQL 17
+        working-directory: apps/sim
+        env:
+          TZ: America/Los_Angeles
+          KNOWLEDGE_ACL_TEST_DATABASE_URL: postgresql://postgres:postgres@127.0.0.1:5432/sim_auth_scim
+        run: bunx vitest run --mode integration lib/workspace-files/search/dispatcher.integration.ts
+
+      - name: Verify file search dispatch deadlines on PostgreSQL 16
+        if: matrix.provision == 'push'
+        working-directory: apps/sim
+        env:
+          TZ: America/Los_Angeles
+          KNOWLEDGE_ACL_TEST_DATABASE_URL: postgresql://postgres:postgres@127.0.0.1:5433/sim_acl_test_dispatch
+        run: |
+          bun -e 'import postgres from "postgres"; const url = new URL(process.env.KNOWLEDGE_ACL_TEST_DATABASE_URL); const database = url.pathname.slice(1); url.pathname = "/postgres"; const sql = postgres(url.toString()); await sql`CREATE DATABASE ${sql(database)}`; await sql.end()'
+          bunx vitest run --mode integration lib/workspace-files/search/dispatcher.integration.ts
+
       - name: Verify SCIM and administration over real HTTP
         working-directory: apps/sim
         env:

apps/sim/lib/workspace-files/search/dispatcher.integration.ts

@@ -1,12 +1,14 @@
 /** Real PostgreSQL cancellation must roll back preparation and release its advisory lock. */
 import { withUtcTimestamps } from '@sim/db/timestamps'
 import { getPostgresErrorCode } from '@sim/utils/errors'
+import { sleep } from '@sim/utils/helpers'
 import { generateId } from '@sim/utils/id'
 import { drizzle, type PostgresJsDatabase } from 'drizzle-orm/postgres-js'
 import postgres from 'postgres'
-import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'
+import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
 
 const database = vi.hoisted(() => ({ current: undefined as PostgresJsDatabase | undefined }))
+const mocks = vi.hoisted(() => ({ batchTrigger: vi.fn() }))
 vi.mock('@sim/db', () => ({
   get db() {
     if (!database.current) throw new Error('Dispatcher test database is not initialized')
@@ -17,8 +19,14 @@ vi.mock('@/lib/workspace-files/search/indexing', () => ({
   indexWorkspaceFileForSearch: vi.fn(),
   markWorkspaceFileSearchIndexFailed: vi.fn(),
 }))
+vi.mock('@trigger.dev/sdk', () => ({ tasks: { batchTrigger: mocks.batchTrigger } }))
+vi.mock('@/lib/core/config/env-flags', () => ({ isTriggerDevEnabled: true }))
+vi.mock('@/lib/core/async-jobs/region', () => ({ resolveTriggerRegion: async () => 'us-east-1' }))
 
-import { prepareWorkspaceFileSearchDispatch } from '@/lib/workspace-files/search/dispatcher'
+import {
+  dispatchWorkspaceFileSearchIndexJobs,
+  prepareWorkspaceFileSearchDispatch,
+} from '@/lib/workspace-files/search/dispatcher'
 
 describe('workspace file search dispatch PostgreSQL deadlines', () => {
   const schemaName = `dispatch_test_${generateId().replaceAll('-', '')}`
@@ -41,11 +49,32 @@ describe('workspace file search dispatch PostgreSQL deadlines', () => {
       id text PRIMARY KEY, after_workspace_id text, after_file_id text,
       completed_at timestamp, updated_at timestamp NOT NULL
     )`
+    await connection`CREATE TABLE workspace_files (
+      id text PRIMARY KEY, workspace_id text NOT NULL, context text NOT NULL,
+      deleted_at timestamp, content_updated_at timestamp NOT NULL
+    )`
+    await connection`CREATE TABLE workspace_file_search_index (
+      file_id text NOT NULL, workspace_id text NOT NULL, source_content_updated_at timestamp NOT NULL,
+      status text NOT NULL, dispatched_at timestamp, updated_at timestamp NOT NULL,
+      PRIMARY KEY (file_id, source_content_updated_at)
+    )`
+    await connection`CREATE TABLE workspace_file_search_dispatch_queue (
+      workspace_id text PRIMARY KEY, enqueued_at timestamp NOT NULL,
+      updated_at timestamp NOT NULL, last_dispatched_at timestamp
+    )`
     await connection`INSERT INTO workspace_file_search_backfill (id, updated_at)
       VALUES ('workspace-file-search-v1', '2026-09-16 00:00:00')`
     database.current = drizzle(connection)
   })
 
+  beforeEach(async () => {
+    mocks.batchTrigger.mockReset()
+    await connection`DROP TRIGGER IF EXISTS slow_backfill ON workspace_file_search_backfill`
+    await connection`TRUNCATE workspace_files, workspace_file_search_index, workspace_file_search_dispatch_queue`
+    await connection`UPDATE workspace_file_search_backfill
+      SET updated_at = '2026-09-16 00:00:00', completed_at = NULL`
+  })
+
   afterAll(async () => {
     try {
       await connection`DROP SCHEMA ${connection(schemaName)} CASCADE`
@@ -108,4 +137,55 @@ describe('workspace file search dispatch PostgreSQL deadlines', () => {
     expect(row.updated_at).toBe('2026-09-16 00:00:00')
     await expectAdvisoryLockReleased()
   }, 20_000)
+
+  it('releases committed claims after contention exceeds the preparation lock deadline', async () => {
+    const fileId = generateId()
+    const workspaceId = generateId()
+    await connection`UPDATE workspace_file_search_backfill SET completed_at = now()`
+    await connection`INSERT INTO workspace_files (id, workspace_id, context, content_updated_at)
+      VALUES (${fileId}, ${workspaceId}, 'workspace', '2026-09-16')`
+    await connection`INSERT INTO workspace_file_search_index
+      (file_id, workspace_id, source_content_updated_at, status, updated_at)
+      VALUES (${fileId}, ${workspaceId}, '2026-09-16', 'pending', now())`
+    await connection`INSERT INTO workspace_file_search_dispatch_queue
+      (workspace_id, enqueued_at, updated_at) VALUES (${workspaceId}, now(), now())`
+
+    const enqueueError = new Error('Queue unavailable')
+    let blocker: Promise<unknown> | undefined
+    mocks.batchTrigger.mockImplementationOnce(async () => {
+      let locked = () => {}
+      const lockReady = new Promise<void>((resolve) => {
+        locked = resolve
+      })
+      blocker = connection.begin(async (tx) => {
+        await tx`SELECT file_id FROM workspace_file_search_index WHERE file_id = ${fileId} FOR UPDATE`
+        locked()
+        await sleep(3_000)
+      })
+      await Promise.race([lockReady, blocker])
+      throw enqueueError
+    })
+
+    try {
+      await expect(dispatchWorkspaceFileSearchIndexJobs()).rejects.toBe(enqueueError)
+      expect(mocks.batchTrigger).toHaveBeenCalledWith('workspace-file-search-index', [
+        expect.objectContaining({
+          payload: {
+            fileId,
+            workspaceId,
+            sourceContentUpdatedAt: '2026-09-16T00:00:00.000Z',
+          },
+        }),
+      ])
+      const [index] = await connection`SELECT dispatched_at FROM workspace_file_search_index
+        WHERE file_id = ${fileId}`
+      expect(index.dispatched_at).toBeNull()
+      const [queued] =
+        await connection`SELECT workspace_id FROM workspace_file_search_dispatch_queue
+        WHERE workspace_id = ${workspaceId}`
+      expect(queued.workspace_id).toBe(workspaceId)
+    } finally {
+      await blocker
+    }
+  })
 })

apps/sim/lib/workspace-files/search/dispatcher.test.ts

@@ -90,9 +90,10 @@ describe('workspace file search dispatch deadlines', () => {
     expect(guards).toContain('10000ms')
     expect(guards).toContain("set_config('lock_timeout', ")
     expect(guards).toContain('2000ms')
-    expect(guards).toContain("set_config('transaction_timeout', ")
+    expect(guards).toContain("current_setting('transaction_timeout', true) is null")
+    expect(guards).toContain("then 'idle_in_transaction_session_timeout'")
+    expect(guards).toContain("else 'transaction_timeout'")
     expect(guards).toContain('20000ms')
-    expect(guards.match(/, true\)/g)).toHaveLength(3)
     expect(JSON.stringify(dbChainMockFns.execute.mock.calls[1][0])).toContain(
       'pg_try_advisory_xact_lock'
     )
@@ -139,31 +140,49 @@ describe('workspace file search dispatch deadlines', () => {
     })
   })
 
-  it('also bounds releasing committed claims when batch submission fails', async () => {
-    queueTableRows(workspaceFileSearchBackfill, [{ completedAt: new Date() }])
-    queueTableRows(workspaceFileSearchIndex, [])
-    queueTableRows(workspaceFileSearchIndex, [{ active: 0 }])
-    queueTableRows(workspaceFileSearchDispatchQueue, [{ workspaceId: 'workspace-1' }])
-    dbChainMockFns.execute
-      .mockResolvedValueOnce([])
-      .mockResolvedValueOnce([{ acquired: true }])
-      .mockResolvedValueOnce([
-        {
-          workspaceId: 'workspace-1',
-          fileId: 'file-1',
-          sourceContentUpdatedAt: new Date('2026-09-16T00:00:00Z'),
-        },
-      ])
-    const error = new Error('Trigger unavailable')
-    mocks.batchTrigger.mockRejectedValueOnce(error)
-
-    await expect(dispatchWorkspaceFileSearchIndexJobs()).rejects.toBe(error)
-
-    expect(dbChainMockFns.transaction).toHaveBeenCalledTimes(2)
-    const guards = dbChainMockFns.execute.mock.calls.filter(([query]) =>
-      JSON.stringify(query).includes('statement_timeout')
-    )
-    expect(guards).toHaveLength(2)
-    expect(dbChainMockFns.set).toHaveBeenCalledWith(expect.objectContaining({ dispatchedAt: null }))
-  })
+  it.each([false, true])(
+    'preserves enqueue failures when claim release fails: %s',
+    async (releaseFails) => {
+      queueTableRows(workspaceFileSearchBackfill, [{ completedAt: new Date() }])
+      queueTableRows(workspaceFileSearchIndex, [])
+      queueTableRows(workspaceFileSearchIndex, [{ active: 0 }])
+      queueTableRows(workspaceFileSearchDispatchQueue, [{ workspaceId: 'workspace-1' }])
+      dbChainMockFns.execute
+        .mockResolvedValueOnce([])
+        .mockResolvedValueOnce([{ acquired: true }])
+        .mockResolvedValueOnce([
+          {
+            workspaceId: 'workspace-1',
+            fileId: 'file-1',
+            sourceContentUpdatedAt: new Date('2026-09-16T00:00:00Z'),
+          },
+        ])
+      const error = new Error('Trigger unavailable')
+      const releaseError = new Error('claim release unavailable')
+      mocks.batchTrigger.mockRejectedValueOnce(error)
+      if (releaseFails) {
+        dbChainMockFns.transaction
+          .mockImplementationOnce(async (callback) => callback(dbChainMock.db))
+          .mockRejectedValueOnce(releaseError)
+      }
+
+      if (releaseFails) {
+        await expect(dispatchWorkspaceFileSearchIndexJobs()).rejects.toMatchObject({
+          errors: [error, releaseError],
+          cause: error,
+        })
+      } else {
+        await expect(dispatchWorkspaceFileSearchIndexJobs()).rejects.toBe(error)
+        expect(dbChainMockFns.set).toHaveBeenCalledWith(
+          expect.objectContaining({ dispatchedAt: null })
+        )
+      }
+
+      expect(dbChainMockFns.transaction).toHaveBeenCalledTimes(2)
+      const guards = dbChainMockFns.execute.mock.calls.filter(([query]) =>
+        JSON.stringify(query).includes('statement_timeout')
+      )
+      expect(guards).toHaveLength(1)
+    }
+  )
 })

apps/sim/lib/workspace-files/search/dispatcher.ts

@@ -72,13 +72,20 @@ async function runDispatchPhase<T>(phase: string, operation: () => Promise<T>):
   }
 }
 
-/** Transaction-local guards also cover claim release; a stalled query must roll back before retry. */
+/** Preparation must roll back before retry; PostgreSQL 16 falls back to an idle transaction guard. */
 async function configureDispatchTimeouts(tx: DbTransaction): Promise<void> {
   await tx.execute(sql`
     SELECT
       set_config('statement_timeout', ${`${FILE_SEARCH_DISPATCH_STATEMENT_TIMEOUT_MS}ms`}, true),
       set_config('lock_timeout', ${`${FILE_SEARCH_DISPATCH_LOCK_TIMEOUT_MS}ms`}, true),
-      set_config('transaction_timeout', ${`${FILE_SEARCH_DISPATCH_TRANSACTION_TIMEOUT_MS}ms`}, true)
+      set_config(
+        case when current_setting('transaction_timeout', true) is null
+          then 'idle_in_transaction_session_timeout'
+          else 'transaction_timeout'
+        end,
+        ${`${FILE_SEARCH_DISPATCH_TRANSACTION_TIMEOUT_MS}ms`},
+        true
+      )
   `)
 }
 
@@ -311,7 +318,7 @@ async function claimQueuedWorkspaceJobs(
   const rows = await tx.execute<{
     workspaceId: string
     fileId: string
-    sourceContentUpdatedAt: Date
+    sourceContentUpdatedAt: string
   }>(sql`
     WITH selected_workspace(workspace_id) AS (
       VALUES ${workspaceValues}
@@ -370,7 +377,7 @@ async function claimQueuedWorkspaceJobs(
     RETURNING
       search_index.workspace_id AS "workspaceId",
       search_index.file_id AS "fileId",
-      search_index.source_content_updated_at AS "sourceContentUpdatedAt"
+      search_index.source_content_updated_at AT TIME ZONE 'UTC' AS "sourceContentUpdatedAt"
   `)
 
   const remainingForWorkspace = tx
@@ -484,7 +491,6 @@ async function releaseDispatchClaims(payloads: readonly WorkspaceFileSearchIndex
   }))
   await runDispatchPhase('release-claims', () =>
     db.transaction(async (tx) => {
-      await configureDispatchTimeouts(tx)
       const filter = revisionFilter(rows)
       if (filter) {
         await tx
@@ -555,11 +561,21 @@ export async function dispatchWorkspaceFileSearchIndexJobs(): Promise<WorkspaceF
       lockAcquired: prepared.lockAcquired,
     }
   } catch (error) {
-    await releaseDispatchClaims(prepared.payloads)
     logger.error('Failed to dispatch workspace file search indexing batch', {
       files: prepared.payloads.length,
-      error: getErrorMessage(error),
+      error: truncate(getErrorMessage(error).split('\nparams: ')[0], 500),
     })
+    try {
+      await releaseDispatchClaims(prepared.payloads)
+    } catch (releaseError) {
+      throw new AggregateError(
+        [error, releaseError],
+        'File search enqueue and claim release failed',
+        {
+          cause: error,
+        }
+      )
+    }
     throw error
   }
 }