Skip to content

Commit f279d9e

Browse files
authored
fix(file-search): fill dispatch capacity in selection order and keep reaped claims ahead of the rotation (#9671)
* fix(file-search): fill dispatch capacity in selection order and reap before the backfill The claim cut ranked every selected owner's candidates by revision age, so with less capacity than candidates a woken owner selected first could lose every slot to an older backlog selected after it, and the claim's stamp then filed it behind the rotation. Claims now fill capacity owner by owner in the pass's selection order (each still up to its free slots). The backfill page ran before the reaper in the same transaction, so an owner it queued as a reconciliation wake-up absorbed the reaper's wake-up (both writes at the transaction time) and the released claim waited behind the rotation. The reaper now runs first; the backfill's wake-ups leave queued owners untouched. Tests: a two-slot pass with a woken owner against an older backlog, and a reaped claim whose owner the backfill queues in the same pass; the capacity filler's files are deleted so the backfill never re-queues it. * fix(file-search): keep the dispatch pass's file-then-queue lock order Reaping before the backfill held the owner's queue row while the backfill took FOR SHARE on its unindexed files; a concurrent save holds its file while its trigger waits on that queue row, so the two could deadlock. The pass runs backfill then reap again, and a wake that conflicts with a row this transaction's backfill just stamped as a reconciliation clears the stamp, so a reaped claim stays ahead of the rotation.
1 parent 5e57170 commit f279d9e

3 files changed

Lines changed: 76 additions & 14 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. 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.
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 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.
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: 40 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -409,10 +409,14 @@ describe('workspace file search dispatch PostgreSQL deadlines', () => {
409409
})
410410
}
411411

412-
/** Fills global capacity with claims of an unqueued owner, leaving `slots` free. */
412+
/**
413+
* Fills global capacity with claims of an owner that is neither queued nor walked by the
414+
* backfill (its files are deleted), leaving `slots` free.
415+
*/
413416
async function leaveSlots(slots: number) {
414417
await seedQueue('workspace-active', 0, FILE_SEARCH_INDEX_MAX_OUTSTANDING - slots)
415418
await connection`DELETE FROM file_search_dispatch_queue WHERE entity_id = 'workspace-active'`
419+
await connection`UPDATE workspace_files SET deleted_at = now() WHERE workspace_id = 'workspace-active'`
416420
}
417421

418422
const wakeUps = FILE_SEARCH_INDEX_DISPATCH_WORKSPACES * 2 + 50
@@ -469,6 +473,40 @@ describe('workspace file search dispatch PostgreSQL deadlines', () => {
469473
expect(result.payloads.map((payload) => payload.fileId)).toEqual(['waiting-workspace-000001'])
470474
})
471475

