fix(file-search): recover dispatch claims abandoned between commit and enqueue (#8209)

* fix(file-search): recover dispatch claims abandoned between commit and enqueue

* chore(db): format dispatch handoff migration snapshot

Author: icecrasher321

apps/sim/lib/workspace-files/search/README.md

@@ -16,6 +16,8 @@ The chunk GIN index uses `fastupdate = off`. Each bounded insert updates the mai
 
 Indexing transactions have separate limits from search: ten seconds per statement, five seconds waiting for a lock, and thirty seconds total on PostgreSQL 17. The outer limit leaves time for ordinary statement cancellation and rollback instead of terminating the connection at the same ten-second deadline. PostgreSQL 16 uses the compatible idle-transaction guard. A canceled batch remains unpublished; the task retry starts a fresh fenced build, and cleanup retires the previous attempt. One row's direct GIN insert is not interruptible, so when storage is saturated a single ordinary chunk can run well past the statement deadline and the cancellation lands only after it; smaller batches cannot prevent that. A statement, lock, or transaction timeout is therefore treated as missing database capacity rather than a bad file: the task retries it after about 2, 4, 8, 16, and 30 minutes (with jitter), six attempts in all, so the retries outlast a slow window instead of landing inside it. Other failures keep three attempts with the runner's short default delays. A run waiting to retry still holds one of its workspace's two outstanding dispatch slots. Only a revision that exhausts its attempts is marked failed; this does not automatically retry revisions already marked failed.
 
+A dispatch claim commits before Trigger.dev accepts its run. A dispatcher that stops in between, for example one killed at its 60-second limit while PostgreSQL is still committing, leaves a claim with no run, and that claim holds one of its workspace's two slots. Each claim therefore carries a two-minute handoff deadline in PostgreSQL time, twice the dispatcher's maximum duration. The deadline is cleared once a run is known to exist: the dispatcher clears it after Trigger.dev accepts the batch, and the worker clears it when the build begins. That write skips rows another transaction holds rather than waiting on them, so it cannot deadlock with a bulk file change. The next dispatch releases a claim whose deadline has passed, logs it, and counts it in its result. The file is claimed again later under a new token, which fences out any run the old claim did get, so nothing re-sent has to be deduplicated. A claim with no deadline, including one made before the column existed, keeps the six-hour stale-dispatch recovery, which covers runs lost after they were handed off. In-process dispatch, used when Trigger.dev is not configured, clears the deadline as soon as it schedules the work in its own process, so a restart there still leaves the scheduled claims to the six-hour window.
+
 The indexing task uses an isolated `medium-2x` Trigger worker (4 GB RAM). Document parsers can materialize expanded content before chunking, so source and extracted-text byte limits do not bound parser memory. Parser complexity guards and the worker's memory budget remain separate protections.
 
 File edits, context changes, and deletion invalidate metadata and expire builds. Chunks have no cascading foreign key to files or workspaces. Cleanup locks at most 100 expired builds with `SKIP LOCKED`, deletes at most 1,000 chunks per transaction, retires empty builds in the same batch, and stops after 10 batches or five seconds. Dispatch pauses while at least 10,000 expired chunks await cleanup, so sustained revisions cannot keep admitting new builds faster than retirement can drain them. Existing ready files remain searchable. Stale workers cannot revive a reclaimed build.

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

@@ -28,7 +28,7 @@ vi.mock('@/lib/uploads/contexts/workspace', () => ({
   getWorkspaceFile: vi.fn(),
   fetchWorkspaceFileBuffer: vi.fn(),
 }))
-vi.mock('@/lib/copilot/tools/server/files/doc-compile', () => ({ resolveServableDoc: vi.fn() }))
+vi.mock('@/lib/mothership/tools/server/files/doc-compile', () => ({ resolveServableDoc: vi.fn() }))
 vi.mock('@/lib/file-parsers', () => ({ parseBuffer: vi.fn(), isSupportedFileType: vi.fn() }))
 
 import {
@@ -180,6 +180,7 @@ describe('chunked workspace file search on PostgreSQL', () => {
       '0358_workspace_file_content_version_precision.sql',
       '0359_workspace_file_search_chunks.sql',
       ginWriteMigration,
+      '0382_workspace_file_search_dispatch_handoff.sql',
     ]) {
       await applyMigration(migration)
     }
@@ -641,6 +642,20 @@ describe('chunked workspace file search on PostgreSQL', () => {
       'pending'
     )
   })
+  it('completes a claim handoff when its run begins, never for an older claim', async () => {
+    const older = new Date('2026-01-01T01:00:00Z')
+    const newer = new Date('2026-01-01T02:00:00Z')
+    await connection`UPDATE workspace_file_search_revision
+      SET dispatched_at = ${newer.toISOString()}::timestamp,
+        handoff_expires_at = clock_timestamp() + interval '2 minutes'`
+    const handoff = async () =>
+      (await connection`SELECT handoff_expires_at FROM workspace_file_search_revision`)[0]
+        .handoff_expires_at
+    expect(await beginFileSearchBuild(revision, older.toISOString())).toBeNull()
+    expect(await handoff()).not.toBeNull()
+    expect(await beginFileSearchBuild(revision, newer.toISOString())).not.toBeNull()
+    expect(await handoff()).toBeNull()
+  })
 
   async function withOccupiedPool(client: postgres.Sql, count: number, run: () => Promise<void>) {
     let release!: () => void

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

@@ -96,6 +96,14 @@ export const FILE_SEARCH_INDEX_MAX_OUTSTANDING = 100
 export const FILE_SEARCH_INDEX_DISPATCH_WORKSPACES = 100
 export const FILE_SEARCH_DISPATCH_INTERVAL_MS = 60 * 1000
 export const FILE_SEARCH_DISPATCH_MAX_DURATION_SECONDS = 60
+/**
+ * How long a claim may wait for its run to be handed off. A claim commits before Trigger.dev
+ * accepts the run, so a dispatcher that stops in between leaves a claim with no run. By twice the
+ * dispatcher task's maximum duration that dispatcher has been stopped, so the next dispatch
+ * releases the claim instead of waiting out {@link FILE_SEARCH_INDEX_STALE_DISPATCH_MS}. Anything it
+ * sent that still lands later is fenced out by the claim's token.
+ */
+export const FILE_SEARCH_DISPATCH_HANDOFF_MS = 2 * FILE_SEARCH_DISPATCH_MAX_DURATION_SECONDS * 1000
 /** Leave room for connection setup, rollback, and task failure reporting before the hard cutoff. */
 export const FILE_SEARCH_DISPATCH_STATEMENT_TIMEOUT_MS = 10 * 1000
 export const FILE_SEARCH_DISPATCH_LOCK_TIMEOUT_MS = 2 * 1000

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

@@ -25,7 +25,11 @@ 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 { FILE_SEARCH_BACKFILL_PAGE_SIZE } from '@/lib/workspace-files/search/constants'
+import {
+  FILE_SEARCH_BACKFILL_PAGE_SIZE,
+  FILE_SEARCH_DISPATCH_HANDOFF_MS,
+  FILE_SEARCH_INDEX_STALE_DISPATCH_MS,
+} from '@/lib/workspace-files/search/constants'
 import {
   dispatchWorkspaceFileSearchIndexJobs,
   prepareWorkspaceFileSearchDispatch,
@@ -80,7 +84,8 @@ describe('workspace file search dispatch PostgreSQL deadlines', () => {
       source_content_updated_at timestamp NOT NULL, status text NOT NULL DEFAULT 'pending',
       build_id text, failure_reason text, line_count integer NOT NULL DEFAULT 0,
       indexed_bytes integer NOT NULL DEFAULT 0, chunk_count integer NOT NULL DEFAULT 0,
-      dispatched_at timestamp, updated_at timestamp NOT NULL DEFAULT now()
+      dispatched_at timestamp, handoff_expires_at timestamp,
+      updated_at timestamp NOT NULL DEFAULT now()
     )`
     await connection`CREATE TABLE workspace_file_search_dispatch_queue (
       workspace_id text PRIMARY KEY, enqueued_at timestamp NOT NULL,
@@ -297,6 +302,155 @@ describe('workspace file search dispatch PostgreSQL deadlines', () => {
     expect(plan).not.toMatch(/Sort Key: search_index(_\d+)?\.updated_at/)
   }, 30_000)
 
+  it('releases a claim abandoned before its enqueue once its handoff expires', async () => {
+    await seedQueue('workspace-1', 3)
+    /** Preparing without enqueueing is a dispatcher stopped between its commit and its enqueue. */
+    const abandoned = await prepareWorkspaceFileSearchDispatch()
+    expect(abandoned.payloads).toHaveLength(2)
+    const deadlines = await connection`SELECT
+      extract(epoch FROM handoff_expires_at - clock_timestamp()) * 1000 AS remaining_ms
+      FROM workspace_file_search_revision WHERE dispatched_at IS NOT NULL`
+    expect(deadlines).toHaveLength(2)
+    for (const { remaining_ms } of deadlines) {
+      expect(Number(remaining_ms)).toBeGreaterThan(FILE_SEARCH_DISPATCH_HANDOFF_MS - 10_000)
+      expect(Number(remaining_ms)).toBeLessThanOrEqual(FILE_SEARCH_DISPATCH_HANDOFF_MS)
+    }
+    /** Until the deadline passes, the claims keep holding their workspace's slots. */
+    expect(await prepareWorkspaceFileSearchDispatch()).toMatchObject({
+      payloads: [],
+      reapedClaims: 0,
+      abandonedClaims: 0,
+    })
+
+    await connection`UPDATE workspace_file_search_revision
+      SET handoff_expires_at = clock_timestamp() - interval '1 millisecond'
+      WHERE dispatched_at IS NOT NULL`
+    const recovered = await prepareWorkspaceFileSearchDispatch()
+    expect(recovered.reapedClaims).toBe(2)
+    expect(recovered.abandonedClaims).toBe(2)
+    expect(recovered.payloads).toHaveLength(2)
+    const abandonedToken = abandoned.payloads[0].dispatchToken
+    if (!abandonedToken) throw new Error('Every claim carries its dispatch token')
+    expect(recovered.payloads.map((payload) => payload.dispatchToken)).not.toContain(abandonedToken)
+    const [left] = await connection`SELECT count(*)::int AS claims
+      FROM workspace_file_search_revision WHERE dispatched_at = ${abandonedToken}::timestamp`
+    expect(left.claims).toBe(0)
+    const [unclaimed] = await connection`SELECT count(*)::int AS deadlines
+      FROM workspace_file_search_revision
+      WHERE dispatched_at IS NULL AND handoff_expires_at IS NOT NULL`
+    expect(unclaimed.deadlines).toBe(0)
+  })
+
+  it('leaves an enqueued claim to its run until the stale-dispatch window', async () => {
+    await seedQueue('workspace-1', 1)
+    /** A millisecond revision, as file writes store, so the handoff must match it exactly. */
+    await connection`UPDATE workspace_files SET content_updated_at = '2026-09-16 12:34:56.789'`
+    await connection`UPDATE workspace_file_search_revision
+      SET source_content_updated_at = '2026-09-16 12:34:56.789'`
+    mocks.batchTrigger.mockResolvedValueOnce({ batchId: 'batch-1' })
+    await expect(dispatchWorkspaceFileSearchIndexJobs()).resolves.toMatchObject({
+      dispatchedFiles: 1,
+    })
+    const [claim] = await connection`SELECT handoff_expires_at FROM workspace_file_search_revision
+      WHERE dispatched_at IS NOT NULL`
+    expect(claim.handoff_expires_at).toBeNull()
+
+    /** A run can wait in its queue or back off for an hour without being taken for abandoned. */
+    await connection`UPDATE workspace_file_search_revision
+      SET dispatched_at = dispatched_at - interval '1 hour' WHERE dispatched_at IS NOT NULL`
+    expect((await prepareWorkspaceFileSearchDispatch()).reapedClaims).toBe(0)
+    await connection`UPDATE workspace_file_search_revision
+      SET dispatched_at = dispatched_at - ${FILE_SEARCH_INDEX_STALE_DISPATCH_MS} * interval '1 millisecond'
+      WHERE dispatched_at IS NOT NULL`
+    expect(await prepareWorkspaceFileSearchDispatch()).toMatchObject({
+      reapedClaims: 1,
+      abandonedClaims: 0,
+    })
+  })
+
+  it('does not complete the handoff of a claim released and claimed again meanwhile', async () => {
+    await seedQueue('workspace-1', 1)
+    mocks.batchTrigger.mockImplementationOnce(async () => {
+      /** Another dispatch releases this claim and claims the revision again under its own token. */
+      await connection`UPDATE workspace_file_search_revision
+        SET dispatched_at = '2099-01-01', handoff_expires_at = '2099-01-01'
+        WHERE dispatched_at IS NOT NULL`
+      return { batchId: 'batch-1' }
+    })
+
+    await dispatchWorkspaceFileSearchIndexJobs()
+
+    const [claim] = await connection`SELECT handoff_expires_at::text AS handoff
+      FROM workspace_file_search_revision WHERE dispatched_at IS NOT NULL`
+    expect(claim.handoff).toBe('2099-01-01 00:00:00')
+  })
+
+  it('records the handoff around a claim another transaction holds instead of waiting', async () => {
+    await seedQueue('workspace-1', 2)
+    let release = () => {}
+    const held = new Promise<void>((resolve) => {
+      release = resolve
+    })
+    let holder: Promise<unknown> | undefined
+    mocks.batchTrigger.mockImplementationOnce(async () => {
+      /** A run beginning its build holds its claim while the handoff is being recorded. */
+      let locked = () => {}
+      const lockReady = new Promise<void>((resolve) => {
+        locked = resolve
+      })
+      holder = connection.begin(async (tx) => {
+        await tx`SELECT file_id FROM workspace_file_search_revision
+          WHERE file_id = 'workspace-1-000001' FOR UPDATE`
+        locked()
+        await held
+      })
+      await lockReady
+      return { batchId: 'batch-1' }
+    })
+    try {
+      await expect(dispatchWorkspaceFileSearchIndexJobs()).resolves.toMatchObject({
+        dispatchedFiles: 2,
+      })
+      const claims = await connection`SELECT file_id,
+        handoff_expires_at IS NOT NULL AS pending_handoff
+        FROM workspace_file_search_revision ORDER BY file_id`
+      expect([...claims]).toEqual([
+        { file_id: 'workspace-1-000001', pending_handoff: true },
+        { file_id: 'workspace-1-000002', pending_handoff: false },
+      ])
+    } finally {
+      release()
+      await holder
+    }
+  })
+
+  it('keeps enqueued claims when recording their handoff fails', async () => {
+    await seedQueue('workspace-1', 1)
+    await connection`CREATE FUNCTION reject_handoff() RETURNS trigger LANGUAGE plpgsql AS $$
+      BEGIN
+        RAISE EXCEPTION 'handoff unavailable';
+      END
+    $$`
+    await connection`CREATE TRIGGER reject_handoff BEFORE UPDATE OF handoff_expires_at
+      ON workspace_file_search_revision FOR EACH ROW
+      WHEN (OLD.handoff_expires_at IS NOT NULL AND NEW.handoff_expires_at IS NULL
+        AND NEW.dispatched_at IS NOT NULL)
+      EXECUTE FUNCTION reject_handoff()`
+    try {
+      mocks.batchTrigger.mockResolvedValueOnce({ batchId: 'batch-1' })
+      await expect(dispatchWorkspaceFileSearchIndexJobs()).resolves.toMatchObject({
+        dispatchedFiles: 1,
+      })
+      const [claim] = await connection`SELECT dispatched_at, handoff_expires_at
+        FROM workspace_file_search_revision`
+      expect(claim.dispatched_at).not.toBeNull()
+      expect(claim.handoff_expires_at).not.toBeNull()
+    } finally {
+      await connection`DROP TRIGGER reject_handoff ON workspace_file_search_revision`
+      await connection`DROP FUNCTION reject_handoff()`
+    }
+  })
+
   it('fails on a locked backfill row and releases the dispatcher lock', async () => {
     let release = () => {}
     let locked = () => {}
@@ -385,9 +539,11 @@ describe('workspace file search dispatch PostgreSQL deadlines', () => {
         },
       }),
     ])
-    const [index] = await connection`SELECT dispatched_at FROM workspace_file_search_revision
+    const [index] =
+      await connection`SELECT dispatched_at, handoff_expires_at FROM workspace_file_search_revision
       WHERE file_id = ${fileId}`
     expect(index.dispatched_at).toBeNull()
+    expect(index.handoff_expires_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)