improvement(outbox): only start the processor when work is due or maintenance is scheduled

Author: waleedlatif1

apps/sim/app/api/webhooks/outbox/process/route.ts

@@ -20,6 +20,9 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
   const requestId = generateRequestId()
   try {
     const accepted = await enqueueOutboxProcessor()
+    if (accepted.backend === null) {
+      return NextResponse.json({ success: true, requestId, triggered: false })
+    }
     if (accepted.backend === 'trigger-dev') {
       logger.info('Outbox processor accepted', { jobId: accepted.jobId })
       return NextResponse.json(

apps/sim/lib/core/outbox/constants.ts

@@ -6,3 +6,5 @@ export const OUTBOX_PROCESSOR_INTERVAL_MS = 60_000
 export const OUTBOX_PROCESSOR_CONCURRENCY = Math.ceil(
   (OUTBOX_PROCESSOR_MAX_DURATION_SECONDS * 1000) / OUTBOX_PROCESSOR_INTERVAL_MS
 )
+/** Every Nth scheduled tick runs the processor's recovery, reaping and pruning even with nothing due. */
+export const OUTBOX_MAINTENANCE_EVERY_INTERVALS = 5

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

@@ -1,24 +1,45 @@
 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 { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 
 const hoisted = vi.hoisted(() => ({
   processor: vi.fn(),
 }))
 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) }
+const mocks = {
+  ...hoisted,
+  trigger: vi.mocked(tasks.trigger),
+  hasDueWork: outboxServiceMockFns.mockHasDueOutboxWork,
+}
+
+/** 12:34 is not a maintenance minute (754 % 5 = 4); 12:35 is. */
+const IDLE_MINUTE = new Date('2026-09-16T12:34:45Z')
+const MAINTENANCE_MINUTE = new Date('2026-09-16T12:35:10Z')
+
+const BACKENDS = [
+  { name: 'Trigger.dev', isTriggerDevEnabled: true },
+  { name: 'self-hosted inline', isTriggerDevEnabled: false },
+] as const
+
+function processorStarted() {
+  return mocks.trigger.mock.calls.length + mocks.processor.mock.calls.length
+}
 
 describe('outbox processor enqueue', () => {
   beforeEach(() => {
     vi.useFakeTimers()
-    vi.setSystemTime(new Date('2026-09-16T12:34:45Z'))
+    vi.setSystemTime(IDLE_MINUTE)
     setEnvFlags({ isTriggerDevEnabled: true })
     mocks.trigger.mockResolvedValue({ id: 'run-1' })
+    mocks.hasDueWork.mockReset()
+    mocks.hasDueWork.mockResolvedValue(true)
   })
   afterEach(() => vi.useRealTimers())
 
@@ -49,4 +70,35 @@ describe('outbox processor enqueue', () => {
     await expect(enqueueOutboxProcessor()).resolves.toEqual({ backend: 'inline', output })
     expect(mocks.trigger).not.toHaveBeenCalled()
   })
+
+  it.each(BACKENDS)(
+    'starts no $name processor on an idle queue outside the maintenance window',
+    async ({ isTriggerDevEnabled }) => {
+      setEnvFlags({ isTriggerDevEnabled })
+      mocks.hasDueWork.mockResolvedValue(false)
+      await expect(enqueueOutboxProcessor()).resolves.toEqual({ backend: null, triggered: false })
+      expect(processorStarted()).toBe(0)
+    }
+  )
+
+  it.each(BACKENDS)(
+    'starts the $name processor on an idle queue in the maintenance window',
+    async ({ isTriggerDevEnabled }) => {
+      setEnvFlags({ isTriggerDevEnabled })
+      vi.setSystemTime(MAINTENANCE_MINUTE)
+      mocks.hasDueWork.mockResolvedValue(false)
+      await enqueueOutboxProcessor()
+      expect(processorStarted()).toBe(1)
+    }
+  )
+
+  it.each(BACKENDS)(
+    'starts the $name processor when the work check fails',
+    async ({ isTriggerDevEnabled }) => {
+      setEnvFlags({ isTriggerDevEnabled })
+      mocks.hasDueWork.mockRejectedValue(new Error('connection refused'))
+      await enqueueOutboxProcessor()
+      expect(processorStarted()).toBe(1)
+    }
+  )
 })

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

@@ -1,17 +1,48 @@
+import { createLogger } from '@sim/logger'
+import { getErrorMessage } from '@sim/utils/errors'
 import { isTriggerDevEnabled } from '@/lib/core/config/env-flags'
 import {
+  OUTBOX_MAINTENANCE_EVERY_INTERVALS,
   OUTBOX_PROCESSOR_INTERVAL_MS,
   OUTBOX_PROCESSOR_MAX_DURATION_SECONDS,
 } from '@/lib/core/outbox/constants'
 import type { OutboxProcessorResult } from '@/lib/core/outbox/processor'
+import { hasDueOutboxWork } from '@/lib/core/outbox/service'
 import type { processOutboxTask } from '@/background/process-outbox'
 
+const logger = createLogger('OutboxProcessorEnqueue')
+
 type OutboxProcessorEnqueueResult =
   | { backend: 'trigger-dev'; jobId: string }
   | { backend: 'inline'; output: OutboxProcessorResult }
+  | { backend: null; triggered: false }
+
+/**
+ * Every tick in the maintenance window runs, so document recovery, background-work reaping and
+ * pruning keep their cadence on an idle queue. Other ticks run only when events are due or a lease
+ * is stale; a gate that cannot answer runs anyway, since billing, seat sync, invitations and
+ * document dispatch all ride the outbox.
+ */
+async function shouldRunOutboxProcessor(now: Date, scheduleWindow: number): Promise<boolean> {
+  if (scheduleWindow % OUTBOX_MAINTENANCE_EVERY_INTERVALS === 0) return true
+  try {
+    return await hasDueOutboxWork(now)
+  } catch (error) {
+    logger.warn('Outbox work check failed; running the processor anyway', {
+      error: getErrorMessage(error),
+    })
+    return true
+  }
+}
 
 /** The database owns delivery state; the cron request waits only for durable worker acceptance. */
 export async function enqueueOutboxProcessor(): Promise<OutboxProcessorEnqueueResult> {
+  const now = new Date()
+  const scheduleWindow = Math.floor(now.getTime() / OUTBOX_PROCESSOR_INTERVAL_MS)
+  if (!(await shouldRunOutboxProcessor(now, scheduleWindow))) {
+    return { backend: null, triggered: false }
+  }
+
   if (!isTriggerDevEnabled) {
     const { runOutboxProcessor } = await import('@/lib/core/outbox/processor')
     return { backend: 'inline', output: await runOutboxProcessor() }
@@ -21,7 +52,6 @@ export async function enqueueOutboxProcessor(): Promise<OutboxProcessorEnqueueRe
     import('@trigger.dev/sdk'),
     import('@/lib/core/async-jobs/region'),
   ])
-  const scheduleWindow = Math.floor(Date.now() / OUTBOX_PROCESSOR_INTERVAL_MS)
   const handle = await tasks.trigger<typeof processOutboxTask>('process-outbox', undefined, {
     idempotencyKey: `process-outbox:${scheduleWindow}`,
     idempotencyKeyTTL: '5m',

apps/sim/lib/core/outbox/queries.ts

@@ -1,7 +1,18 @@
+import { db } from '@sim/db'
 import { outboxEvent } from '@sim/db/schema'
-import { sql } from 'drizzle-orm'
+import { and, eq, exists, lte, or, type SQL, sql } from 'drizzle-orm'
 
 const MAX_READY_EVENT_TYPES = 128
+/** How long a `processing` lease may go without a terminal write before the reaper reclaims it. */
+export const STUCK_PROCESSING_THRESHOLD_MS = 10 * 60 * 1000
+
+/** A `processing` row whose lease the reaper may reclaim at `now`. */
+export function isStuckProcessing(now: Date): SQL | undefined {
+  return and(
+    eq(outboxEvent.status, 'processing'),
+    lte(outboxEvent.lockedAt, new Date(now.getTime() - STUCK_PROCESSING_THRESHOLD_MS))
+  )
+}
 
 /**
  * Walks the pending index one type at a time, reading only its earliest availability.
@@ -42,3 +53,18 @@ export function readyEventTypesQuery(now: Date) {
     LIMIT ${MAX_READY_EVENT_TYPES}
   `
 }
+
+/**
+ * One row, `due`: whether `processOutboxEvents` would act at `now`. The pending leg is the
+ * claim phase's own discovery walk, so it reads only each type's head and cannot be planned as a
+ * scan of future or completed rows; the lease leg is the reaper's predicate, over the few rows in
+ * `processing`.
+ */
+export function dueOutboxWorkQuery(now: Date) {
+  const staleLease = db
+    .select({ id: outboxEvent.id })
+    .from(outboxEvent)
+    .where(isStuckProcessing(now))
+    .limit(1)
+  return sql`SELECT ${or(sql`EXISTS (${readyEventTypesQuery(now)})`, exists(staleLease))} AS due`
+}

apps/sim/lib/core/outbox/retention.ts

@@ -10,9 +10,11 @@ const logger = createLogger('OutboxRetention')
 /** How long a completed event stays readable for operators after it was enqueued. */
 export const COMPLETED_OUTBOX_RETENTION_MS = 7 * 24 * 60 * 60_000
 /**
- * Rows deleted per type per run. The processor runs once a minute, so each type drains at most
- * 1,000 rows × 1,440 runs = 1.44M rows a day: a steady trickle whose WAL and dead tuples
- * autovacuum absorbs, yet five times what recovery can enqueue (200 per run × 1,440 runs).
+ * Rows deleted per type per run. The processor runs on every fifth minute's maintenance tick and
+ * on any other minute with due work, so each type drains at least 1,000 rows × 288 runs = 288k
+ * rows a day and at most 1.44M: a steady trickle whose WAL and dead tuples autovacuum absorbs.
+ * Recovery enqueues only inside a run (at most 200), so pruning keeps five times its pace at any
+ * run count.
  */
 export const OUTBOX_PRUNE_BATCH_SIZE = 1_000
 

apps/sim/lib/core/outbox/service.integration.ts

@@ -19,8 +19,13 @@ vi.mock('@sim/db', () => ({
   },
 }))
 
-import { readyEventTypesQuery } from '@/lib/core/outbox/queries'
 import {
+  dueOutboxWorkQuery,
+  readyEventTypesQuery,
+  STUCK_PROCESSING_THRESHOLD_MS,
+} from '@/lib/core/outbox/queries'
+import {
+  hasDueOutboxWork,
   type OutboxHandler,
   processOutboxEvents,
   withOutboxHandlerTimeout,
@@ -386,4 +391,62 @@ describe('outbox scheduling in PostgreSQL', () => {
     const second = await processOutboxEvents({}, { batchSize: 0 })
     expect(second.reaped).toBe(5)
   })
+
+  describe('due-work gate', () => {
+    const now = new Date('2026-09-16T12:34:00.000Z')
+
+    async function insertRow(
+      status: 'pending' | 'processing' | 'completed' | 'dead_letter',
+      times: { availableAt?: Date; lockedAt?: Date | null }
+    ) {
+      eventTypes.add('test.outbox.gate')
+      await db.insert(outboxEvent).values({
+        id: generateId(),
+        eventType: 'test.outbox.gate',
+        payload: {},
+        status,
+        createdAt: new Date(now.getTime() - 60 * 60_000),
+        availableAt: times.availableAt ?? new Date(now.getTime() - 60 * 60_000),
+        lockedAt: times.lockedAt ?? null,
+      })
+    }
+
+    it('reports no work for future, freshly leased, completed and dead-lettered rows', async () => {
+      expect(await hasDueOutboxWork(now)).toBe(false)
+      await insertRow('pending', { availableAt: new Date(now.getTime() + 1) })
+      await insertRow('processing', {
+        lockedAt: new Date(now.getTime() - STUCK_PROCESSING_THRESHOLD_MS + 1),
+      })
+      await insertRow('completed', {})
+      await insertRow('dead_letter', {})
+      expect(await hasDueOutboxWork(now)).toBe(false)
+    })
+
+    it('reports a pending event due exactly now, as the claim phase would take it', async () => {
+      await insertRow('pending', { availableAt: now })
+      expect(await hasDueOutboxWork(now)).toBe(true)
+    })
+
+    it('reports a lease exactly at the stale threshold, as the reaper would reclaim it', async () => {
+      await insertRow('processing', {
+        lockedAt: new Date(now.getTime() - STUCK_PROCESSING_THRESHOLD_MS),
+      })
+      expect(await hasDueOutboxWork(now)).toBe(true)
+    })
+
+    it('probes both legs through indexes beside a large completed and future backlog', async () => {
+      await seedBacklog('test.outbox.gate', 50_000, 'completed')
+      await seedBacklog('test.outbox.gate', 50_000, 'pending', new Date(Date.now() + 60 * 60_000))
+      await connection`VACUUM (ANALYZE) outbox_event`
+      const gateNow = new Date()
+      expect(await hasDueOutboxWork(gateNow)).toBe(false)
+
+      const plans = await db.execute<{ 'QUERY PLAN': { Plan: QueryPlan }[] }>(sql`
+        EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ${dueOutboxWorkQuery(gateNow)}
+      `)
+      const plan = plans[0]['QUERY PLAN'][0].Plan
+      expect(planNodes(plan).some((node) => node['Node Type'] === 'Seq Scan')).toBe(false)
+      expect(plan['Shared Hit Blocks'] + plan['Shared Read Blocks']).toBeLessThan(100)
+    }, 60_000)
+  })
 })

apps/sim/lib/core/outbox/service.ts

@@ -5,7 +5,12 @@ import { describeError, toError } from '@sim/utils/errors'
 import { generateId } from '@sim/utils/id'
 import { truncate } from '@sim/utils/string'
 import { and, asc, desc, eq, inArray, lte, sql } from 'drizzle-orm'
-import { readyEventTypesQuery } from '@/lib/core/outbox/queries'
+import {
+  dueOutboxWorkQuery,
+  isStuckProcessing,
+  readyEventTypesQuery,
+  STUCK_PROCESSING_THRESHOLD_MS,
+} from '@/lib/core/outbox/queries'
 
 const logger = createLogger('OutboxService')
 
@@ -24,7 +29,6 @@ const MAX_REAPED_EVENTS = 1_000
 function toPersistedHandlerError(error: unknown): string {
   return truncate(toError(error).message.split(/\nparams: /)[0], MAX_PERSISTED_ERROR_LENGTH)
 }
-const STUCK_PROCESSING_THRESHOLD_MS = 10 * 60 * 1000 // 10 minutes
 const MAX_BACKOFF_MS = 60 * 60 * 1000 // 1 hour
 const BASE_BACKOFF_MS = 1000 // 1 second, doubled per attempt
 /** Ordinary handlers keep a short window; longer handlers explicitly opt in below the stale-lease limit. */
@@ -542,18 +546,26 @@ export async function processOutboxEventById(
   return runHandler(event, handlers)
 }
 
+/**
+ * Whether {@link processOutboxEvents} would find anything at `now`: a pending event the claim
+ * phase may take, or a stale lease the reaper may reclaim.
+ */
+export async function hasDueOutboxWork(now: Date): Promise<boolean> {
+  const [row] = await db.execute<{ due: boolean }>(dueOutboxWorkQuery(now))
+  return row?.due === true
+}
+
 /**
  * Reaper: move `processing` rows whose worker died (stale `lockedAt`)
  * back to `pending` so another worker can pick them up. Without this,
  * a SIGKILL between claim and result-write would permanently strand
  * the row in `processing`.
  */
 async function reapStuckProcessingRows(): Promise<number> {
-  const stuckBefore = new Date(Date.now() - STUCK_PROCESSING_THRESHOLD_MS)
   const stuckRows = db
     .select({ id: outboxEvent.id })
     .from(outboxEvent)
-    .where(and(eq(outboxEvent.status, 'processing'), lte(outboxEvent.lockedAt, stuckBefore)))
+    .where(isStuckProcessing(new Date()))
     .orderBy(asc(outboxEvent.lockedAt), asc(outboxEvent.id))
     .limit(MAX_REAPED_EVENTS)
     .for('update', { skipLocked: true })

packages/testing/src/mocks/outbox-service.mock.ts

@@ -69,6 +69,7 @@ export const outboxServiceMockFns = {
   }),
   mockFindDeadLetteredEvents: vi.fn(),
   mockHasInflightOutboxEvent: vi.fn(),
+  mockHasDueOutboxWork: vi.fn(),
   mockProcessOutboxEvents: vi.fn(),
   mockProcessOutboxEventById: vi.fn(),
 }
@@ -97,6 +98,7 @@ export const outboxServiceMock = {
   outboxPayloadHasSourceOperationId: outboxServiceMockFns.mockOutboxPayloadHasSourceOperationId,
   findDeadLetteredEvents: outboxServiceMockFns.mockFindDeadLetteredEvents,
   hasInflightOutboxEvent: outboxServiceMockFns.mockHasInflightOutboxEvent,
+  hasDueOutboxWork: outboxServiceMockFns.mockHasDueOutboxWork,
   processOutboxEvents: outboxServiceMockFns.mockProcessOutboxEvents,
   processOutboxEventById: outboxServiceMockFns.mockProcessOutboxEventById,
 }