fix(workspace-files): keep the file-search claim on the ordered pending index instead of a hash join (#8128)

Author: waleedlatif1

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

@@ -89,7 +89,8 @@ describe('workspace file search dispatch PostgreSQL deadlines', () => {
     await connection`CREATE INDEX workspace_files_workspace_active_keyset_idx
       ON workspace_files (workspace_id, id)
       WHERE deleted_at IS NULL AND context = 'workspace' AND workspace_id IS NOT NULL`
-    await connection`CREATE INDEX ON workspace_file_search_revision
+    await connection`CREATE INDEX workspace_file_search_revision_pending_idx
+      ON workspace_file_search_revision
       (workspace_id, updated_at, file_id, source_content_updated_at)
       WHERE status = 'pending' AND dispatched_at IS NULL`
     await connection`CREATE INDEX ON workspace_file_search_revision (workspace_id, dispatched_at)
@@ -263,6 +264,39 @@ describe('workspace file search dispatch PostgreSQL deadlines', () => {
     expect(plan).toMatch(/Index Cond:.*ROW\(/)
   }, 30_000)
 
+  it('claims from the ordered pending index rather than a hash join over the backlog', async () => {
+    await seedQueue('workspace-1', 10_000)
+    await connection`ANALYZE workspace_files`
+    await connection`ANALYZE workspace_file_search_revision`
+
+    statements.length = 0
+    await prepareWorkspaceFileSearchDispatch()
+
+    const claim = statements.find((statement) =>
+      statement.query.includes('FOR UPDATE OF search_index SKIP LOCKED')
+    )
+    expect(claim).toBeDefined()
+
+    const plan = await connection.begin(async (tx) => {
+      /**
+       * At fixture scale the planner already nests the join, so it is pinned to the choice
+       * production makes when it misestimates the timestamp equi-join under `FOR UPDATE`. A join
+       * spelling then hashes every file and every pending revision of the workspace and sorts the
+       * whole backlog before the top-N cut; the correlated LATERAL cannot be flattened into that
+       * join, so the claim stays an ordered walk of the pending index.
+       */
+      await tx`SET LOCAL enable_nestloop = off`
+      const rows = await tx.unsafe(`EXPLAIN ${claim?.query}`, claim?.params as never[])
+      return rows.map((row: Record<string, unknown>) => row['QUERY PLAN']).join('\n')
+    })
+
+    /** The locked candidate scan must be fed by the ordered index walk, not a sorted hash join. */
+    expect(plan).toMatch(
+      /LockRows[^\n]*\n\s*-> {2}Nested Loop[^\n]*\n\s*-> {2}Index Scan using workspace_file_search_revision_pending_idx/
+    )
+    expect(plan).not.toMatch(/Sort Key: search_index(_\d+)?\.updated_at/)
+  }, 30_000)
+
   it('fails on a locked backfill row and releases the dispatcher lock', async () => {
     let release = () => {}
     let locked = () => {}

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

@@ -303,7 +303,16 @@ async function reapStaleClaims(tx: DbTransaction, now: Date): Promise<number> {
   return rows.length
 }
 
-/** Probe each workspace's available slots and lock candidates before the update, skipping busy rows. */
+/**
+ * Probe each workspace's available slots and lock candidates before the update, skipping busy rows.
+ *
+ * The live-file check is a correlated LATERAL with `LIMIT 1` rather than a join. `FOR UPDATE`
+ * forbids parallel plans and the planner estimates the timestamp equi-join at about one row, so
+ * a join becomes a hash join over every file and every pending revision of the workspace before
+ * the top-N sort, which exceeds the statement timeout on a large backlog. A LATERAL with LIMIT
+ * cannot be flattened into that join, so the claim stays an ordered walk of the pending index
+ * that stops after the batch size.
+ */
 async function claimQueuedWorkspaceJobs(
   tx: DbTransaction,
   workspaceIds: readonly string[],
@@ -339,12 +348,15 @@ async function claimQueuedWorkspaceJobs(
         SELECT search_index.workspace_id, search_index.file_id,
           search_index.source_content_updated_at, search_index.updated_at
         FROM workspace_file_search_revision AS search_index
-        INNER JOIN workspace_files AS file
-          ON file.id = search_index.file_id
-          AND file.workspace_id = search_index.workspace_id
-          AND file.context = 'workspace'
-          AND file.deleted_at IS NULL
-          AND file.content_updated_at = search_index.source_content_updated_at
+        CROSS JOIN LATERAL (
+          SELECT 1 FROM workspace_files AS file
+          WHERE file.id = search_index.file_id
+            AND file.workspace_id = search_index.workspace_id
+            AND file.context = 'workspace'
+            AND file.deleted_at IS NULL
+            AND file.content_updated_at = search_index.source_content_updated_at
+          LIMIT 1
+        ) AS live_file
         WHERE search_index.workspace_id = selected.workspace_id
           AND search_index.status = 'pending' AND search_index.dispatched_at IS NULL
         ORDER BY search_index.updated_at, search_index.file_id, search_index.source_content_updated_at