From 3a57638c906a95776a6fa8602cd208d177efea33 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 4 Nov 2026 01:44:34 -0800 Subject: [PATCH 1/2] fix(file-search): claim a file write that lands on an owner the backfill already queued Backfill reconciliation wake-ups enter the owner queue as if just dispatched so they rotate in behind owners woken for a write. A write that landed while such a row was still queued only advanced its updated_at through the file trigger's upsert, so it kept the reconciliation stamp and waited behind every older-stamped owner. The same held for an owner that was served, left queued with work, and then woken again. Claims now take owners woken since their last dispatch first: never dispatched, or updated_at later than last_dispatched_at, which every real wake-up (file write, released claim, reaped claim, artifact-ready requeue) advances and a reconciliation wake-up never does; it no longer touches an owner already queued. Within each group the dispatch-age rotation is unchanged. Queue timestamps the comparison reads come from the transaction clock, the same clock the trigger writes. --- apps/sim/lib/workspace-files/search/README.md | 2 +- .../workspace-files/search/artifact-ready.ts | 5 +- .../search/dispatcher.integration.ts | 64 ++++++++++++++----- .../lib/workspace-files/search/dispatcher.ts | 49 ++++++++------ 4 files changed, 83 insertions(+), 37 deletions(-) diff --git a/apps/sim/lib/workspace-files/search/README.md b/apps/sim/lib/workspace-files/search/README.md index 578acef6918..25a96fa157a 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. 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. diff --git a/apps/sim/lib/workspace-files/search/artifact-ready.ts b/apps/sim/lib/workspace-files/search/artifact-ready.ts index c41698dc70f..65364ef4c46 100644 --- a/apps/sim/lib/workspace-files/search/artifact-ready.ts +++ b/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 } diff --git a/apps/sim/lib/workspace-files/search/dispatcher.integration.ts b/apps/sim/lib/workspace-files/search/dispatcher.integration.ts index d5f75f9fb87..2f24138dc06 100644 --- a/apps/sim/lib/workspace-files/search/dispatcher.integration.ts +++ b/apps/sim/lib/workspace-files/search/dispatcher.integration.ts @@ -408,25 +408,58 @@ 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('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([]) @@ -434,7 +467,7 @@ describe('workspace file search dispatch PostgreSQL deadlines', () => { 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 +478,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'` diff --git a/apps/sim/lib/workspace-files/search/dispatcher.ts b/apps/sim/lib/workspace-files/search/dispatcher.ts index 8ac0307fc13..031774e3fa4 100644 --- a/apps/sim/lib/workspace-files/search/dispatcher.ts +++ b/apps/sim/lib/workspace-files/search/dispatcher.ts @@ -171,11 +171,25 @@ 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})` + /** * 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 +199,18 @@ async function enqueueOwners( ): Promise { 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 +687,7 @@ export async function prepareWorkspaceFileSearchDispatch( }) .from(fileSearchDispatchQueue) .orderBy( + sql`${wokenSinceDispatch} DESC`, sql`${fileSearchDispatchQueue.lastDispatchedAt} ASC NULLS FIRST`, asc(fileSearchDispatchQueue.enqueuedAt), asc(fileSearchDispatchQueue.entityType), @@ -687,7 +700,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 }) From c548150b595dcf0fbeea79074679bed59afa8f2e Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 4 Nov 2026 02:02:50 -0800 Subject: [PATCH 2/2] 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. --- .../search/dispatcher.integration.ts | 19 ++++++++++++++++++- .../lib/workspace-files/search/dispatcher.ts | 16 +++++++++++++++- 2 files changed, 33 insertions(+), 2 deletions(-) diff --git a/apps/sim/lib/workspace-files/search/dispatcher.integration.ts b/apps/sim/lib/workspace-files/search/dispatcher.integration.ts index 2f24138dc06..b56b4369a8f 100644 --- a/apps/sim/lib/workspace-files/search/dispatcher.integration.ts +++ b/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', diff --git a/apps/sim/lib/workspace-files/search/dispatcher.ts b/apps/sim/lib/workspace-files/search/dispatcher.ts index 031774e3fa4..7683ec4a072 100644 --- a/apps/sim/lib/workspace-files/search/dispatcher.ts +++ b/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),