fix(outbox): process cron batches outside HTTP requests (#7911)

* fix(outbox): process cron batches outside HTTP requests

* fix(outbox): preserve polling capacity with larger workers

Author: icecrasher321

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

@@ -0,0 +1,62 @@
+/**
+ * @vitest-environment node
+ */
+import { createMockRequest } from '@sim/testing'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const mocks = vi.hoisted(() => ({ enqueue: vi.fn(), verifyCronAuth: vi.fn() }))
+vi.mock('@/lib/auth/internal', () => ({ verifyCronAuth: mocks.verifyCronAuth }))
+vi.mock('@/lib/core/outbox/enqueue', () => ({ enqueueOutboxProcessor: mocks.enqueue }))
+
+import { GET } from '@/app/api/webhooks/outbox/process/route'
+
+const request = () =>
+  createMockRequest('GET', undefined, {}, 'http://localhost:3000/api/webhooks/outbox/process')
+
+describe('outbox cron route', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    mocks.verifyCronAuth.mockReturnValue(null)
+  })
+  it('authenticates before accepting any background work', async () => {
+    mocks.verifyCronAuth.mockReturnValue(new Response(null, { status: 401 }))
+    expect((await GET(request())).status).toBe(401)
+    expect(mocks.enqueue).not.toHaveBeenCalled()
+  })
+  it('acknowledges a durably accepted task with 202', async () => {
+    mocks.enqueue.mockResolvedValue({ backend: 'trigger-dev', jobId: 'run-1' })
+    const response = await GET(request())
+    expect(response.status).toBe(202)
+    await expect(response.json()).resolves.toEqual({
+      success: true,
+      requestId: expect.any(String),
+      triggered: true,
+      backend: 'trigger-dev',
+      jobId: 'run-1',
+    })
+  })
+  it('preserves the existing 200 response for synchronous self-hosted runs', async () => {
+    const output = {
+      result: { processed: 1, retried: 0, deadLettered: 0, leaseLost: 0, reaped: 0 },
+      reapedBackgroundWork: 0,
+      recoveredDocuments: 0,
+    }
+    mocks.enqueue.mockResolvedValue({ backend: 'inline', output })
+    const response = await GET(request())
+    expect(response.status).toBe(200)
+    await expect(response.json()).resolves.toEqual({
+      success: true,
+      requestId: expect.any(String),
+      ...output,
+    })
+  })
+  it('returns an error when durable acceptance fails', async () => {
+    mocks.enqueue.mockRejectedValue(new Error('Trigger unavailable'))
+    const response = await GET(request())
+    expect(response.status).toBe(500)
+    await expect(response.json()).resolves.toMatchObject({
+      success: false,
+      error: 'Trigger unavailable',
+    })
+  })
+})

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

@@ -1,118 +1,35 @@
-import { db } from '@sim/db'
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
 import { type NextRequest, NextResponse } from 'next/server'
-import { adminInvitationOperationOutboxHandlers } from '@/lib/admin/invitation-operation'
-import { adminMemberOperationOutboxHandlers } from '@/lib/admin/member-operation'
 import { verifyCronAuth } from '@/lib/auth/internal'
