diff --git a/apps/sim/lib/workspace-files/search/dispatcher.integration.ts b/apps/sim/lib/workspace-files/search/dispatcher.integration.ts index 6edbe58aafa..38dfe884f89 100644 --- a/apps/sim/lib/workspace-files/search/dispatcher.integration.ts +++ b/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) => 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 = () => {} diff --git a/apps/sim/lib/workspace-files/search/dispatcher.ts b/apps/sim/lib/workspace-files/search/dispatcher.ts index 3385f9242f0..50d908f82a9 100644 --- a/apps/sim/lib/workspace-files/search/dispatcher.ts +++ b/apps/sim/lib/workspace-files/search/dispatcher.ts @@ -303,7 +303,16 @@ async function reapStaleClaims(tx: DbTransaction, now: Date): Promise { 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