From ebb550ca23b1f047049c34fbca8e4377c21d42ac Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 4 Nov 2026 02:32:22 -0800 Subject: [PATCH 1/2] fix(file-search): fill dispatch capacity in selection order and reap before the backfill The claim cut ranked every selected owner's candidates by revision age, so with less capacity than candidates a woken owner selected first could lose every slot to an older backlog selected after it, and the claim's stamp then filed it behind the rotation. Claims now fill capacity owner by owner in the pass's selection order (each still up to its free slots). The backfill page ran before the reaper in the same transaction, so an owner it queued as a reconciliation wake-up absorbed the reaper's wake-up (both writes at the transaction time) and the released claim waited behind the rotation. The reaper now runs first; the backfill's wake-ups leave queued owners untouched. Tests: a two-slot pass with a woken owner against an older backlog, and a reaped claim whose owner the backfill queues in the same pass; the capacity filler's files are deleted so the backfill never re-queues it. --- apps/sim/lib/workspace-files/search/README.md | 2 +- .../search/dispatcher.integration.ts | 42 ++++++++++++++++++- .../lib/workspace-files/search/dispatcher.ts | 38 ++++++++++++----- 3 files changed, 68 insertions(+), 14 deletions(-) diff --git a/apps/sim/lib/workspace-files/search/README.md b/apps/sim/lib/workspace-files/search/README.md index 25a96fa157a..64e27ad598a 100644 --- a/apps/sim/lib/workspace-files/search/README.md +++ b/apps/sim/lib/workspace-files/search/README.md @@ -71,6 +71,6 @@ The design uses PostgreSQL's documented [TOAST behavior](https://www.postgresql. Migration `0426_file_search_owner_scope.sql` adds canonical workspace/Project ownership to the existing build, revision, and chunk pipeline. Legacy workspace inserts are translated by triggers, and the existing workspace trigram index remains usable. Project rows use a separate owner-scoped trigram index on the same chunk table. Organizations and personal users are not enabled search owners. -The new `file_search_dispatch_queue` replaces the workspace-keyed scheduler because older workers cannot decode a Project without a workspace. During expansion, workspace mutations enqueue both queues, while the new dispatcher reads the owner queue. Both deployments share the dispatcher lock, revision claim token, build lease, and publication fence. Drain old app dispatchers and Trigger workers/retries before enabling Project dispatch. Retire the legacy queue and its dual enqueue in a later contract deploy; they are not a second indexing pipeline. The `file-search-owners-v3` cursor performs bounded owner/file keyset backfill and hourly reconciliation without rewriting extracted text in the migration. Claims take owners woken since their last dispatch first (`wokenSinceDispatch` in `dispatcher.ts`); backfill wake-ups enter as if just dispatched and never touch an owner already queued, so a page of them cannot delay a file write by several passes. +The new `file_search_dispatch_queue` replaces the workspace-keyed scheduler because older workers cannot decode a Project without a workspace. During expansion, workspace mutations enqueue both queues, while the new dispatcher reads the owner queue. Both deployments share the dispatcher lock, revision claim token, build lease, and publication fence. Drain old app dispatchers and Trigger workers/retries before enabling Project dispatch. Retire the legacy queue and its dual enqueue in a later contract deploy; they are not a second indexing pipeline. The `file-search-owners-v3` cursor performs bounded owner/file keyset backfill and hourly reconciliation without rewriting extracted text in the migration. Each pass selects owners woken since their last dispatch that still have a free slot first (`wokenSinceDispatch` in `dispatcher.ts`), then the rest by dispatch age, and fills its capacity owner by owner in that order. Backfill wake-ups enter as if just dispatched and never touch an owner already queued, so a page of them rotates in behind owners woken for work. Generated document indexing never executes source. A cached artifact must match its owner and current input manifest. Each published build stores at most 500 input identities. Queries exclude a build when any input is changed, archived, deleted, or outside the owner, and report it as pending coverage. Input mutations wake their owner queue; the dispatcher retires stale manifests in bounded batches before rebuilding. Expired build cleanup also removes its dependency records. A cache miss indexes the stored generation source as text until an authorized rendering path creates a current artifact. diff --git a/apps/sim/lib/workspace-files/search/dispatcher.integration.ts b/apps/sim/lib/workspace-files/search/dispatcher.integration.ts index b56b4369a8f..454152d1653 100644 --- a/apps/sim/lib/workspace-files/search/dispatcher.integration.ts +++ b/apps/sim/lib/workspace-files/search/dispatcher.integration.ts @@ -409,10 +409,14 @@ describe('workspace file search dispatch PostgreSQL deadlines', () => { }) } - /** Fills global capacity with claims of an unqueued owner, leaving `slots` free. */ + /** + * Fills global capacity with claims of an owner that is neither queued nor walked by the + * backfill (its files are deleted), leaving `slots` free. + */ async function leaveSlots(slots: number) { await seedQueue('workspace-active', 0, FILE_SEARCH_INDEX_MAX_OUTSTANDING - slots) await connection`DELETE FROM file_search_dispatch_queue WHERE entity_id = 'workspace-active'` + await connection`UPDATE workspace_files SET deleted_at = now() WHERE workspace_id = 'workspace-active'` } const wakeUps = FILE_SEARCH_INDEX_DISPATCH_WORKSPACES * 2 + 50 @@ -469,6 +473,40 @@ describe('workspace file search dispatch PostgreSQL deadlines', () => { expect(result.payloads.map((payload) => payload.fileId)).toEqual(['waiting-workspace-000001']) }) + it('fills the free capacity in selection order, so a woken owner is not outbid by an older backlog', async () => { + await leaveSlots(2) + // Served ten minutes ago and not woken since, with older pending files than the fresh write. + await seedQueue('backlog-workspace', 3) + await connection`UPDATE file_search_dispatch_queue SET last_dispatched_at = now() - interval '10 minutes', + updated_at = now() - interval '10 minutes' WHERE entity_id = 'backlog-workspace'` + await writeFile('fresh-workspace') + + const result = await prepareWorkspaceFileSearchDispatch() + expect(result.payloads.map((payload) => payload.fileId).sort()).toEqual([ + 'backlog-workspace-000001', + 'fresh-workspace-written', + ]) + }) + + it('keeps a reaped claim woken when the backfill queues its owner in the same pass', async () => { + await leaveSlots(1) + // Its only claim was abandoned before its run was enqueued; nothing has queued its owner yet. + await seedQueue('reaped-workspace', 0, 1) + await connection`DELETE FROM file_search_dispatch_queue WHERE entity_id = 'reaped-workspace'` + await connection`UPDATE workspace_file_search_revision + SET handoff_expires_at = clock_timestamp() - interval '1 millisecond' + WHERE file_id = 'reaped-workspace-000001'` + // Served an hour ago and not woken since: it wins the single slot only if the reap is lost. + await seedQueue('waiting-workspace', 1) + await connection`UPDATE file_search_dispatch_queue SET last_dispatched_at = now() - interval '1 hour', + updated_at = now() - interval '1 hour' WHERE entity_id = 'waiting-workspace'` + await connection`UPDATE workspace_file_search_backfill SET completed_at = NULL` + + const result = await prepareWorkspaceFileSearchDispatch() + expect(result.reapedClaims).toBe(1) + expect(result.payloads.map((payload) => payload.fileId)).toEqual(['reaped-workspace-000001']) + }) + it('serves a fresh write before work the backfill found, then the backfill work', async () => { await leaveSlots(0) await connection`UPDATE workspace_file_search_backfill SET completed_at = NULL` @@ -494,7 +532,7 @@ describe('workspace file search dispatch PostgreSQL deadlines', () => { it('leaves an owner queued by a file write at the front when a backfill page passes it', async () => { await leaveSlots(1) - // Queued first and served an hour ago: it wins the single slot only if the write is demoted. + // Woken and served an hour ago: it wins the single slot only if the write is demoted. await seedQueue('earlier-served', 1) await connection`UPDATE file_search_dispatch_queue SET last_dispatched_at = now() - interval '1 hour' WHERE entity_id = 'earlier-served'` diff --git a/apps/sim/lib/workspace-files/search/dispatcher.ts b/apps/sim/lib/workspace-files/search/dispatcher.ts index 7683ec4a072..9fbfc0802e4 100644 --- a/apps/sim/lib/workspace-files/search/dispatcher.ts +++ b/apps/sim/lib/workspace-files/search/dispatcher.ts @@ -163,6 +163,10 @@ function rowOwner(row: { entityType: string; entityId: string }): EditableFileOw return resolveFileSearchOwner({ owner: row }) } +function ownerKey(owner: EditableFileOwner): string { + return JSON.stringify([owner.entityType, owner.entityId]) +} + function queuedOwner(owner: EditableFileOwner): SQL { return sql`(${fileSearchDispatchQueue.entityType} = ${owner.entityType} AND ${fileSearchDispatchQueue.entityId} = ${owner.entityId})` } @@ -180,9 +184,10 @@ const transactionNow = sql`now()` /** * Whether an owner was woken after its last dispatch: never dispatched, or touched since. A file - * write, a released claim, and a reaped claim advance `updated_at`; a claim stamps both columns - * with one time, and a reconciliation wake-up enters with them equal. A write whose transaction - * began before the claim that stamped its owner reads as earlier and waits one rotation. + * write, a released or reaped claim, and an artifact-ready requeue advance `updated_at`; a claim + * stamps both columns with one time, and a reconciliation wake-up enters with them equal. A write + * whose transaction began before the claim that stamped its owner reads as earlier, so the owner + * keeps the dispatch-age rotation instead. */ const wokenSinceDispatch = sql`(${fileSearchDispatchQueue.lastDispatchedAt} IS NULL OR ${fileSearchDispatchQueue.updatedAt} > ${fileSearchDispatchQueue.lastDispatchedAt})` @@ -211,7 +216,7 @@ async function enqueueOwners( now: Date, { reconciliation = false }: { reconciliation?: boolean } = {} ): Promise { - const unique = [...new Map(owners.map((owner) => [JSON.stringify(owner), owner])).values()] + const unique = [...new Map(owners.map((owner) => [ownerKey(owner), owner])).values()] if (unique.length === 0) return const target = [fileSearchDispatchQueue.entityType, fileSearchDispatchQueue.entityId] const insert = tx.insert(fileSearchDispatchQueue).values( @@ -355,7 +360,7 @@ async function seedBackfillPage(tx: DbTransaction, now: Date): Promise { ) .onConflictDoNothing() } - /** Every owner the page passed is woken, indexed or not, so the pass reconciles its builds. */ + // Every owner the page passed is queued, indexed or not, so a pass reconciles its builds. await enqueueOwners(tx, rows.map(rowOwner), now, { reconciliation: true }) const last = rows.at(-1) @@ -495,6 +500,11 @@ async function reconcileOwnerDependencies( } } +/** + * Claims up to each owner's free slots, filling `remainingGlobalCapacity` owner by owner in the + * order of `owners` (the pass's selection order), so a woken owner is never outbid by an older + * backlog selected after it. + */ async function claimQueuedOwnerJobs( tx: DbTransaction, owners: readonly EditableFileOwner[], @@ -503,7 +513,7 @@ async function claimQueuedOwnerJobs( ): Promise { if (owners.length === 0 || remainingGlobalCapacity <= 0) return [] const ownerValues = sql.join( - owners.map((owner) => sql`(${owner.entityType}, ${owner.entityId})`), + owners.map((owner, rank) => sql`(${owner.entityType}, ${owner.entityId}, ${rank}::int)`), sql`, ` ) const rows = await tx.execute<{ @@ -512,8 +522,8 @@ async function claimQueuedOwnerJobs( fileId: string sourceContentUpdatedAt: string }>(sql` - WITH selected_owner(entity_type, entity_id) AS (VALUES ${ownerValues}), candidates AS MATERIALIZED ( - SELECT queued.* FROM selected_owner selected + WITH selected_owner(entity_type, entity_id, claim_rank) AS (VALUES ${ownerValues}), candidates AS MATERIALIZED ( + SELECT queued.*, selected.claim_rank FROM selected_owner selected CROSS JOIN LATERAL ( SELECT count(*)::int AS active_count FROM ( SELECT 1 FROM workspace_file_search_revision active @@ -542,7 +552,7 @@ async function claimQueuedOwnerJobs( LIMIT greatest(0, ${FILE_SEARCH_INDEX_WORKSPACE_OUTSTANDING} - owner_active.active_count) FOR UPDATE OF search_index SKIP LOCKED ) queued - ORDER BY queued.updated_at, queued.entity_type, queued.entity_id, queued.file_id, queued.source_content_updated_at + ORDER BY selected.claim_rank, queued.updated_at, queued.file_id, queued.source_content_updated_at LIMIT ${remainingGlobalCapacity} ) UPDATE workspace_file_search_revision search_index SET dispatched_at = ${now.toISOString()}::timestamp, @@ -650,11 +660,13 @@ export async function prepareWorkspaceFileSearchDispatch( } const now = new Date() - const backfilledFiles = await runDispatchPhase('backfill', () => seedBackfillPage(tx, now)) + // Reaped owners are queued woken before the backfill page, whose wake-ups leave queued + // owners untouched; the other order would file a reaped claim behind them. const { reaped: reapedClaims, abandoned: abandonedClaims } = await runDispatchPhase( 'reap', () => reapStaleClaims(tx, now) ) + const backfilledFiles = await runDispatchPhase('backfill', () => seedBackfillPage(tx, now)) const [{ active, cleanupBacklogged }] = await tx.execute<{ active: number cleanupBacklogged: boolean @@ -717,7 +729,11 @@ export async function prepareWorkspaceFileSearchDispatch( .set({ lastDispatchedAt: transactionNow, updatedAt: transactionNow }) .where(queuedOwnerFilter(candidates)) } - const owners = await lockFileSearchOwners(tx, candidates, { skipBusy: true }) + const locked = new Set( + (await lockFileSearchOwners(tx, candidates, { skipBusy: true })).map(ownerKey) + ) + // Locking sorts owners by key; claims keep the selection order. + const owners = candidates.filter((owner) => locked.has(ownerKey(owner))) await reconcileOwnerDependencies(tx, owners, now) const payloads = await runDispatchPhase('claim', () => claimQueuedOwnerJobs(tx, owners, remainingGlobalCapacity, now) From 55167644241840f9724da7bf6320384f45b1c8f6 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 4 Nov 2026 02:46:40 -0800 Subject: [PATCH 2/2] fix(file-search): keep the dispatch pass's file-then-queue lock order Reaping before the backfill held the owner's queue row while the backfill took FOR SHARE on its unindexed files; a concurrent save holds its file while its trigger waits on that queue row, so the two could deadlock. The pass runs backfill then reap again, and a wake that conflicts with a row this transaction's backfill just stamped as a reconciliation clears the stamp, so a reaped claim stays ahead of the rotation. --- .../sim/lib/workspace-files/search/dispatcher.ts | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/apps/sim/lib/workspace-files/search/dispatcher.ts b/apps/sim/lib/workspace-files/search/dispatcher.ts index 9fbfc0802e4..5c3e38b9875 100644 --- a/apps/sim/lib/workspace-files/search/dispatcher.ts +++ b/apps/sim/lib/workspace-files/search/dispatcher.ts @@ -229,7 +229,15 @@ async function enqueueOwners( ) await (reconciliation ? insert.onConflictDoNothing({ target }) - : insert.onConflictDoUpdate({ target, set: { updatedAt: transactionNow } })) + : insert.onConflictDoUpdate({ + target, + set: { + updatedAt: transactionNow, + // A row this pass's backfill just queued as a reconciliation is woken as if the wake had + // queued it first; a stamp from an earlier transaction is always older than `now()`. + lastDispatchedAt: sql`CASE WHEN ${fileSearchDispatchQueue.lastDispatchedAt} = ${transactionNow} THEN NULL ELSE ${fileSearchDispatchQueue.lastDispatchedAt} END`, + }, + })) } /** @@ -660,13 +668,13 @@ export async function prepareWorkspaceFileSearchDispatch( } const now = new Date() - // Reaped owners are queued woken before the backfill page, whose wake-ups leave queued - // owners untouched; the other order would file a reaped claim behind them. + // The backfill locks files before the reap locks queue rows: a file save holds its file + // while its trigger waits on the owner's queue row, so the other order can deadlock. + const backfilledFiles = await runDispatchPhase('backfill', () => seedBackfillPage(tx, now)) const { reaped: reapedClaims, abandoned: abandonedClaims } = await runDispatchPhase( 'reap', () => reapStaleClaims(tx, now) ) - const backfilledFiles = await runDispatchPhase('backfill', () => seedBackfillPage(tx, now)) const [{ active, cleanupBacklogged }] = await tx.execute<{ active: number cleanupBacklogged: boolean