-import { enterpriseOwnerClaimOutboxHandlers } from '@/lib/billing/enterprise-owner-claim'
-import { enterpriseIssuanceOutboxHandlers } from '@/lib/billing/enterprise-provisioning'
-import { membershipBillingOutboxHandlers } from '@/lib/billing/organizations/membership-reconciliation'
-import { billingOutboxHandlers } from '@/lib/billing/webhooks/outbox-handlers'
-import { processOutboxEvents } from '@/lib/core/outbox/service'
-import { DeadlineExceededError } from '@/lib/core/utils/deadline'
+import { enqueueOutboxProcessor } from '@/lib/core/outbox/enqueue'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
-import { directGrantOutboxHandlers } from '@/lib/invitations/direct-grant'
-import { slackSearchOutboxHandlers } from '@/lib/knowledge/application/slack-search/outbox'
-import { getConnectorFailureDiagnostic } from '@/lib/knowledge/connectors/connector-error'
-import { knowledgeDocumentProcessingOutboxHandlers } from '@/lib/knowledge/documents/processing-outbox-handler'
-import { recoverKnowledgeDocumentProcessing } from '@/lib/knowledge/documents/processing-recovery'
-import { organizationResourceCleanupOutboxHandlers } from '@/lib/organizations/resource-cleanup'
-import { permissionAccessRequestOutboxHandlers } from '@/lib/permission-access-requests/notifications'
-import { workspaceFileLiveDocOutboxHandlers } from '@/lib/uploads/contexts/workspace/workspace-file-live-doc-outbox'
-import { workspaceFileStorageCleanupOutboxHandlers } from '@/lib/uploads/contexts/workspace/workspace-file-storage-cleanup-outbox'
-import { workflowDeploymentOutboxHandlers } from '@/lib/workflows/deployment-outbox'
-import { invitationMigrationOutboxHandlers } from '@/lib/workspaces/admin-move'
-import { workspaceOperationOutboxHandlers } from '@/lib/workspaces/operations/outbox'
-import { forkContentOutboxHandlers } from '@/ee/workspace-forking/application/content-outbox'
-import { reapStaleBackgroundWork } from '@/ee/workspace-forking/lib/background-work/store'
 
 const logger = createLogger('OutboxProcessorAPI')
 
 export const dynamic = 'force-dynamic'
+/** Self-hosted deployments without Trigger.dev retain the synchronous processing window. */
 export const maxDuration = 800
 
