Skip to content

Commit 81eb564

Browse files
authored
fix(file-search): claim owners woken by a file write ahead of backfill reconciliation (#9655)
The backfill walk wakes every owner a page passes so the dispatcher reconciles its builds. Those wake-ups entered the owner queue never dispatched, the same as an owner the file trigger woke for a write, and claims take never-dispatched owners first in enqueue order. A page across many small owners therefore put dozens of wake-ups with nothing to claim ahead of a fresh upload, which then waited several passes to be indexed. Owners with pending work that had been dispatched before waited behind all of them. Backfill wake-ups now enter as if just dispatched, so write-woken owners are claimed first and the wake-ups still rotate in by dispatch age. An owner already queued keeps its place.
1 parent 3261cf5 commit 81eb564

3 files changed

Lines changed: 97 additions & 4 deletions

File tree

‎apps/sim/lib/workspace-files/search/README.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -71,6 +71,6 @@ The design uses PostgreSQL's documented [TOAST behavior](https://www.postgresql.
7171

7272
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.
7373

74-
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.
74+
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.
7575

7676
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‎

Lines changed: 79 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ import {
2929
FILE_SEARCH_BACKFILL_PAGE_SIZE,
3030
FILE_SEARCH_DISPATCH_HANDOFF_MS,
3131
FILE_SEARCH_INDEX_DISPATCH_WORKSPACES,
32+
FILE_SEARCH_INDEX_MAX_OUTSTANDING,
3233
FILE_SEARCH_INDEX_STALE_DISPATCH_MS,
3334
FILE_SEARCH_RECONCILE_INTERVAL_MS,
3435
} from '@/lib/workspace-files/search/constants'
@@ -380,6 +381,84 @@ describe('workspace file search dispatch PostgreSQL deadlines', () => {
380381
expect(row.active).toBe(100)
381382
})
382383

384+
/** Live files with ready revisions: owners a backfill page wakes with nothing to claim. */
385+
async function seedIndexedOwners(prefix: string, count: number) {
386+
await connection`INSERT INTO workspace_files (id, workspace_id, context, content_updated_at)
387+
SELECT ${prefix} || lpad(n::text, 4, '0') || '-file', ${prefix} || lpad(n::text, 4, '0'),
388+
'workspace', '2026-09-16'
389+
FROM generate_series(1, ${count}) n`
390+
await connection`INSERT INTO workspace_file_search_revision
391+
(file_id, workspace_id, entity_type, entity_id, source_content_updated_at, status, updated_at)
392+
SELECT id, workspace_id, 'workspace', workspace_id, content_updated_at, 'ready', '2026-09-16'
393+
FROM workspace_files WHERE id LIKE ${`${prefix}%`}`
394+
}
395+
396+
/**
397+
* Writes a file through the real `workspace_file_search_mark_pending` trigger, attached only for
398+
* this insert so the other fixtures keep seeding revisions and queue rows directly.
399+
*/
400+
async function writeFile(workspaceId: string) {
401+
await connection.begin(async (tx) => {
402+
await tx`CREATE TRIGGER write_marks_pending
403+
AFTER INSERT OR UPDATE OF content_updated_at, deleted_at, context, workspace_id
404+
ON workspace_files FOR EACH ROW EXECUTE FUNCTION workspace_file_search_mark_pending()`
405+
await tx`INSERT INTO workspace_files (id, workspace_id, context, content_updated_at)
406+
VALUES (${`${workspaceId}-written`}, ${workspaceId}, 'workspace', '2026-09-16')`
407+
await tx`DROP TRIGGER write_marks_pending ON workspace_files`
408+
})
409+
}
410+
411+
it('serves an owner woken by a file write ahead of a backfill page of reconciliation wake-ups', async () => {
412+
const wakeUps = FILE_SEARCH_INDEX_DISPATCH_WORKSPACES * 2 + 50
413+
await seedIndexedOwners('reconciled-', wakeUps)
414+
const first = await prepareWorkspaceFileSearchDispatch()
415+
expect(first.backfilledFiles).toBe(wakeUps)
416+
expect(first.payloads).toEqual([])
417+
418+
await writeFile('written-workspace')
419+
const next = await prepareWorkspaceFileSearchDispatch()
420+
expect(next.payloads.map((payload) => payload.fileId)).toEqual(['written-workspace-written'])
421+
})
422+
423+
it('serves a fresh write before work the backfill found, then the backfill work', async () => {
424+
await seedQueue('workspace-active', 0, FILE_SEARCH_INDEX_MAX_OUTSTANDING)
425+
await connection`DELETE FROM file_search_dispatch_queue WHERE entity_id = 'workspace-active'`
426+
await connection`UPDATE workspace_file_search_backfill SET completed_at = NULL`
427+
await connection`INSERT INTO workspace_files (id, workspace_id, context, content_updated_at)
428+
VALUES ('backfill-found-file', 'backfill-found', 'workspace', '2026-09-16')`
429+
/** No capacity: the page seeds and queues the unindexed file's owner but claims nothing. */
430+
const seeded = await prepareWorkspaceFileSearchDispatch()
431+
expect(seeded.backfilledFiles).toBeGreaterThan(0)
432+
expect(seeded.payloads).toEqual([])
433+
434+
await writeFile('fresh-write')
435+
const served: string[] = []
436+
for (let pass = 0; pass < 2; pass++) {
437+
/** Free exactly one slot, so each pass claims for one owner. */
438+
await connection`UPDATE workspace_file_search_revision SET status = 'ready'
439+
WHERE file_id = (SELECT file_id FROM workspace_file_search_revision
440+
WHERE status = 'pending' AND dispatched_at IS NOT NULL ORDER BY file_id LIMIT 1)`
441+
const result = await prepareWorkspaceFileSearchDispatch()
442+
served.push(...result.payloads.map((payload) => payload.fileId))
443+
}
444+
expect(served).toEqual(['fresh-write-written', 'backfill-found-file'])
445+
})
446+
447+
it('leaves an owner queued by a file write at the front when a backfill page passes it', async () => {
448+
await seedQueue('workspace-active', 0, FILE_SEARCH_INDEX_MAX_OUTSTANDING - 1)
449+
await connection`DELETE FROM file_search_dispatch_queue WHERE entity_id = 'workspace-active'`
450+
/** Queued first and served an hour ago: it wins the single slot only if the write is demoted. */
451+
await seedQueue('earlier-served', 1)
452+
await connection`UPDATE file_search_dispatch_queue
453+
SET last_dispatched_at = now() - interval '1 hour' WHERE entity_id = 'earlier-served'`
454+
await writeFile('written-workspace')
455+
await connection`UPDATE workspace_file_search_backfill SET completed_at = NULL`
456+
457+
const result = await prepareWorkspaceFileSearchDispatch()
458+
expect(result.backfilledFiles).toBeGreaterThan(0)
459+
expect(result.payloads.map((payload) => payload.fileId)).toEqual(['written-workspace-written'])
460+
})
461+
383462
it('walks every live workspace file exactly once across backfill pages', async () => {
384463
/** Deliberately not a whole number of pages, so the short final page ends the walk. */
385464
const files =

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

Lines changed: 17 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -171,16 +171,30 @@ function queuedOwnerFilter(owners: readonly EditableFileOwner[]): SQL | undefine
171171
return or(...owners.map(queuedOwner))
172172
}
173173

174+
/**
175+
* Queues owners for the claim pass. A `reconciliation` wake-up, from the backfill walk, enters as
176+
* if just dispatched: claims take owners never dispatched first, so marking it served keeps a page
177+
* of owners with nothing new behind the owners the file trigger woke for a write, and its row still
178+
* rotates in by dispatch age. An owner already queued keeps its place either way.
179+
*/
174180
async function enqueueOwners(
175181
tx: DbTransaction,
176182
owners: readonly EditableFileOwner[],
177-
now: Date
183+
now: Date,
184+
{ reconciliation = false }: { reconciliation?: boolean } = {}
178185
): Promise<void> {
179186
const unique = [...new Map(owners.map((owner) => [JSON.stringify(owner), owner])).values()]
180187
if (unique.length === 0) return
181188
await tx
182189
.insert(fileSearchDispatchQueue)
183-
.values(unique.map((owner) => ({ ...owner, enqueuedAt: now, updatedAt: now })))
190+
.values(
191+
unique.map((owner) => ({
192+
...owner,
193+
enqueuedAt: now,
194+
updatedAt: now,
195+
lastDispatchedAt: reconciliation ? now : null,
196+
}))
197+
)
184198
.onConflictDoUpdate({
185199
target: [fileSearchDispatchQueue.entityType, fileSearchDispatchQueue.entityId],
186200
set: { updatedAt: now },
@@ -316,7 +330,7 @@ async function seedBackfillPage(tx: DbTransaction, now: Date): Promise<number> {
316330
.onConflictDoNothing()
317331
}
318332
/** Every owner the page passed is woken, indexed or not, so the pass reconciles its builds. */
319-
await enqueueOwners(tx, rows.map(rowOwner), now)
333+
await enqueueOwners(tx, rows.map(rowOwner), now, { reconciliation: true })
320334

321335
const last = rows.at(-1)
322336
await tx

0 commit comments

Comments
 (0)