fix(file-search): fill dispatch capacity in selection order and keep reaped claims ahead of the rotation

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.

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'`

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<void> {
-  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(
@@ -224,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`,
+        },
+      }))
 }
 
 /**
@@ -355,7 +368,7 @@ async function seedBackfillPage(tx: DbTransaction, now: Date): Promise<number> {
       )
       .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 +508,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 +521,7 @@ async function claimQueuedOwnerJobs(
 ): Promise<WorkspaceFileSearchIndexPayload[]> {
   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 +530,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 +560,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,6 +668,8 @@ export async function prepareWorkspaceFileSearchDispatch(
         }
 
         const now = new Date()
+        // 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',
@@ -717,7 +737,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)