Fix workflow throughput admission

apps/sim/app/api/workflows/[id]/execute/route.async.test.ts

@@ -34,6 +34,11 @@ const {
   mockReleaseExecutionIdClaim,
   mockReleaseExecutionSlot,
   mockRequireBillingAttributionHeader,
+  mockShouldExecuteInline,
+  mockStartJob,
+  mockCompleteJob,
+  mockMarkJobFailed,
+  mockExecuteWorkflowJob,
   mockValidatePublicApiAllowed,
 } = vi.hoisted(() => ({
   mockAssertBillingAttributionSnapshot: vi.fn((value: unknown) => {
@@ -52,6 +57,11 @@ const {
   mockReleaseExecutionIdClaim: vi.fn(),
   mockReleaseExecutionSlot: vi.fn(),
   mockRequireBillingAttributionHeader: vi.fn(),
+  mockShouldExecuteInline: vi.fn(() => false),
+  mockStartJob: vi.fn(),
+  mockCompleteJob: vi.fn(),
+  mockMarkJobFailed: vi.fn(),
+  mockExecuteWorkflowJob: vi.fn(),
   mockValidatePublicApiAllowed: vi.fn(),
 }))
 
@@ -114,11 +124,11 @@ vi.mock('@/lib/execution/payloads/store', () => ({
 vi.mock('@/lib/core/async-jobs', () => ({
   getJobQueue: vi.fn().mockResolvedValue({
     enqueue: mockEnqueue,
-    startJob: vi.fn(),
-    completeJob: vi.fn(),
-    markJobFailed: vi.fn(),
+    startJob: mockStartJob,
+    completeJob: mockCompleteJob,
+    markJobFailed: mockMarkJobFailed,
   }),
-  shouldExecuteInline: vi.fn().mockReturnValue(false),
+  shouldExecuteInline: mockShouldExecuteInline,
 }))
 
 vi.mock('@/lib/core/utils/urls', () => ({
@@ -136,7 +146,7 @@ vi.mock('@/lib/execution/call-chain', () => ({
 vi.mock('@/lib/logs/execution/logging-session', () => loggingSessionMock)
 
 vi.mock('@/background/workflow-execution', () => ({
-  executeWorkflowJob: vi.fn(),
+  executeWorkflowJob: mockExecuteWorkflowJob,
 }))
 
 vi.mock('@sim/utils/id', () => ({
@@ -147,6 +157,7 @@ vi.mock('@sim/utils/id', () => ({
   ),
 }))
 
+import { getAdmissionGateStatus, tryAdmit } from '@/lib/core/admission/gate'
 import { storeLargeValue } from '@/lib/execution/payloads/store'
 import { POST } from './route'
 
@@ -163,7 +174,10 @@ const billingAttribution = {
   payerSubscription: null,
 }
 
-function createSessionReplayRequest(executionId: string): NextRequest {
+function createSessionReplayRequest(
+  executionId: string,
+  executionMode: 'async' | 'sync' = 'async'
+): NextRequest {
   return createMockRequest(
     'POST',
     {
@@ -173,7 +187,8 @@ function createSessionReplayRequest(executionId: string): NextRequest {
     },
     {
       'Content-Type': 'application/json',
-      'X-Execution-Mode': 'async',
+      Cookie: 'session=value',
+      ...(executionMode === 'async' ? { 'X-Execution-Mode': 'async' } : {}),
     }
   )
 }
@@ -283,6 +298,8 @@ describe('workflow execute async route', () => {
   beforeEach(() => {
     vi.clearAllMocks()
     resetDbChainMock()
+    mockShouldExecuteInline.mockReturnValue(false)
+    mockExecuteWorkflowJob.mockResolvedValue({ success: true })
     mockGenerateId.mockReset().mockReturnValue('execution-123')
     mockClaimExecutionId.mockImplementation(async (executionId: string) => ({
       key: `workflow-execution-id:${executionId}`,
@@ -387,6 +404,81 @@ describe('workflow execute async route', () => {
     )
   })
 
+  it('retains the admission ticket until database-backed async execution finishes', async () => {
+    mockShouldExecuteInline.mockReturnValue(true)
+    let resolveInlineExecution!: () => void
+    const inlineExecution = new Promise<void>((resolve) => {
+      resolveInlineExecution = resolve
+    })
+    mockExecuteWorkflowJob.mockReturnValueOnce(inlineExecution)
+
+    const response = await POST(
+      createMockRequest(
+        'POST',
+        { input: { hello: 'world' } },
+        {
+          'Content-Type': 'application/json',
+          'X-Execution-Mode': 'async',
+        }
+      ),
+      { params: Promise.resolve({ id: 'workflow-1' }) }
+    )
+
+    expect(response.status).toBe(202)
+    expect(getAdmissionGateStatus().inflight).toBe(1)
+
+    resolveInlineExecution()
+    await vi.waitFor(() => {
+      expect(getAdmissionGateStatus().inflight).toBe(0)
+    })
+    expect(mockCompleteJob).toHaveBeenCalledWith('job-123', undefined)
+  })
+
+  it('applies admission backpressure to session-backed async executions', async () => {
+    hybridAuthMockFns.mockHasExternalApiCredentials.mockReturnValue(false)
+    const heldTickets = Array.from({ length: getAdmissionGateStatus().maxInflight }, () =>
+      tryAdmit()
+    ).filter((ticket): ticket is NonNullable<ReturnType<typeof tryAdmit>> => ticket !== null)
+
+    try {
+      const response = await POST(
+        createSessionReplayRequest('66666666-6666-4666-8666-666666666666'),
+        {
+          params: Promise.resolve({ id: 'workflow-1' }),
+        }
+      )
+
+      expect(response.status).toBe(429)
+      expect(mockCheckHybridAuth).not.toHaveBeenCalled()
+      expect(mockPreprocessExecution).not.toHaveBeenCalled()
+    } finally {
+      for (const ticket of heldTickets) {
+        ticket.release()
+      }
+    }
+  })
+
+  it('leaves session-backed synchronous executions on their existing path', async () => {
+    hybridAuthMockFns.mockHasExternalApiCredentials.mockReturnValue(false)
+    const heldTickets = Array.from({ length: getAdmissionGateStatus().maxInflight }, () =>
+      tryAdmit()
+    ).filter((ticket): ticket is NonNullable<ReturnType<typeof tryAdmit>> => ticket !== null)
+
+    try {
+      const response = await POST(
+        createSessionReplayRequest('77777777-7777-4777-8777-777777777777', 'sync'),
+        { params: Promise.resolve({ id: 'workflow-1' }) }
+      )
+
+      expect(response.status).not.toBe(429)
+      expect(mockCheckHybridAuth).toHaveBeenCalled()
+    } finally {
+      for (const ticket of heldTickets) {
+        ticket.release()
+      }
+    }
+  })
+
   it('preserves a first-use execution ID supplied by an authenticated session', async () => {
     const requestedExecutionId = '11111111-1111-4111-8111-111111111111'
     const response = await POST(createSessionReplayRequest(requestedExecutionId), {

apps/sim/app/api/workflows/[id]/execute/route.ts

@@ -19,7 +19,11 @@ import {
   type BillingAttributionSnapshot,
   requireBillingAttributionHeader,
 } from '@/lib/billing/core/billing-attribution'
-import { admissionRejectedResponse, tryAdmit } from '@/lib/core/admission/gate'
+import {
+  type AdmissionTicket,
+  admissionRejectedResponse,
+  tryAdmit,
+} from '@/lib/core/admission/gate'
 import { getJobQueue, shouldExecuteInline } from '@/lib/core/async-jobs'
 import { isAsyncJobEnqueueError } from '@/lib/core/async-jobs/types'
 import {
@@ -283,6 +287,12 @@ interface AsyncExecutionResult {
   retainExecutionClaim: boolean
 }
 
+/** Mutable bag so async inline work can retain the HTTP admission ticket. */
+interface AdmissionLease {
+  ticket: AdmissionTicket
+  retained: boolean
+}
+
 type ValidatedPreprocessContext = {
   actorUserId: string
   workflow: PreprocessExecutionSuccess['workflowRecord']
@@ -319,7 +329,10 @@ function requirePreprocessedExecutionContext(
   }
 }
 
-async function handleAsyncExecution(params: AsyncExecutionParams): Promise<AsyncExecutionResult> {
+async function handleAsyncExecution(
+  params: AsyncExecutionParams,
+  admission?: AdmissionLease
+): Promise<AsyncExecutionResult> {
   const {
     requestId,
     workflowId,
@@ -442,7 +455,17 @@ async function handleAsyncExecution(params: AsyncExecutionParams): Promise<Async
 
   asyncLogger.info('Queued async workflow execution', { jobId })
 
-  if (shouldExecuteInline()) {
+  /**
+   * DB-backend path runs executions in this process. Hold the admission ticket
+   * until the run finishes so ADMISSION_GATE_MAX_INFLIGHT bounds concurrent
+   * executions (not just concurrent HTTP handlers). Trigger.dev path releases
+   * on HTTP return — the remote worker enforces its own concurrency.
+   */
+  const executeInline = shouldExecuteInline()
+  if (executeInline) {
+    if (admission) {
+      admission.retained = true
+    }
     void (async () => {
       let workerOwnsReservation = false
       try {
@@ -471,6 +494,8 @@ async function handleAsyncExecution(params: AsyncExecutionParams): Promise<Async
             error: toError(markFailedError).message,
           })
         }
+      } finally {
+        admission?.ticket.release()
       }
     })()
   }
@@ -500,7 +525,15 @@ async function handleAsyncExecution(params: AsyncExecutionParams): Promise<Async
 export const POST = withRouteHandler(
   async (req: NextRequest, { params }: { params: Promise<{ id: string }> }) => {
     const isSessionRequest = req.headers.has('cookie') && !hasExternalApiCredentials(req.headers)
-    if (isSessionRequest) {
+    const isAsyncRequest = req.headers.get('x-execution-mode') === 'async'
+
+    /**
+     * Session-backed sync and SSE runs retain their existing lifecycle because
+     * their work completes through the response/stream. Session-backed async
+     * runs use the same inline worker path as API-key requests and must consume
+     * an admission ticket until that work completes.
+     */
+    if (isSessionRequest && !isAsyncRequest) {
       return handleExecutePost(req, params)
     }
 
@@ -509,17 +542,21 @@ export const POST = withRouteHandler(
       return admissionRejectedResponse()
     }
 
+    const admission: AdmissionLease = { ticket, retained: false }
     try {
-      return await handleExecutePost(req, params)
+      return await handleExecutePost(req, params, admission)
     } finally {
-      ticket.release()
+      if (!admission.retained) {
+        ticket.release()
+      }
     }
   }
 )
 
 async function handleExecutePost(
   req: NextRequest,
-  params: Promise<{ id: string }>
+  params: Promise<{ id: string }>,
+  admission?: AdmissionLease
 ): Promise<NextResponse | Response> {
   const requestId = generateRequestId()
   const { id: workflowId } = await params
@@ -1051,17 +1088,20 @@ async function handleExecutePost(
     reqLogger.info('Preprocessing passed')
 
     if (isAsyncMode) {
-      const asyncResult = await handleAsyncExecution({
-        requestId,
-        workflowId,
-        userId: actorUserId,
-        billingAttribution,
-        workspaceId,
-        input,
-        triggerType: loggingTriggerType,
-        executionId,
-        callChain,
-      })
+      const asyncResult = await handleAsyncExecution(
+        {
+          requestId,
+          workflowId,
+          userId: actorUserId,
+          billingAttribution,
+          workspaceId,
+          input,
+          triggerType: loggingTriggerType,
+          executionId,
+          callChain,
+        },
+        admission
+      )
       executionIdClaimCommitted = asyncResult.retainExecutionClaim
       return asyncResult.response
     }

apps/sim/lib/core/admission/gate.ts

@@ -5,7 +5,11 @@ import { env } from '@/lib/core/config/env'
 
 const logger = createLogger('AdmissionGate')
 
-const MAX_INFLIGHT = Number.parseInt(env.ADMISSION_GATE_MAX_INFLIGHT ?? '') || 500
+/**
+ * Default matches the web DB pool (`primaryMax=10`) so admitted in-process work
+ * cannot silently outrun shared connections + heap. Override per pod via env.
+ */
+const MAX_INFLIGHT = Number.parseInt(env.ADMISSION_GATE_MAX_INFLIGHT ?? '') || 10
 
 let inflight = 0
 
@@ -19,6 +23,10 @@ export interface AdmissionTicket {
  * Zero external calls — purely in-process atomic counter. Each pod maintains its
  * own counter, so the effective aggregate limit across N pods is N × MAX_INFLIGHT.
  * Configure ADMISSION_GATE_MAX_INFLIGHT per pod based on what each pod can sustain.
+ *
+ * Callers that run work inline after returning HTTP 202 must retain the ticket
+ * until that work finishes — releasing on response alone does not bound
+ * concurrent executions.
  */
 export function tryAdmit(): AdmissionTicket | null {
   if (inflight >= MAX_INFLIGHT) {

apps/sim/lib/core/config/env.ts

@@ -294,7 +294,7 @@ export const env = createEnv({
 
 
     // Admission & Burst Protection
-    ADMISSION_GATE_MAX_INFLIGHT:           z.string().optional().default('500'),   // Max concurrent in-flight execution requests per pod
+    ADMISSION_GATE_MAX_INFLIGHT:           z.string().optional().default('10'),    // Max concurrent in-flight executions per pod (aligned with web DB pool primaryMax=10)
     API_MAX_JSON_BODY_BYTES:               z.string().optional().default('52428800'),// Default max JSON request body size for contract routes (50 MB)
     CHAT_MAX_REQUEST_BYTES:                z.string().optional().default('230686720'),// Max request body size for the public deployed-chat endpoint (220 MB; covers 15 base64 file attachments)
     WEBHOOK_MAX_REQUEST_BYTES:             z.string().optional().default('10485760'),// Max request body size for public webhook receiver endpoints (10 MB; provider payloads rarely exceed a few MB)

apps/sim/lib/core/telemetry.ts

@@ -22,6 +22,7 @@ import { getErrorMessage, toError } from '@sim/utils/errors'
 import { TraceAttr } from '@/lib/copilot/generated/trace-attributes-v1'
 import type { TraceSpan } from '@/lib/logs/types'
 import { hostedKeyMetrics } from '@/lib/monitoring/metrics'
+import { recordWorkflowExecution } from '@/lib/workflows/executor/execution-metrics'
 
 /**
  * GenAI Semantic Convention Attributes
@@ -372,6 +373,12 @@ export function createOTelSpansForWorkflowExecution(params: {
     }
 
     rootSpan.end(new Date(params.endTime))
+    recordWorkflowExecution({
+      workflowId: params.workflowId,
+      trigger: params.trigger,
+      status: params.status,
+      durationMs: params.totalDurationMs,
+    })
 
     logger.debug('Created OTel spans for workflow execution', {
       workflowId: params.workflowId,