476+
it('fills the free capacity in selection order, so a woken owner is not outbid by an older backlog', async () => {
477+
await leaveSlots(2)
478+
// Served ten minutes ago and not woken since, with older pending files than the fresh write.
479+
await seedQueue('backlog-workspace', 3)
480+
await connection`UPDATE file_search_dispatch_queue SET last_dispatched_at = now() - interval '10 minutes',
481+
updated_at = now() - interval '10 minutes' WHERE entity_id = 'backlog-workspace'`
482+
await writeFile('fresh-workspace')
483+
484+
const result = await prepareWorkspaceFileSearchDispatch()
485+
expect(result.payloads.map((payload) => payload.fileId).sort()).toEqual([
486+
'backlog-workspace-000001',
487+
'fresh-workspace-written',
488+
])
489+
})
490+
491+
it('keeps a reaped claim woken when the backfill queues its owner in the same pass', async () => {
492+
await leaveSlots(1)
493+
// Its only claim was abandoned before its run was enqueued; nothing has queued its owner yet.
494+
await seedQueue('reaped-workspace', 0, 1)
495+
await connection`DELETE FROM file_search_dispatch_queue WHERE entity_id = 'reaped-workspace'`
496+
await connection`UPDATE workspace_file_search_revision
497+
SET handoff_expires_at = clock_timestamp() - interval '1 millisecond'
498+
WHERE file_id = 'reaped-workspace-000001'`
499+
// Served an hour ago and not woken since: it wins the single slot only if the reap is lost.
500+
await seedQueue('waiting-workspace', 1)
501+
await connection`UPDATE file_search_dispatch_queue SET last_dispatched_at = now() - interval '1 hour',
502+
updated_at = now() - interval '1 hour' WHERE entity_id = 'waiting-workspace'`
503+
await connection`UPDATE workspace_file_search_backfill SET completed_at = NULL`
504+
505+
const result = await prepareWorkspaceFileSearchDispatch()
506+
expect(result.reapedClaims).toBe(1)
507+
expect(result.payloads.map((payload) => payload.fileId)).toEqual(['reaped-workspace-000001'])
508+
})
509+
472510
it('serves a fresh write before work the backfill found, then the backfill work', async () => {
473511
await leaveSlots(0)
474512
await connection`UPDATE workspace_file_search_backfill SET completed_at = NULL`
@@ -494,7 +532,7 @@ describe('workspace file search dispatch PostgreSQL deadlines', () => {
494532

495533
it('leaves an owner queued by a file write at the front when a backfill page passes it', async () => {
496534
await leaveSlots(1)
497-
// Queued first and served an hour ago: it wins the single slot only if the write is demoted.
535+
// Woken and served an hour ago: it wins the single slot only if the write is demoted.
498536
await seedQueue('earlier-served', 1)
499537
await connection`UPDATE file_search_dispatch_queue
500538
SET last_dispatched_at = now() - interval '1 hour' WHERE entity_id = 'earlier-served'`

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

Lines changed: 35 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -163,6 +163,10 @@ function rowOwner(row: { entityType: string; entityId: string }): EditableFileOw
163163
return resolveFileSearchOwner({ owner: row })
164164
}
165165

166+
function ownerKey(owner: EditableFileOwner): string {
167+
return JSON.stringify([owner.entityType, owner.entityId])
168+
}
169+
166170
function queuedOwner(owner: EditableFileOwner): SQL {
167171
return sql`(${fileSearchDispatchQueue.entityType} = ${owner.entityType} AND ${fileSearchDispatchQueue.entityId} = ${owner.entityId})`
168172
}
@@ -180,9 +184,10 @@ const transactionNow = sql`now()`
180184

181185
/**
182186
* Whether an owner was woken after its last dispatch: never dispatched, or touched since. A file
183-
* write, a released claim, and a reaped claim advance `updated_at`; a claim stamps both columns
184-
* with one time, and a reconciliation wake-up enters with them equal. A write whose transaction
185-
* began before the claim that stamped its owner reads as earlier and waits one rotation.
187+
* write, a released or reaped claim, and an artifact-ready requeue advance `updated_at`; a claim
188+
* stamps both columns with one time, and a reconciliation wake-up enters with them equal. A write
189+
* whose transaction began before the claim that stamped its owner reads as earlier, so the owner
190+
* keeps the dispatch-age rotation instead.
186191
*/
187192
const wokenSinceDispatch = sql`(${fileSearchDispatchQueue.lastDispatchedAt} IS NULL OR ${fileSearchDispatchQueue.updatedAt} > ${fileSearchDispatchQueue.lastDispatchedAt})`
188193

@@ -211,7 +216,7 @@ async function enqueueOwners(
211216
now: Date,
212217
{ reconciliation = false }: { reconciliation?: boolean } = {}
213218
): Promise<void> {
214-
const unique = [...new Map(owners.map((owner) => [JSON.stringify(owner), owner])).values()]
219+
const unique = [...new Map(owners.map((owner) => [ownerKey(owner), owner])).values()]
215220
if (unique.length === 0) return
216221
const target = [fileSearchDispatchQueue.entityType, fileSearchDispatchQueue.entityId]
217222
const insert = tx.insert(fileSearchDispatchQueue).values(
@@ -224,7 +229,15 @@ async function enqueueOwners(
224229
)
225230
await (reconciliation
226231
? insert.onConflictDoNothing({ target })
227-
: insert.onConflictDoUpdate({ target, set: { updatedAt: transactionNow } }))
232+
: insert.onConflictDoUpdate({
233+
target,
234+
set: {
235+
updatedAt: transactionNow,
236+
// A row this pass's backfill just queued as a reconciliation is woken as if the wake had
237+
// queued it first; a stamp from an earlier transaction is always older than `now()`.
238+
lastDispatchedAt: sql`CASE WHEN ${fileSearchDispatchQueue.lastDispatchedAt} = ${transactionNow} THEN NULL ELSE ${fileSearchDispatchQueue.lastDispatchedAt} END`,
239+
},
240+
}))
228241
}
229242

230243
/**
@@ -355,7 +368,7 @@ async function seedBackfillPage(tx: DbTransaction, now: Date): Promise<number> {
355368
)
356369
.onConflictDoNothing()
357370
}
358-
/** Every owner the page passed is woken, indexed or not, so the pass reconciles its builds. */
371+
// Every owner the page passed is queued, indexed or not, so a pass reconciles its builds.
359372
await enqueueOwners(tx, rows.map(rowOwner), now, { reconciliation: true })
360373

361374
const last = rows.at(-1)
@@ -495,6 +508,11 @@ async function reconcileOwnerDependencies(
495508
}
496509
}
497510

511+
/**
512+
* Claims up to each owner's free slots, filling `remainingGlobalCapacity` owner by owner in the
513+
* order of `owners` (the pass's selection order), so a woken owner is never outbid by an older
514+
* backlog selected after it.
515+
*/
498516
async function claimQueuedOwnerJobs(
499517
tx: DbTransaction,
500518
owners: readonly EditableFileOwner[],
@@ -503,7 +521,7 @@ async function claimQueuedOwnerJobs(
503521
): Promise<WorkspaceFileSearchIndexPayload[]> {
504522
if (owners.length === 0 || remainingGlobalCapacity <= 0) return []
505523
const ownerValues = sql.join(
506-
owners.map((owner) => sql`(${owner.entityType}, ${owner.entityId})`),
524+
owners.map((owner, rank) => sql`(${owner.entityType}, ${owner.entityId}, ${rank}::int)`),
507525
sql`, `
508526
)
509527
const rows = await tx.execute<{
@@ -512,8 +530,8 @@ async function claimQueuedOwnerJobs(
512530
fileId: string
513531
sourceContentUpdatedAt: string
514532
}>(sql`
515-
WITH selected_owner(entity_type, entity_id) AS (VALUES ${ownerValues}), candidates AS MATERIALIZED (
516-
SELECT queued.* FROM selected_owner selected
533+
WITH selected_owner(entity_type, entity_id, claim_rank) AS (VALUES ${ownerValues}), candidates AS MATERIALIZED (
534+
SELECT queued.*, selected.claim_rank FROM selected_owner selected
517535
CROSS JOIN LATERAL (
518536
SELECT count(*)::int AS active_count FROM (
519537
SELECT 1 FROM workspace_file_search_revision active
@@ -542,7 +560,7 @@ async function claimQueuedOwnerJobs(
542560
LIMIT greatest(0, ${FILE_SEARCH_INDEX_WORKSPACE_OUTSTANDING} - owner_active.active_count)
543561
FOR UPDATE OF search_index SKIP LOCKED
544562
) queued
545-
ORDER BY queued.updated_at, queued.entity_type, queued.entity_id, queued.file_id, queued.source_content_updated_at
563+
ORDER BY selected.claim_rank, queued.updated_at, queued.file_id, queued.source_content_updated_at
546564
LIMIT ${remainingGlobalCapacity}
547565
) UPDATE workspace_file_search_revision search_index
548566
SET dispatched_at = ${now.toISOString()}::timestamp,
@@ -650,6 +668,8 @@ export async function prepareWorkspaceFileSearchDispatch(
650668
}
651669

652670
const now = new Date()
671+
// The backfill locks files before the reap locks queue rows: a file save holds its file
672+
// while its trigger waits on the owner's queue row, so the other order can deadlock.
653673
const backfilledFiles = await runDispatchPhase('backfill', () => seedBackfillPage(tx, now))
654674
const { reaped: reapedClaims, abandoned: abandonedClaims } = await runDispatchPhase(
655675
'reap',
@@ -717,7 +737,11 @@ export async function prepareWorkspaceFileSearchDispatch(
717737
.set({ lastDispatchedAt: transactionNow, updatedAt: transactionNow })
718738
.where(queuedOwnerFilter(candidates))
719739
}
720-
const owners = await lockFileSearchOwners(tx, candidates, { skipBusy: true })
740+
const locked = new Set(
741+
(await lockFileSearchOwners(tx, candidates, { skipBusy: true })).map(ownerKey)
742+
)
743+
// Locking sorts owners by key; claims keep the selection order.
744+
const owners = candidates.filter((owner) => locked.has(ownerKey(owner)))
721745
await reconcileOwnerDependencies(tx, owners, now)
722746
const payloads = await runDispatchPhase('claim', () =>
723747
claimQueuedOwnerJobs(tx, owners, remainingGlobalCapacity, now)

0 commit comments

Comments
 (0)