-const handlers = {
-  ...slackSearchOutboxHandlers,
-  ...adminInvitationOperationOutboxHandlers,
-  ...adminMemberOperationOutboxHandlers,
-  ...billingOutboxHandlers,
-  ...membershipBillingOutboxHandlers,
-  ...enterpriseIssuanceOutboxHandlers,
-  ...enterpriseOwnerClaimOutboxHandlers,
-  ...invitationMigrationOutboxHandlers,
-  ...directGrantOutboxHandlers,
-  ...knowledgeDocumentProcessingOutboxHandlers,
-  ...organizationResourceCleanupOutboxHandlers,
-  ...permissionAccessRequestOutboxHandlers,
-  ...workspaceFileLiveDocOutboxHandlers,
-  ...workspaceFileStorageCleanupOutboxHandlers,
-  ...workflowDeploymentOutboxHandlers,
-  ...workspaceOperationOutboxHandlers,
-  ...forkContentOutboxHandlers,
-} as const
-
+/** The cron secret authorizes this lifecycle endpoint; hosted processing runs outside the HTTP request. */
 export const GET = withRouteHandler(async (request: NextRequest) => {
-  const requestId = generateRequestId()
+  const authError = verifyCronAuth(request, 'Outbox processor')
+  if (authError) return authError
 
+  const requestId = generateRequestId()
   try {
-    const authError = verifyCronAuth(request, 'Outbox processor')
-    if (authError) {
-      return authError
-    }
-
-    const startedAt = Date.now()
-    const result = await processOutboxEvents(handlers, {
-      batchSize: 500,
-      maxRuntimeMs: 760_000,
-      minRemainingMs: 95_000,
-    })
-
-    let recoveredDocuments = 0
-    try {
-      if (Date.now() - startedAt < 770_000) {
-        recoveredDocuments = await recoverKnowledgeDocumentProcessing()
-      }
-    } catch (error) {
-      logger.error('Stored document recovery failed', {
-        requestId,
-        error: getConnectorFailureDiagnostic(error) ?? {
-          category: error instanceof DeadlineExceededError ? 'deadline' : 'internal',
-          message:
-            error instanceof DeadlineExceededError
-              ? error.message
-              : 'Unexpected stored-document recovery failure',
-        },
-      })
-    }
-
-    // Reap fork background-work rows stuck `processing` past their TTL (worker crash /
-    // restart has no in-task hook). Independent of the outbox; a failure here must not
-    // fail the outbox run, so it's guarded separately.
-    let reapedBackgroundWork = 0
-    try {
-      reapedBackgroundWork = await reapStaleBackgroundWork(db)
-    } catch (error) {
-      logger.error('Background-work reap failed', { requestId, error: toError(error).message })
+    const accepted = await enqueueOutboxProcessor()
+    if (accepted.backend === 'trigger-dev') {
+      logger.info('Outbox processor accepted', { jobId: accepted.jobId })
+      return NextResponse.json(
+        { success: true, requestId, triggered: true, ...accepted },
+        { status: 202 }
+      )
     }
-
-    logger.info('Outbox processing completed', {
-      requestId,
-      ...result,
-      reapedBackgroundWork,
-      recoveredDocuments,
-    })
-
-    return NextResponse.json({
-      success: true,
-      requestId,
-      result,
-      reapedBackgroundWork,
-      recoveredDocuments,
-    })
+    return NextResponse.json({ success: true, requestId, ...accepted.output })
   } catch (error) {
-    logger.error('Outbox processing failed', { requestId, error: toError(error).message })
+    logger.error('Outbox processing failed', { error: toError(error).message })
     return NextResponse.json(
       { success: false, requestId, error: toError(error).message },
       { status: 500 }

apps/sim/background/process-outbox.test.ts

@@ -0,0 +1,30 @@
+/**
+ * @vitest-environment node
+ */
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const mocks = vi.hoisted(() => ({ processor: vi.fn() }))
+vi.mock('@trigger.dev/sdk', () => ({ task: (config: unknown) => config }))
+vi.mock('@/lib/core/outbox/processor', () => ({ runOutboxProcessor: mocks.processor }))
+
+import { processOutboxTask } from '@/background/process-outbox'
+
+describe('outbox processor task', () => {
+  beforeEach(() => vi.clearAllMocks())
+  it('bounds worker concurrency and lets the durable outbox own event retries', async () => {
+    expect(processOutboxTask).toMatchObject({
+      id: 'process-outbox',
+      machine: 'medium-2x',
+      maxDuration: 900,
+      retry: { maxAttempts: 1 },
+      queue: { name: 'process-outbox', concurrencyLimit: 15 },
+    })
+    const output = { result: { processed: 3 }, recoveredDocuments: 0, reapedBackgroundWork: 0 }
+    mocks.processor.mockResolvedValueOnce(output)
+    await expect(processOutboxTask.run()).resolves.toEqual(output)
+  })
+  it('surfaces processor failures to Trigger', async () => {
+    mocks.processor.mockRejectedValueOnce(new Error('database unavailable'))
+    await expect(processOutboxTask.run()).rejects.toThrow('database unavailable')
+  })
+})

apps/sim/background/process-outbox.ts

@@ -0,0 +1,20 @@
+import { task } from '@trigger.dev/sdk'
+import {
+  OUTBOX_PROCESSOR_CONCURRENCY,
+  OUTBOX_PROCESSOR_MAX_DURATION_SECONDS,
+} from '@/lib/core/outbox/constants'
+import { runOutboxProcessor } from '@/lib/core/outbox/processor'
+
+/** Runs bounded outbox delivery beyond the cron caller's HTTP deadline. */
+export const processOutboxTask = task({
+  id: 'process-outbox',
+  machine: 'medium-2x',
+  maxDuration: OUTBOX_PROCESSOR_MAX_DURATION_SECONDS,
+  /** Per-event retries live in the outbox; a later cron tick recovers interrupted work. */
+  retry: { maxAttempts: 1 },
+  queue: {
+    name: 'process-outbox',
+    concurrencyLimit: OUTBOX_PROCESSOR_CONCURRENCY,
+  },
+  run: () => runOutboxProcessor(),
+})

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

@@ -0,0 +1,8 @@
+export const OUTBOX_PROCESSOR_MAX_RUNTIME_MS = 760_000
+export const OUTBOX_PROCESSOR_RECOVERY_CUTOFF_MS = 770_000
+export const OUTBOX_PROCESSOR_MAX_DURATION_SECONDS = 900
+export const OUTBOX_PROCESSOR_INTERVAL_MS = 60_000
+/** Allow every scheduled tick to start even when earlier workers use their full execution window. */
+export const OUTBOX_PROCESSOR_CONCURRENCY = Math.ceil(
+  (OUTBOX_PROCESSOR_MAX_DURATION_SECONDS * 1000) / OUTBOX_PROCESSOR_INTERVAL_MS
+)

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

@@ -0,0 +1,73 @@
+/**
+ * @vitest-environment node
+ */
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+
+const mocks = vi.hoisted(() => ({
+  trigger: vi.fn(),
+  processor: vi.fn(),
+  enabled: true,
+}))
+vi.mock('@trigger.dev/sdk', () => ({ tasks: { trigger: mocks.trigger } }))
+vi.mock('@/lib/core/config/env-flags', () => ({
+  get isTriggerDevEnabled() {
+    return mocks.enabled
+  },
+}))
+vi.mock('@/lib/core/async-jobs/region', () => ({ resolveTriggerRegion: async () => 'us-east-1' }))
+vi.mock('@/lib/core/outbox/processor', () => ({ runOutboxProcessor: mocks.processor }))
+
+import { enqueueOutboxProcessor } from '@/lib/core/outbox/enqueue'
+
+describe('outbox processor enqueue', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    vi.useFakeTimers()
+    vi.setSystemTime(new Date('2026-09-16T12:34:45Z'))
+    mocks.enabled = true
+    mocks.trigger.mockResolvedValue({ id: 'run-1' })
+  })
+  afterEach(() => vi.useRealTimers())
+
+  it('returns durable acceptance without doing outbox work in the request', async () => {
+    await expect(enqueueOutboxProcessor()).resolves.toEqual({
+      backend: 'trigger-dev',
+      jobId: 'run-1',
+    })
+    expect(mocks.trigger).toHaveBeenCalledWith('process-outbox', undefined, {
+      idempotencyKey: `process-outbox:${Math.floor(Date.now() / 60_000)}`,
+      idempotencyKeyTTL: '5m',
+      maxDuration: 900,
+      region: 'us-east-1',
+    })
+    expect(mocks.processor).not.toHaveBeenCalled()
+  })
+
+  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('fails closed on an enqueue error without starting concurrent inline work', async () => {
+    mocks.trigger.mockRejectedValueOnce(new Error('Trigger unavailable'))
+    await expect(enqueueOutboxProcessor()).rejects.toThrow('Trigger unavailable')
+    expect(mocks.processor).not.toHaveBeenCalled()
+  })
+
+  it('preserves synchronous processing for self-hosted deployments without Trigger', async () => {
+    mocks.enabled = false
+    const output = {
+      result: { processed: 4, retried: 0, deadLettered: 0, leaseLost: 0, reaped: 0 },
+      recoveredDocuments: 2,
+      reapedBackgroundWork: 1,
+    }
+    mocks.processor.mockResolvedValueOnce(output)
+    await expect(enqueueOutboxProcessor()).resolves.toEqual({ backend: 'inline', output })
+    expect(mocks.trigger).not.toHaveBeenCalled()
+  })
+})

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

@@ -0,0 +1,32 @@
+import { isTriggerDevEnabled } from '@/lib/core/config/env-flags'
+import {
+  OUTBOX_PROCESSOR_INTERVAL_MS,
+  OUTBOX_PROCESSOR_MAX_DURATION_SECONDS,
+} from '@/lib/core/outbox/constants'
+import type { OutboxProcessorResult } from '@/lib/core/outbox/processor'
+import type { processOutboxTask } from '@/background/process-outbox'
+
+type OutboxProcessorEnqueueResult =
+  | { backend: 'trigger-dev'; jobId: string }
+  | { backend: 'inline'; output: OutboxProcessorResult }
+
+/** The database owns delivery state; the cron request waits only for durable worker acceptance. */
+export async function enqueueOutboxProcessor(): Promise<OutboxProcessorEnqueueResult> {
+  if (!isTriggerDevEnabled) {
+    const { runOutboxProcessor } = await import('@/lib/core/outbox/processor')
+    return { backend: 'inline', output: await runOutboxProcessor() }
+  }
+
+  const [{ tasks }, { resolveTriggerRegion }] = await Promise.all([
+    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',
+    maxDuration: OUTBOX_PROCESSOR_MAX_DURATION_SECONDS,
+    region: await resolveTriggerRegion(),
+  })
+  return { backend: 'trigger-dev', jobId: handle.id }
+}