fix(file-search): rotate a woken owner whose slots are full by dispatch age

An owner already holding both of its outstanding claims can claim nothing, but a write between
passes kept marking it woken, so with little global capacity left it took every owner turn and
the free slot went unused. Woken owners now go first only while they have a free slot; a full
owner keeps the dispatch-age rotation it had before.

Also gives the due-reconcile work-check case the same one-minute margin as its not-yet-due
counterpart: exactly at the interval it compared a database timestamp with the application
clock and flaked on skew.

Author: waleedlatif1

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

@@ -31,6 +31,7 @@ import {
   FILE_SEARCH_INDEX_DISPATCH_WORKSPACES,
   FILE_SEARCH_INDEX_MAX_OUTSTANDING,
   FILE_SEARCH_INDEX_STALE_DISPATCH_MS,
+  FILE_SEARCH_INDEX_WORKSPACE_OUTSTANDING,
   FILE_SEARCH_RECONCILE_INTERVAL_MS,
 } from '@/lib/workspace-files/search/constants'
 import {
@@ -454,6 +455,20 @@ describe('workspace file search dispatch PostgreSQL deadlines', () => {
     expect(result.payloads.map((payload) => payload.fileId)).toEqual(['woken-workspace-000001'])
   })
 
+  it('rotates a woken owner whose slots are full behind an older waiting owner', async () => {
+    await leaveSlots(1 + FILE_SEARCH_INDEX_WORKSPACE_OUTSTANDING)
+    // Both of its slots are taken and it has more files; a write woke it after its last dispatch.
+    await seedQueue('busy-workspace', 3, FILE_SEARCH_INDEX_WORKSPACE_OUTSTANDING)
+    await connection`UPDATE file_search_dispatch_queue
+      SET last_dispatched_at = now() - interval '10 minutes' WHERE entity_id = 'busy-workspace'`
+    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'`
+
+    const result = await prepareWorkspaceFileSearchDispatch()
+    expect(result.payloads.map((payload) => payload.fileId)).toEqual(['waiting-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`
@@ -1027,8 +1042,10 @@ describe('workspace file search dispatch PostgreSQL deadlines', () => {
       },
       {
         work: 'a backfill reconcile that is due',
+        // A minute past the interval, mirroring the not-yet-due case: the check compares a database
+        // timestamp with the application clock, so exactly at the boundary it depends on clock skew.
         seed: () => connection`UPDATE workspace_file_search_backfill
-          SET completed_at = now() - make_interval(secs => ${reconcileSeconds})`,
+          SET completed_at = now() - make_interval(secs => ${reconcileSeconds + 60})`,
       },
       {
         work: 'a queued workspace',

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

@@ -186,6 +186,20 @@ const transactionNow = sql`now()`
  */
 const wokenSinceDispatch = sql`(${fileSearchDispatchQueue.lastDispatchedAt} IS NULL OR ${fileSearchDispatchQueue.updatedAt} > ${fileSearchDispatchQueue.lastDispatchedAt})`
 
+/**
+ * Whether the queued owner already holds every claim it may have outstanding. Such an owner can
+ * claim nothing this pass, so being woken must not put it ahead of owners that can: it keeps
+ * rotating by dispatch age, or a busy owner written to between passes would take every turn while
+ * the free global capacity goes unused.
+ */
+const ownerSlotsFull = sql`(SELECT count(*) FROM (
+  SELECT 1 FROM workspace_file_search_revision active
+  WHERE coalesce(active.entity_type, 'workspace') = ${fileSearchDispatchQueue.entityType}
+    AND coalesce(active.entity_id, active.workspace_id) = ${fileSearchDispatchQueue.entityId}
+    AND active.status = 'pending' AND active.dispatched_at IS NOT NULL
+  LIMIT ${FILE_SEARCH_INDEX_WORKSPACE_OUTSTANDING}
+) active_claims) >= ${FILE_SEARCH_INDEX_WORKSPACE_OUTSTANDING}`
+
 /**
  * Queues owners for the claim pass. A `reconciliation` wake-up, from the backfill walk, enters as
  * if just dispatched and never touches an owner already queued, so a page of owners with nothing
@@ -687,7 +701,7 @@ export async function prepareWorkspaceFileSearchDispatch(
           })
           .from(fileSearchDispatchQueue)
           .orderBy(
-            sql`${wokenSinceDispatch} DESC`,
+            sql`(${wokenSinceDispatch} AND NOT ${ownerSlotsFull}) DESC`,
             sql`${fileSearchDispatchQueue.lastDispatchedAt} ASC NULLS FIRST`,
             asc(fileSearchDispatchQueue.enqueuedAt),
             asc(fileSearchDispatchQueue.entityType),