From 6ead6b7e7c8a08b74218d3bc215bb8b6baccd779 Mon Sep 17 00:00:00 2001 From: Vikhyath Mondreti Date: Wed, 16 Sep 2026 19:16:49 -0700 Subject: [PATCH 1/2] fix(outbox): process cron batches outside HTTP requests --- .../api/webhooks/outbox/process/route.test.ts | 62 +++++++++ .../app/api/webhooks/outbox/process/route.ts | 113 +++-------------- apps/sim/background/process-outbox.test.ts | 29 +++++ apps/sim/background/process-outbox.ts | 18 +++ apps/sim/lib/core/outbox/constants.ts | 4 + apps/sim/lib/core/outbox/enqueue.test.ts | 74 +++++++++++ apps/sim/lib/core/outbox/enqueue.ts | 34 +++++ apps/sim/lib/core/outbox/processor.test.ts | 119 ++++++++++++++++++ apps/sim/lib/core/outbox/processor.ts | 101 +++++++++++++++ 9 files changed, 456 insertions(+), 98 deletions(-) create mode 100644 apps/sim/app/api/webhooks/outbox/process/route.test.ts create mode 100644 apps/sim/background/process-outbox.test.ts create mode 100644 apps/sim/background/process-outbox.ts create mode 100644 apps/sim/lib/core/outbox/constants.ts create mode 100644 apps/sim/lib/core/outbox/enqueue.test.ts create mode 100644 apps/sim/lib/core/outbox/enqueue.ts create mode 100644 apps/sim/lib/core/outbox/processor.test.ts create mode 100644 apps/sim/lib/core/outbox/processor.ts diff --git a/apps/sim/app/api/webhooks/outbox/process/route.test.ts b/apps/sim/app/api/webhooks/outbox/process/route.test.ts new file mode 100644 index 00000000000..9000a9f59ec --- /dev/null +++ b/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', + }) + }) +}) diff --git a/apps/sim/app/api/webhooks/outbox/process/route.ts b/apps/sim/app/api/webhooks/outbox/process/route.ts index fe9b2c3a4f0..6f590e0d33e 100644 --- a/apps/sim/app/api/webhooks/outbox/process/route.ts +++ b/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 } diff --git a/apps/sim/background/process-outbox.test.ts b/apps/sim/background/process-outbox.test.ts new file mode 100644 index 00000000000..762aa90b8fa --- /dev/null +++ b/apps/sim/background/process-outbox.test.ts @@ -0,0 +1,29 @@ +/** + * @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', + maxDuration: 900, + retry: { maxAttempts: 1 }, + queue: { name: 'process-outbox', concurrencyLimit: 4 }, + }) + 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') + }) +}) diff --git a/apps/sim/background/process-outbox.ts b/apps/sim/background/process-outbox.ts new file mode 100644 index 00000000000..3acfb29ba1c --- /dev/null +++ b/apps/sim/background/process-outbox.ts @@ -0,0 +1,18 @@ +import { task } from '@trigger.dev/sdk' +import { 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: 'small-2x', + maxDuration: OUTBOX_PROCESSOR_MAX_DURATION_SECONDS, + /** Per-event retries live in the outbox; a later cron tick recovers interrupted work. */ + retry: { maxAttempts: 1 }, + /** Keep other deliveries moving while a worker handles a long operation. */ + queue: { + name: 'process-outbox', + concurrencyLimit: 4, + }, + run: () => runOutboxProcessor(), +}) diff --git a/apps/sim/lib/core/outbox/constants.ts b/apps/sim/lib/core/outbox/constants.ts new file mode 100644 index 00000000000..6a61b30f3ad --- /dev/null +++ b/apps/sim/lib/core/outbox/constants.ts @@ -0,0 +1,4 @@ +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 diff --git a/apps/sim/lib/core/outbox/enqueue.test.ts b/apps/sim/lib/core/outbox/enqueue.test.ts new file mode 100644 index 00000000000..d569e7ad598 --- /dev/null +++ b/apps/sim/lib/core/outbox/enqueue.test.ts @@ -0,0 +1,74 @@ +/** + * @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', + ttl: '1m', + }) + 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() + }) +}) diff --git a/apps/sim/lib/core/outbox/enqueue.ts b/apps/sim/lib/core/outbox/enqueue.ts new file mode 100644 index 00000000000..e597c863dcb --- /dev/null +++ b/apps/sim/lib/core/outbox/enqueue.ts @@ -0,0 +1,34 @@ +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 { + 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('process-outbox', undefined, { + idempotencyKey: `process-outbox:${scheduleWindow}`, + idempotencyKeyTTL: '5m', + maxDuration: OUTBOX_PROCESSOR_MAX_DURATION_SECONDS, + region: await resolveTriggerRegion(), + /** Expired ticks are superseded by the next poll of the same durable outbox. */ + ttl: '1m', + }) + return { backend: 'trigger-dev', jobId: handle.id } +} diff --git a/apps/sim/lib/core/outbox/processor.test.ts b/apps/sim/lib/core/outbox/processor.test.ts new file mode 100644 index 00000000000..e63b025eca5 --- /dev/null +++ b/apps/sim/lib/core/outbox/processor.test.ts @@ -0,0 +1,119 @@ +/** + * @vitest-environment node + */ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +const mocks = vi.hoisted(() => ({ + process: vi.fn(), + recover: vi.fn(), + reap: vi.fn(), +})) +vi.mock('@/lib/core/outbox/service', () => ({ processOutboxEvents: mocks.process })) +vi.mock('@/lib/knowledge/documents/processing-recovery', () => ({ + recoverKnowledgeDocumentProcessing: mocks.recover, +})) +vi.mock('@/ee/workspace-forking/lib/background-work/store', () => ({ + reapStaleBackgroundWork: mocks.reap, +})) +vi.mock('@/lib/knowledge/connectors/connector-error', () => ({ + getConnectorFailureDiagnostic: () => undefined, +})) +vi.mock('@/lib/admin/invitation-operation', () => ({ adminInvitationOperationOutboxHandlers: {} })) +vi.mock('@/lib/admin/member-operation', () => ({ adminMemberOperationOutboxHandlers: {} })) +vi.mock('@/lib/billing/enterprise-owner-claim', () => ({ enterpriseOwnerClaimOutboxHandlers: {} })) +vi.mock('@/lib/billing/enterprise-provisioning', () => ({ enterpriseIssuanceOutboxHandlers: {} })) +vi.mock('@/lib/billing/organizations/membership-reconciliation', () => ({ + membershipBillingOutboxHandlers: {}, +})) +vi.mock('@/lib/billing/webhooks/outbox-handlers', () => ({ billingOutboxHandlers: {} })) +vi.mock('@/lib/invitations/direct-grant', () => ({ directGrantOutboxHandlers: {} })) +vi.mock('@/lib/knowledge/application/slack-search/outbox', () => ({ + slackSearchOutboxHandlers: {}, +})) +vi.mock('@/lib/knowledge/documents/processing-outbox-handler', () => ({ + knowledgeDocumentProcessingOutboxHandlers: {}, +})) +vi.mock('@/lib/organizations/resource-cleanup', () => ({ + organizationResourceCleanupOutboxHandlers: {}, +})) +vi.mock('@/lib/permission-access-requests/notifications', () => ({ + permissionAccessRequestOutboxHandlers: {}, +})) +vi.mock('@/lib/uploads/contexts/workspace/workspace-file-live-doc-outbox', () => ({ + workspaceFileLiveDocOutboxHandlers: {}, +})) +vi.mock('@/lib/uploads/contexts/workspace/workspace-file-storage-cleanup-outbox', () => ({ + workspaceFileStorageCleanupOutboxHandlers: {}, +})) +vi.mock('@/lib/workflows/deployment-outbox', () => ({ workflowDeploymentOutboxHandlers: {} })) +vi.mock('@/lib/workspaces/admin-move', () => ({ invitationMigrationOutboxHandlers: {} })) +vi.mock('@/lib/workspaces/operations/outbox', () => ({ workspaceOperationOutboxHandlers: {} })) +vi.mock('@/ee/workspace-forking/application/content-outbox', () => ({ + forkContentOutboxHandlers: {}, +})) + +import { runOutboxProcessor } from '@/lib/core/outbox/processor' + +describe('outbox processor recovery', () => { + const result = { processed: 5, retried: 1, deadLettered: 0, leaseLost: 0, reaped: 0 } + + beforeEach(() => { + vi.resetAllMocks() + vi.useFakeTimers() + mocks.process.mockResolvedValue(result) + mocks.recover.mockResolvedValue(2) + mocks.reap.mockResolvedValue(3) + }) + afterEach(() => vi.useRealTimers()) + + it('preserves the processing limits and reports independent recovery work', async () => { + await expect(runOutboxProcessor()).resolves.toEqual({ + result, + recoveredDocuments: 2, + reapedBackgroundWork: 3, + }) + expect(mocks.process).toHaveBeenCalledWith(expect.any(Object), { + batchSize: 500, + maxRuntimeMs: 760_000, + minRemainingMs: 95_000, + }) + }) + + it('still reaps expired background work when document recovery fails', async () => { + mocks.recover.mockRejectedValueOnce(new Error('document recovery unavailable')) + await expect(runOutboxProcessor()).resolves.toEqual({ + result, + recoveredDocuments: 0, + reapedBackgroundWork: 3, + }) + }) + + it('retains successful delivery results when the background-work reaper fails', async () => { + mocks.reap.mockRejectedValueOnce(new Error('reaper unavailable')) + await expect(runOutboxProcessor()).resolves.toEqual({ + result, + recoveredDocuments: 2, + reapedBackgroundWork: 0, + }) + }) + + it('skips document recovery after the processing budget is exhausted', async () => { + mocks.process.mockImplementationOnce(async () => { + vi.advanceTimersByTime(770_000) + return result + }) + await expect(runOutboxProcessor()).resolves.toEqual({ + result, + recoveredDocuments: 0, + reapedBackgroundWork: 3, + }) + expect(mocks.recover).not.toHaveBeenCalled() + }) + + it('propagates delivery failures to the worker instead of reporting success', async () => { + mocks.process.mockRejectedValueOnce(new Error('database unavailable')) + await expect(runOutboxProcessor()).rejects.toThrow('database unavailable') + expect(mocks.recover).not.toHaveBeenCalled() + expect(mocks.reap).not.toHaveBeenCalled() + }) +}) diff --git a/apps/sim/lib/core/outbox/processor.ts b/apps/sim/lib/core/outbox/processor.ts new file mode 100644 index 00000000000..7c9fbe9fea7 --- /dev/null +++ b/apps/sim/lib/core/outbox/processor.ts @@ -0,0 +1,101 @@ +import { db } from '@sim/db' +import { createLogger } from '@sim/logger' +import { toError } from '@sim/utils/errors' +import { adminInvitationOperationOutboxHandlers } from '@/lib/admin/invitation-operation' +import { adminMemberOperationOutboxHandlers } from '@/lib/admin/member-operation' +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 { + OUTBOX_PROCESSOR_MAX_RUNTIME_MS, + OUTBOX_PROCESSOR_RECOVERY_CUTOFF_MS, +} from '@/lib/core/outbox/constants' +import { type ProcessOutboxResult, processOutboxEvents } from '@/lib/core/outbox/service' +import { DeadlineExceededError } from '@/lib/core/utils/deadline' +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('OutboxProcessor') + +const handlers = { + ...slackSearchOutboxHandlers, + ...adminInvitationOperationOutboxHandlers, + ...adminMemberOperationOutboxHandlers, + ...billingOutboxHandlers, + ...membershipBillingOutboxHandlers, + ...enterpriseIssuanceOutboxHandlers, + ...enterpriseOwnerClaimOutboxHandlers, + ...invitationMigrationOutboxHandlers, + ...directGrantOutboxHandlers, + ...knowledgeDocumentProcessingOutboxHandlers, + ...organizationResourceCleanupOutboxHandlers, + ...permissionAccessRequestOutboxHandlers, + ...workspaceFileLiveDocOutboxHandlers, + ...workspaceFileStorageCleanupOutboxHandlers, + ...workflowDeploymentOutboxHandlers, + ...workspaceOperationOutboxHandlers, + ...forkContentOutboxHandlers, +} as const + +export interface OutboxProcessorResult { + result: ProcessOutboxResult + recoveredDocuments: number + reapedBackgroundWork: number +} + +/** Processes one bounded batch and its recovery work in either the worker or self-hosted cron. */ +export async function runOutboxProcessor(): Promise { + const startedAt = Date.now() + const result = await processOutboxEvents(handlers, { + batchSize: 500, + maxRuntimeMs: OUTBOX_PROCESSOR_MAX_RUNTIME_MS, + minRemainingMs: 95_000, + }) + + let recoveredDocuments = 0 + try { + if (Date.now() - startedAt < OUTBOX_PROCESSOR_RECOVERY_CUTOFF_MS) { + recoveredDocuments = await recoverKnowledgeDocumentProcessing() + } + } catch (error) { + logger.error('Stored document recovery failed', { + error: getConnectorFailureDiagnostic(error) ?? { + category: error instanceof DeadlineExceededError ? 'deadline' : 'internal', + message: + error instanceof DeadlineExceededError + ? error.message + : 'Unexpected stored-document recovery failure', + }, + }) + } + + /** Reap independently so an expired fork lease cannot prevent outbox delivery. */ + let reapedBackgroundWork = 0 + try { + reapedBackgroundWork = await reapStaleBackgroundWork(db) + } catch (error) { + logger.error('Background-work reap failed', { error: toError(error).message }) + } + + const output = { result, reapedBackgroundWork, recoveredDocuments } + logger.info('Outbox processing completed', { + ...result, + reapedBackgroundWork, + recoveredDocuments, + durationMs: Date.now() - startedAt, + }) + return output +} From 8ed9294b1f91f02999c529df8e6140a2380c66ad Mon Sep 17 00:00:00 2001 From: Vikhyath Mondreti Date: Wed, 16 Sep 2026 19:39:23 -0700 Subject: [PATCH 2/2] fix(outbox): preserve polling capacity with larger workers --- apps/sim/background/process-outbox.test.ts | 3 ++- apps/sim/background/process-outbox.ts | 10 ++++++---- apps/sim/lib/core/outbox/constants.ts | 4 ++++ apps/sim/lib/core/outbox/enqueue.test.ts | 1 - apps/sim/lib/core/outbox/enqueue.ts | 2 -- 5 files changed, 12 insertions(+), 8 deletions(-) diff --git a/apps/sim/background/process-outbox.test.ts b/apps/sim/background/process-outbox.test.ts index 762aa90b8fa..1a1534fdb13 100644 --- a/apps/sim/background/process-outbox.test.ts +++ b/apps/sim/background/process-outbox.test.ts @@ -14,9 +14,10 @@ describe('outbox processor task', () => { 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: 4 }, + queue: { name: 'process-outbox', concurrencyLimit: 15 }, }) const output = { result: { processed: 3 }, recoveredDocuments: 0, reapedBackgroundWork: 0 } mocks.processor.mockResolvedValueOnce(output) diff --git a/apps/sim/background/process-outbox.ts b/apps/sim/background/process-outbox.ts index 3acfb29ba1c..bdcf6679e93 100644 --- a/apps/sim/background/process-outbox.ts +++ b/apps/sim/background/process-outbox.ts @@ -1,18 +1,20 @@ import { task } from '@trigger.dev/sdk' -import { OUTBOX_PROCESSOR_MAX_DURATION_SECONDS } from '@/lib/core/outbox/constants' +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: 'small-2x', + 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 }, - /** Keep other deliveries moving while a worker handles a long operation. */ queue: { name: 'process-outbox', - concurrencyLimit: 4, + concurrencyLimit: OUTBOX_PROCESSOR_CONCURRENCY, }, run: () => runOutboxProcessor(), }) diff --git a/apps/sim/lib/core/outbox/constants.ts b/apps/sim/lib/core/outbox/constants.ts index 6a61b30f3ad..c19580a9832 100644 --- a/apps/sim/lib/core/outbox/constants.ts +++ b/apps/sim/lib/core/outbox/constants.ts @@ -2,3 +2,7 @@ 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 +) diff --git a/apps/sim/lib/core/outbox/enqueue.test.ts b/apps/sim/lib/core/outbox/enqueue.test.ts index d569e7ad598..b6d4d992185 100644 --- a/apps/sim/lib/core/outbox/enqueue.test.ts +++ b/apps/sim/lib/core/outbox/enqueue.test.ts @@ -39,7 +39,6 @@ describe('outbox processor enqueue', () => { idempotencyKeyTTL: '5m', maxDuration: 900, region: 'us-east-1', - ttl: '1m', }) expect(mocks.processor).not.toHaveBeenCalled() }) diff --git a/apps/sim/lib/core/outbox/enqueue.ts b/apps/sim/lib/core/outbox/enqueue.ts index e597c863dcb..95f2bf9e03d 100644 --- a/apps/sim/lib/core/outbox/enqueue.ts +++ b/apps/sim/lib/core/outbox/enqueue.ts @@ -27,8 +27,6 @@ export async function enqueueOutboxProcessor(): Promise