fix(file-search): claim a file write that lands on an owner the backfill already queued

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. Each claim pass takes owners never dispatched first, so the owners a backfill page wakes enter the queue as if just dispatched: a page of reconciliation wake-ups rotates in by dispatch age behind the owners the file trigger woke for a write, instead of delaying them 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. 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.
 
 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/artifact-ready.ts

@@ -162,12 +162,13 @@ export async function markFileSearchArtifactReadyInTx(
       },
       setWhere: predicate,
     })
+  // The dispatcher stamps queue rows with the transaction clock and claims rows touched since.
   await tx
     .insert(fileSearchDispatchQueue)
-    .values({ ...owner, enqueuedAt: now, updatedAt: now })
+    .values({ ...owner, enqueuedAt: now, updatedAt: sql`now()` })
     .onConflictDoUpdate({
       target: [fileSearchDispatchQueue.entityType, fileSearchDispatchQueue.entityId],
-      set: { updatedAt: now },
+      set: { updatedAt: sql`now()` },
     })
   return true
 }

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 {
@@ -408,33 +409,80 @@ describe('workspace file search dispatch PostgreSQL deadlines', () => {
     })
   }
 
-  it('serves an owner woken by a file write ahead of a backfill page of reconciliation wake-ups', async () => {
-    const wakeUps = FILE_SEARCH_INDEX_DISPATCH_WORKSPACES * 2 + 50
-    await seedIndexedOwners('reconciled-', wakeUps)
-    const first = await prepareWorkspaceFileSearchDispatch()
-    expect(first.backfilledFiles).toBe(wakeUps)
-    expect(first.payloads).toEqual([])
+  /** Fills global capacity with claims of an unqueued owner, 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 writeFile('written-workspace')
-    const next = await prepareWorkspaceFileSearchDispatch()
-    expect(next.payloads.map((payload) => payload.fileId)).toEqual(['written-workspace-written'])
+  const wakeUps = FILE_SEARCH_INDEX_DISPATCH_WORKSPACES * 2 + 50
+  it.each([
+    ['an owner the page did not reach', 'written-workspace', 0],
+    ['an owner the page left queued', `reconciled-${String(wakeUps).padStart(4, '0')}`, 1],
+  ])(
+    'serves a file write to %s ahead of a backfill page of reconciliation wake-ups',
+    async (_, owner, queuedRows) => {
+      await seedIndexedOwners('reconciled-', wakeUps)
+      const first = await prepareWorkspaceFileSearchDispatch()
+      expect(first.backfilledFiles).toBe(wakeUps)
+      expect(first.payloads).toEqual([])
+      // A queued owner's write takes the conflict path; an unqueued owner's takes the insert path.
+      const [queued] = await connection`SELECT count(*)::int AS count
+      FROM file_search_dispatch_queue WHERE entity_id = ${owner}`
+      expect(queued.count).toBe(queuedRows)
+
+      await writeFile(owner)
+      const next = await prepareWorkspaceFileSearchDispatch()
+      expect(next.payloads.map((payload) => payload.fileId)).toEqual([`${owner}-written`])
+    }
+  )
+
+  it('keeps a queued reconciliation wake-up behind woken work when the backfill passes it again', async () => {
+    await leaveSlots(1)
+    // Queued by an earlier page, with nothing to claim; its stamp is older than the woken owner's.
+    await seedIndexedOwners('reconciled-', 1)
+    await connection`INSERT INTO file_search_dispatch_queue
+      (entity_type, entity_id, enqueued_at, updated_at, last_dispatched_at)
+      VALUES ('workspace', 'reconciled-0001', now() - interval '2 hours',
+        now() - interval '2 hours', now() - interval '2 hours')`
+    await seedQueue('woken-workspace', 1)
+    await connection`UPDATE file_search_dispatch_queue
+      SET last_dispatched_at = now() - interval '1 hour' WHERE entity_id = 'woken-workspace'`
+    await connection`UPDATE workspace_file_search_backfill SET completed_at = NULL`
+
+    const result = await prepareWorkspaceFileSearchDispatch()
+    expect(result.backfilledFiles).toBeGreaterThan(0)
+    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 seedQueue('workspace-active', 0, FILE_SEARCH_INDEX_MAX_OUTSTANDING)
-    await connection`DELETE FROM file_search_dispatch_queue WHERE entity_id = 'workspace-active'`
+    await leaveSlots(0)
     await connection`UPDATE workspace_file_search_backfill SET completed_at = NULL`
     await connection`INSERT INTO workspace_files (id, workspace_id, context, content_updated_at)
       VALUES ('backfill-found-file', 'backfill-found', 'workspace', '2026-09-16')`
-    /** No capacity: the page seeds and queues the unindexed file's owner but claims nothing. */
+    // No capacity: the page seeds and queues the unindexed file's owner but claims nothing.
     const seeded = await prepareWorkspaceFileSearchDispatch()
     expect(seeded.backfilledFiles).toBeGreaterThan(0)
     expect(seeded.payloads).toEqual([])
 
     await writeFile('fresh-write')
     const served: string[] = []
     for (let pass = 0; pass < 2; pass++) {
-      /** Free exactly one slot, so each pass claims for one owner. */
+      // Free exactly one slot, so each pass claims for one owner.
       await connection`UPDATE workspace_file_search_revision SET status = 'ready'
         WHERE file_id = (SELECT file_id FROM workspace_file_search_revision
           WHERE status = 'pending' AND dispatched_at IS NOT NULL ORDER BY file_id LIMIT 1)`
@@ -445,9 +493,8 @@ 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 seedQueue('workspace-active', 0, FILE_SEARCH_INDEX_MAX_OUTSTANDING - 1)
-    await connection`DELETE FROM file_search_dispatch_queue WHERE entity_id = 'workspace-active'`
-    /** Queued first and served an hour ago: it wins the single slot only if the write is demoted. */
+    await leaveSlots(1)
+    // Queued first 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'`
@@ -995,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

@@ -171,11 +171,39 @@ function queuedOwnerFilter(owners: readonly EditableFileOwner[]): SQL | undefine
   return or(...owners.map(queuedOwner))
 }
 
+/**
+ * The transaction's start time. Queue rows compare `updated_at` with `last_dispatched_at`, and
+ * the `workspace_file_search_mark_pending` trigger writes `updated_at` the same way, so both sides
+ * share one clock.
+ */
+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.
+ */
+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: claims take owners never dispatched first, so marking it served keeps a page
- * of owners with nothing new behind the owners the file trigger woke for a write, and its row still
- * rotates in by dispatch age. An owner already queued keeps its place either way.
+ * if just dispatched and never touches an owner already queued, so a page of owners with nothing
+ * new rotates in behind the owners woken for work.
  */
 async function enqueueOwners(
   tx: DbTransaction,
@@ -185,20 +213,18 @@ async function enqueueOwners(
 ): Promise<void> {
   const unique = [...new Map(owners.map((owner) => [JSON.stringify(owner), owner])).values()]
   if (unique.length === 0) return
-  await tx
-    .insert(fileSearchDispatchQueue)
-    .values(
-      unique.map((owner) => ({
-        ...owner,
-        enqueuedAt: now,
-        updatedAt: now,
-        lastDispatchedAt: reconciliation ? now : null,
-      }))
-    )
-    .onConflictDoUpdate({
-      target: [fileSearchDispatchQueue.entityType, fileSearchDispatchQueue.entityId],
-      set: { updatedAt: now },
-    })
+  const target = [fileSearchDispatchQueue.entityType, fileSearchDispatchQueue.entityId]
+  const insert = tx.insert(fileSearchDispatchQueue).values(
+    unique.map((owner) => ({
+      ...owner,
+      enqueuedAt: now,
+      updatedAt: transactionNow,
+      lastDispatchedAt: reconciliation ? transactionNow : null,
+    }))
+  )
+  await (reconciliation
+    ? insert.onConflictDoNothing({ target })
+    : insert.onConflictDoUpdate({ target, set: { updatedAt: transactionNow } }))
 }
 
 /**
@@ -675,6 +701,7 @@ export async function prepareWorkspaceFileSearchDispatch(
           })
           .from(fileSearchDispatchQueue)
           .orderBy(
+            sql`(${wokenSinceDispatch} AND NOT ${ownerSlotsFull}) DESC`,
             sql`${fileSearchDispatchQueue.lastDispatchedAt} ASC NULLS FIRST`,
             asc(fileSearchDispatchQueue.enqueuedAt),
             asc(fileSearchDispatchQueue.entityType),
@@ -687,7 +714,7 @@ export async function prepareWorkspaceFileSearchDispatch(
           // Rotate busy owners too; otherwise a full locked page starves every later owner.
           await tx
             .update(fileSearchDispatchQueue)
-            .set({ lastDispatchedAt: now, updatedAt: now })
+            .set({ lastDispatchedAt: transactionNow, updatedAt: transactionNow })
             .where(queuedOwnerFilter(candidates))
         }
         const owners = await lockFileSearchOwners(tx, candidates, { skipBusy: true })