test(cron): assert scheduled-pass outcomes through a deduplicating Trigger fake; plan test on production indexes

Author: waleedlatif1

apps/sim/lib/core/outbox/enqueue.test.ts

@@ -1,6 +1,10 @@
 import { asyncJobsRegionMock } from '@sim/testing/mocks/async-jobs-region.mock'
 import { setEnvFlags } from '@sim/testing/mocks/env-flags.mock'
 import { outboxServiceMock, outboxServiceMockFns } from '@sim/testing/mocks/outbox-service.mock'
+import {
+  createIdempotentTasksTrigger,
+  triggerSdkMockFns,
+} from '@sim/testing/mocks/trigger-sdk.mock'
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 
 const hoisted = vi.hoisted(() => ({
@@ -10,12 +14,11 @@ vi.mock('@/lib/core/async-jobs/region', () => asyncJobsRegionMock)
 vi.mock('@/lib/core/outbox/processor', () => ({ runOutboxProcessor: hoisted.processor }))
 vi.mock('@/lib/core/outbox/service', () => outboxServiceMock)
 
-import { tasks } from '@trigger.dev/sdk'
 import { enqueueOutboxProcessor } from '@/lib/core/outbox/enqueue'
 
 const mocks = {
   ...hoisted,
-  trigger: vi.mocked(tasks.trigger),
+  trigger: triggerSdkMockFns.mockTasksTrigger,
   hasDueWork: outboxServiceMockFns.mockHasDueOutboxWork,
 }
 
@@ -48,36 +51,28 @@ describe('outbox processor enqueue', () => {
     vi.useFakeTimers()
     vi.setSystemTime(IDLE_MINUTE)
     setEnvFlags({ isTriggerDevEnabled: true })
-    mocks.trigger.mockResolvedValue({ id: 'run-1' })
+    mocks.trigger.mockImplementation(createIdempotentTasksTrigger())
     mocks.processor.mockReset()
     mocks.processor.mockResolvedValue(INLINE_OUTPUT)
     mocks.hasDueWork.mockReset()
     mocks.hasDueWork.mockResolvedValue(true)
   })
   afterEach(() => vi.useRealTimers())
 
-  it('deduplicates duplicate ticks while allowing the next minute to drain more work', async () => {
-    await enqueueOutboxProcessor()
-    await enqueueOutboxProcessor()
-    vi.advanceTimersByTime(60_000)
-    await enqueueOutboxProcessor()
-    const keys = mocks.trigger.mock.calls.map((call) => call[2].idempotencyKey)
-    expect(keys[0]).toBe(keys[1])
-    expect(keys[2]).not.toBe(keys[0])
-  })
-
-  it('keys the run by the window its tick was gated in when the work check outlasts it', async () => {
+  it('folds ticks into the run of the minute their work check was made in', async () => {
+    const first = await enqueueOutboxProcessor()
+    const duplicate = await enqueueOutboxProcessor()
     mocks.hasDueWork.mockImplementationOnce(async () => {
       vi.advanceTimersByTime(60_000)
       return true
     })
-    await enqueueOutboxProcessor()
-    expect(mocks.trigger).toHaveBeenCalledWith('process-outbox', undefined, {
-      idempotencyKey: 'process-outbox:29826034',
-      idempotencyKeyTTL: '5m',
-      maxDuration: 900,
-      region: 'us-east-1',
-    })
+    const checkedAcrossTheBoundary = await enqueueOutboxProcessor()
+    const nextMinute = await enqueueOutboxProcessor()
+
+    expect(first.jobId).toBe('run-1')
+    expect(duplicate.jobId).toBe('run-1')
+    expect(checkedAcrossTheBoundary.jobId).toBe('run-1')
+    expect(nextMinute.jobId).toBe('run-2')
   })
 
   it('fails closed on an enqueue error without starting concurrent inline work', async () => {

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

@@ -18,10 +18,6 @@ vi.mock('@/lib/workspace-files/search/indexing', () => ({
   indexWorkspaceFileForSearch: vi.fn(),
   markWorkspaceFileSearchIndexFailed: vi.fn(),
 }))
-vi.mock('@/lib/workspace-files/search/index-state', async (importOriginal) => ({
-  ...(await importOriginal<typeof import('@/lib/workspace-files/search/index-state')>()),
-  cleanupFileSearchBuilds: vi.fn().mockResolvedValue(0),
-}))
 vi.mock('@trigger.dev/sdk', () => ({ tasks: { batchTrigger: mocks.batchTrigger } }))
 vi.mock('@/lib/core/config/env-flags', () => ({ isTriggerDevEnabled: true }))
 vi.mock('@/lib/core/async-jobs/region', () => ({ resolveTriggerRegion: async () => 'us-east-1' }))
@@ -39,6 +35,19 @@ import {
   prepareWorkspaceFileSearchDispatch,
 } from '@/lib/workspace-files/search/dispatcher'
 
+interface QueryPlan {
+  'Node Type': string
+  'Relation Name'?: string
+  'Index Name'?: string
+  'Shared Hit Blocks': number
+  'Shared Read Blocks': number
+  Plans?: QueryPlan[]
+}
+
+function planNodes(plan: QueryPlan): QueryPlan[] {
+  return [plan, ...(plan.Plans ?? []).flatMap(planNodes)]
+}
+
 describe('workspace file search dispatch PostgreSQL deadlines', () => {
   const schemaName = `dispatch_test_${generateId().replaceAll('-', '')}`
   const databaseUrl = readTestDatabaseUrl()
@@ -93,7 +102,8 @@ describe('workspace file search dispatch PostgreSQL deadlines', () => {
       ON workspace_file_search_revision
       (workspace_id, updated_at, file_id, source_content_updated_at)
       WHERE status = 'pending' AND dispatched_at IS NULL`
-    await connection`CREATE INDEX ON workspace_file_search_revision (workspace_id, dispatched_at)
+    await connection`CREATE INDEX workspace_file_search_revision_active_idx
+      ON workspace_file_search_revision (dispatched_at, workspace_id)
       WHERE status = 'pending' AND dispatched_at IS NOT NULL`
     await connection`CREATE TABLE workspace_file_search_build (id text PRIMARY KEY, expires_at timestamp)`
     await connection`CREATE INDEX workspace_file_search_build_cleanup_idx
@@ -627,22 +637,39 @@ describe('workspace file search dispatch PostgreSQL deadlines', () => {
       await expect(hasWorkspaceFileSearchDispatchWork(new Date())).resolves.toBe(true)
     })
 
-    it('probes the cleanup and active-claim indexes rather than scanning', async () => {
+    it('probes builds and claims through their indexes beside a large settled backlog', async () => {
       await seedIdleDeployment()
+      await connection`INSERT INTO workspace_file_search_revision
+        (file_id, workspace_id, source_content_updated_at, status)
+        SELECT 'settled-' || n, 'workspace-' || (n % 50), '2026-09-16', 'ready'
+        FROM generate_series(1, 100000) n`
+      await connection`INSERT INTO workspace_file_search_build (id, expires_at)
+        SELECT 'published-' || n, NULL FROM generate_series(1, 100000) n`
+      await connection`VACUUM (ANALYZE) workspace_file_search_revision, workspace_file_search_build`
       statements.length = 0
       await expect(hasWorkspaceFileSearchDispatchWork(new Date())).resolves.toBe(false)
 
       const probe = statements.find((statement) => statement.query.includes('"hasWork"'))
       expect(probe).toBeDefined()
-      const plan = await connection.begin(async (tx) => {
-        await tx`SET LOCAL enable_seqscan = off`
-        const rows = await tx.unsafe(`EXPLAIN ${probe?.query}`, probe?.params as never[])
-        return rows.map((row: Record<string, unknown>) => row['QUERY PLAN']).join('\n')
-      })
-
-      expect(plan).toContain('workspace_file_search_build_cleanup_idx')
-      expect(plan).toMatch(/using workspace_file_search_revision_workspace_id_dispatched_at_idx/)
-      expect(plan).not.toMatch(/Seq Scan/)
-    })
+      const [explained] = await connection.unsafe(
+        `EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ${probe?.query}`,
+        probe?.params as never[]
+      )
+      const plan = (explained['QUERY PLAN'] as { Plan: QueryPlan }[])[0].Plan
+      const nodes = planNodes(plan)
+      const indexes = nodes.map((node) => node['Index Name'])
+
+      expect(indexes).toContain('workspace_file_search_build_cleanup_idx')
+      expect(indexes).toContain('workspace_file_search_revision_active_idx')
+      expect(
+        nodes.filter(
+          (node) =>
+            node['Node Type'] === 'Seq Scan' &&
+            node['Relation Name'] !== 'workspace_file_search_dispatch_queue'
+        )
+      ).toEqual([])
+      /** Buffer work, unlike wall-clock time, catches a backlog scan even on a warm local database. */
+      expect(plan['Shared Hit Blocks'] + plan['Shared Read Blocks']).toBeLessThan(100)
+    }, 60_000)
   })
 })

apps/sim/lib/workspace-files/search/enqueue-dispatch.test.ts

@@ -1,60 +1,61 @@
-import {
-  asyncJobsRegionMock,
-  asyncJobsRegionMockFns,
-} from '@sim/testing/mocks/async-jobs-region.mock'
-import { backgroundTaskMock, backgroundTaskMockFns } from '@sim/testing/mocks/background-task.mock'
+import { asyncJobsRegionMock } from '@sim/testing/mocks/async-jobs-region.mock'
+import { backgroundTaskMock } from '@sim/testing/mocks/background-task.mock'
 import { resetEnvFlagsMock, setEnvFlags } from '@sim/testing/mocks/env-flags.mock'
+import {
+  createIdempotentTasksTrigger,
+  triggerSdkMockFns,
+} from '@sim/testing/mocks/trigger-sdk.mock'
 import { afterAll, afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 
 const mocks = vi.hoisted(() => ({
-  dispatch: vi.fn(),
   hasWork: vi.fn(),
 }))
 
 vi.mock('@/lib/core/async-jobs/region', () => asyncJobsRegionMock)
 vi.mock('@/lib/core/utils/background', () => backgroundTaskMock)
 vi.mock('@/lib/workspace-files/search/dispatcher', () => ({
-  dispatchWorkspaceFileSearchIndexJobs: mocks.dispatch,
+  dispatchWorkspaceFileSearchIndexJobs: vi.fn(),
   hasWorkspaceFileSearchDispatchWork: mocks.hasWork,
 }))
 
-import { tasks } from '@trigger.dev/sdk'
 import { enqueueWorkspaceFileSearchDispatch } from '@/lib/workspace-files/search/enqueue-dispatch'
 
-const mockTrigger = vi.mocked(tasks.trigger)
-
 afterAll(resetEnvFlagsMock)
 
 describe('workspace file search dispatcher enqueue', () => {
   beforeEach(() => {
     vi.useFakeTimers()
     vi.setSystemTime(new Date('2026-08-29T12:34:45.000Z'))
     setEnvFlags({ isTriggerDevEnabled: true })
-    asyncJobsRegionMockFns.mockResolveTriggerRegion.mockResolvedValue('us-east-1')
-    mockTrigger.mockResolvedValue({ id: 'run-1' })
+    triggerSdkMockFns.mockTasksTrigger.mockImplementation(createIdempotentTasksTrigger())
+    mocks.hasWork.mockReset()
     mocks.hasWork.mockResolvedValue(true)
   })
 
   afterEach(() => {
     vi.useRealTimers()
   })
 
-  it('waits only for durable Trigger.dev acceptance and does not run the dispatcher inline', async () => {
+  it('returns the Trigger.dev run once it is accepted', async () => {
     await expect(enqueueWorkspaceFileSearchDispatch()).resolves.toEqual({
       triggered: true,
       backend: 'trigger-dev',
       jobId: 'run-1',
     })
+  })
 
-    expect(mockTrigger).toHaveBeenCalledWith('workspace-file-search-dispatch', undefined, {
-      idempotencyKey: 'workspace-file-search-dispatch:29800114',
-      idempotencyKeyTTL: '5m',
-      maxDuration: 60,
-      region: 'us-east-1',
-      ttl: '5m',
+  it('folds ticks into the run of the window their work check was made in', async () => {
+    const first = await enqueueWorkspaceFileSearchDispatch()
+    mocks.hasWork.mockImplementationOnce(async () => {
+      vi.advanceTimersByTime(60_000)
+      return true
     })
-    expect(mocks.dispatch).not.toHaveBeenCalled()
-    expect(backgroundTaskMockFns.mockRunDetached).not.toHaveBeenCalled()
+    const checkedAcrossTheBoundary = await enqueueWorkspaceFileSearchDispatch()
+    const nextWindow = await enqueueWorkspaceFileSearchDispatch()
+
+    expect(first.jobId).toBe('run-1')
+    expect(checkedAcrossTheBoundary.jobId).toBe('run-1')
+    expect(nextWindow.jobId).toBe('run-2')
   })
 
   it.each([
@@ -71,18 +72,16 @@ describe('workspace file search dispatcher enqueue', () => {
         backend: null,
         jobId: null,
       })
-      expect(mockTrigger).not.toHaveBeenCalled()
-      expect(backgroundTaskMockFns.mockRunDetached).not.toHaveBeenCalled()
     }
   )
 
   it('still starts a dispatcher run when the work check fails', async () => {
     mocks.hasWork.mockRejectedValue(new Error('statement timeout'))
 
-    await expect(enqueueWorkspaceFileSearchDispatch()).resolves.toMatchObject({
+    await expect(enqueueWorkspaceFileSearchDispatch()).resolves.toEqual({
       triggered: true,
       backend: 'trigger-dev',
+      jobId: 'run-1',
     })
-    expect(mockTrigger).toHaveBeenCalledTimes(1)
   })
 })

packages/testing/src/mocks/trigger-sdk.mock.ts

@@ -83,6 +83,31 @@ function createMockTask<T extends object>(config: T) {
   }
 }
 
+/**
+ * A `tasks.trigger` implementation that deduplicates like Trigger.dev: a trigger whose task already
+ * holds its `idempotencyKey` returns that run's id, and a new key (or no key) starts a new run
+ * (`run-1`, `run-2`, …). Key TTLs are not modelled: a key is held for the fake's lifetime.
+ *
+ * @example
+ * ```ts
+ * triggerSdkMockFns.mockTasksTrigger.mockImplementation(createIdempotentTasksTrigger())
+ * ```
+ */
+export function createIdempotentTasksTrigger() {
+  const runIdByKey = new Map<string, string>()
+  let runCount = 0
+  return async (taskId: string, _payload?: unknown, options?: unknown): Promise<unknown> => {
+    const key = (options as { idempotencyKey?: unknown } | undefined)?.idempotencyKey
+    const scopedKey = typeof key === 'string' ? `${taskId}:${key}` : null
+    const existing = scopedKey ? runIdByKey.get(scopedKey) : undefined
+    if (existing) return { id: existing }
+    runCount += 1
+    const id = `run-${runCount}`
+    if (scopedKey) runIdByKey.set(scopedKey, id)
+    return { id }
+  }
+}
+
 /**
  * Controllable mock functions for `@trigger.dev/sdk` (globally mocked in `apps/sim/vitest.setup.ts`).
  *