diff --git a/apps/sim/app/api/workflows/[id]/execute/route.async.test.ts b/apps/sim/app/api/workflows/[id]/execute/route.async.test.ts index ce55f06ee23..90e03530dab 100644 --- a/apps/sim/app/api/workflows/[id]/execute/route.async.test.ts +++ b/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((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> => 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> => 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), { diff --git a/apps/sim/app/api/workflows/[id]/execute/route.ts b/apps/sim/app/api/workflows/[id]/execute/route.ts index 49f50b3ae21..d3604092f2b 100644 --- a/apps/sim/app/api/workflows/[id]/execute/route.ts +++ b/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 { +async function handleAsyncExecution( + params: AsyncExecutionParams, + admission?: AdmissionLease +): Promise { const { requestId, workflowId, @@ -442,7 +455,17 @@ async function handleAsyncExecution(params: AsyncExecutionParams): Promise { let workerOwnsReservation = false try { @@ -471,6 +494,8 @@ async function handleAsyncExecution(params: AsyncExecutionParams): Promise }) => { 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 { 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 } diff --git a/apps/sim/lib/core/admission/gate.ts b/apps/sim/lib/core/admission/gate.ts index f3c4866a246..95c886c2833 100644 --- a/apps/sim/lib/core/admission/gate.ts +++ b/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) { diff --git a/apps/sim/lib/core/config/env.ts b/apps/sim/lib/core/config/env.ts index c86eb9e7ff4..cf202b68aca 100644 --- a/apps/sim/lib/core/config/env.ts +++ b/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) diff --git a/apps/sim/lib/core/telemetry.ts b/apps/sim/lib/core/telemetry.ts index 125592d0ad7..3f825f7c273 100644 --- a/apps/sim/lib/core/telemetry.ts +++ b/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, diff --git a/apps/sim/lib/workflows/executor/execution-metrics.ts b/apps/sim/lib/workflows/executor/execution-metrics.ts new file mode 100644 index 00000000000..1f2f328150d --- /dev/null +++ b/apps/sim/lib/workflows/executor/execution-metrics.ts @@ -0,0 +1,68 @@ +/** + * Workflow execution OTel metrics — throughput counter + latency histogram. + * + * Wired into `createOTelSpansForWorkflowExecution` (the single funnel that all + * four `LoggingSession.complete` paths call), so the counter fires exactly once + * per completed execution. + * + * Bounded cardinality: only `workflow_id`, `trigger`, and `status` are label + * dimensions. Never add `execution_id`, `workspace_id`, or `user_id` — those + * explode Prometheus series. + */ +import { type Counter, type Histogram, metrics } from '@opentelemetry/api' + +/** Matches `copilot/request/metrics.ts` latency buckets for histogram_quantile validity. */ +const LATENCY_BUCKETS_MS = [ + 50, 100, 250, 500, 1000, 2000, 5000, 10000, 20000, 30000, 60000, 120000, 300000, +] + +interface WorkflowMeterInstruments { + executionCount: Counter + executionDuration: Histogram +} + +let cached: WorkflowMeterInstruments | undefined + +/** + * Lazy init: Turbopack/Next can evaluate this module before the NodeSDK + * installs the real MeterProvider, so resolve instruments on first use (a + * no-op meter before then simply drops records — same pattern as + * `copilot/request/metrics.ts`). + */ +function instruments(): WorkflowMeterInstruments { + if (cached) return cached + const meter = metrics.getMeter('sim-workflow-executor') + cached = { + executionCount: meter.createCounter('workflow.execution.count'), + executionDuration: meter.createHistogram('workflow.execution.duration', { + unit: 'ms', + advice: { explicitBucketBoundaries: LATENCY_BUCKETS_MS }, + }), + } + return cached +} + +/** + * Record one completed workflow execution as a counter increment and latency + * histogram sample. Called once per execution from the telemetry funnel. + * + * Never throws — inherits the caller's try/catch guard. + */ +export function recordWorkflowExecution(params: { + workflowId: string + trigger: string + status: 'success' | 'error' + durationMs: number +}): void { + const { workflowId, trigger, status, durationMs } = params + const { executionCount, executionDuration } = instruments() + + const attrs = { + workflow_id: workflowId || 'unknown', + trigger: trigger || 'unknown', + status, + } + + executionCount.add(1, attrs) + executionDuration.record(durationMs, attrs) +} diff --git a/apps/sim/package.json b/apps/sim/package.json index a7800a9f904..aea4e106527 100644 --- a/apps/sim/package.json +++ b/apps/sim/package.json @@ -19,7 +19,7 @@ "load:workflow:isolation": "BASE_URL=${BASE_URL:-http://localhost:3000} ISOLATION_DURATION=${ISOLATION_DURATION:-30} TOTAL_RATE=${TOTAL_RATE:-9} WORKSPACE_A_WEIGHT=${WORKSPACE_A_WEIGHT:-8} WORKSPACE_B_WEIGHT=${WORKSPACE_B_WEIGHT:-1} bunx artillery run scripts/load/workflow-isolation.yml", "build": "bun run build:sandbox-bundles && NODE_OPTIONS='--max-old-space-size=8192' next build", "build:sandbox-bundles": "bun run ./lib/execution/sandbox/bundles/build.ts", - "start": "next start", + "start": "NODE_OPTIONS='--max-old-space-size=4096' next start", "prepare": "cd ../.. && bun husky", "test": "vitest run", "test:watch": "vitest", diff --git a/apps/sim/scripts/load/README.md b/apps/sim/scripts/load/README.md index 43a08976431..3dc871a2952 100644 --- a/apps/sim/scripts/load/README.md +++ b/apps/sim/scripts/load/README.md @@ -10,7 +10,7 @@ These local-only Artillery scenarios exercise `POST /api/workflows/[id]/execute` The default rates are tuned for these local limits: -- `ADMISSION_GATE_MAX_INFLIGHT=500` +- `ADMISSION_GATE_MAX_INFLIGHT=10` The admission gate caps total in-flight requests per pod. @@ -23,7 +23,7 @@ Default profile: - Starts at `2` requests per second - Ramps to `8` requests per second - Holds there for `20` seconds -- Good for validating queueing against a Free workspace concurrency of `5` +- Good for validating queueing against a Free workspace concurrency of `10` ```bash WORKFLOW_ID= \ @@ -102,6 +102,6 @@ Optional variables: - `load:workflow` is an alias for `load:workflow:baseline` - All scenarios send `x-execution-mode: async` -- Artillery output will show request counts and response codes, which is usually enough for quick local verification -- At these defaults, you should see `429` responses when approaching `ADMISSION_GATE_MAX_INFLIGHT=500` +- The baseline asserts `202` responses, but completion throughput still requires querying `workflow_execution_logs.ended_at`; request counts alone do not measure workflow throughput +- At these defaults, you should see `429` responses when approaching `ADMISSION_GATE_MAX_INFLIGHT=10` - If you still see lots of `429` or `ETIMEDOUT` responses locally, lower the rates again before increasing durations diff --git a/apps/sim/scripts/load/workflow-concurrency.yml b/apps/sim/scripts/load/workflow-concurrency.yml index a981d438e91..ecb58815a7d 100644 --- a/apps/sim/scripts/load/workflow-concurrency.yml +++ b/apps/sim/scripts/load/workflow-concurrency.yml @@ -22,3 +22,5 @@ scenarios: input: source: artillery-baseline runId: "{{ $uuid }}" + expect: + - statusCode: 202 diff --git a/docs/workflow-throughput-one-pager.md b/docs/workflow-throughput-one-pager.md new file mode 100644 index 00000000000..1f87207ed1e --- /dev/null +++ b/docs/workflow-throughput-one-pager.md @@ -0,0 +1,192 @@ +# Sim workflow execution throughput and bottlenecks + +## Executive summary + +The incident was not that the execute route could not return `202` quickly. The +database-backed async path released its admission permit when the HTTP response +returned, while the workflow continued in the web process. Requests accumulated +until the process became slow and unreachable. + +The fix in PR 5791 keeps the permit until local inline execution finishes. This +improves overload behavior and makes backpressure visible as `429`; it does not +claim that workflow execution itself became faster. + +Throughput means terminal workflow executions per second. `202` responses per +second measure request admission, not completed work. + +## Scope and measurement contract + +- **Local evidence:** one web process, database async backend, synthetic workflow. +- **Production evidence:** Trigger.dev async backend, worker process, Trigger.dev + queue limits, trigger DB pool, and shared PostgreSQL/Redis. +- **Success throughput:** completed execution-log rows per second: + `status = 'completed'`, `level = 'info'`, grouped by `ended_at`. +- **Terminal throughput:** completed, failed, and cancelled rows per second, + grouped by `ended_at`. Exclude `pending`; it represents a paused run. +- **Admission signals:** `202`, `429`, `503`, response latency, and enqueue + acceptance. These explain backpressure but are not throughput. +- **Saturation signals:** running execution count, worker/queue depth, DB pool + pressure, Redis errors, CPU, RSS, and heap. + +The branch adds `workflow.execution.count` and +`workflow.execution.duration` in +[execution-metrics.ts](../apps/sim/lib/workflows/executor/execution-metrics.ts). +Until production export and every terminal path are validated, use the database +terminal rows as the immediate ground truth. The current OTel recorder is +secondary because some completion paths only emit telemetry when trace spans are +present. Trigger.dev queue depth is also external to Sim; `async_jobs` is only a +database-backend proxy. + +### Before: the permit represented the HTTP response + +```mermaid +sequenceDiagram + participant C as Caller + participant R as ExecuteRoute + participant G as AdmissionGate + participant Q as DatabaseBackend + participant E as InlineExecution + + C->>R: POST /execute + R->>G: tryAdmit() + G-->>R: permit + R->>Q: enqueue workflow + Q-->>R: job accepted + R-->>C: 202 Accepted + R->>G: release permit + Q->>E: continue execution in web process + Note over E: More requests enter while old runs remain active + E-->>Q: complete +``` + +### After: the permit represents in-process work + +```mermaid +sequenceDiagram + participant C as Caller + participant R as ExecuteRoute + participant G as AdmissionGate + participant Q as DatabaseBackend + participant E as InlineExecution + + C->>R: POST /execute + R->>G: tryAdmit() + alt capacity available + G-->>R: permit + R->>Q: enqueue workflow + Q-->>R: job accepted + R-->>C: 202 Accepted + Q->>E: execute inline in web process + E-->>Q: complete + Q->>G: release permit + else capacity full + G-->>R: no permit + R-->>C: 429 Retry-After + end +``` + +### Local and production execution paths + +```mermaid +flowchart LR + Client --> Route[Execute route] + Route --> Gate[Admission gate] + Route --> Preprocess[Auth billing rate limits] + Preprocess --> Enqueue[Async enqueue] + Enqueue --> Local[Database backend] + Enqueue --> Trigger[Trigger.dev backend] + Local --> WebExecutor[Web process executor] + Trigger --> Worker[Trigger worker executor] + WebExecutor --> WebPool[sim-app pool max 10] + Worker --> TriggerPool[sim-trigger pool max 5] + WebExecutor --> Shared[PostgreSQL Redis storage tools] + Worker --> Shared +``` + +## Bottleneck register + +Each Evidence entry is a falsifier, not merely a plausible explanation. A +chokepoint is proven only when the stated resource signal saturates while +terminal throughput plateaus or degrades, and the alternative signals do not +explain the result. + +| Bottleneck | Scope / status | Evidence (replayable falsifier) | Fix | +|---|---|---|---| +| Admission permit released before database-backed async work finished | Local database backend / **fixed in PR 5791** | Run `workflow-concurrency.yml` at `8` RPS with the pre-fix behavior: high `202`, rising `ECONNREFUSED`, and low terminal completions. Repeat after the fix: `429` appears at the configured cap and the health endpoint remains reachable. Count terminal rows with the SQL below. | Retain the ticket until `executeWorkflowJob` finishes; release it in the inline runner’s `finally`. | +| Admission setting does not match deployed process capacity | Web pods / **code default 10, Helm and docker-compose override 500** | Set `ADMISSION_GATE_MAX_INFLIGHT=5`, run the baseline above the cap, and verify `429` responses plus `Admission gate rejecting request` logs. Compare the same run with the deployed value and record whether running work exceeds the intended per-pod budget. | Align deployment overrides with measured per-pod capacity, or document why a higher value is safe. | +| Workflow work duration limits completion rate | Local and production | Use identical workflows with deterministic work of approximately `300 ms`, `1 s`, and `2 s`. Hold arrival rate constant; terminal throughput should fall as duration rises while the process remains healthy. If it does not, CPU duration is not the binding limit. | No workflow fix by itself; tune admission/worker concurrency or reduce work cost after measuring. | +| Web DB pool is shared with inline execution | Local database backend / `sim-app` primary pool max 10 | During a load hold, poll `pg_stat_activity` by `application_name`. A DB choke requires `sim-app` active connections near 10, rising query age or lock wait, and flat terminal throughput. Pool size alone is not proof. | Keep inline admission bounded; reduce queries per execution; optimize or move eligible reads only after runtime evidence. | +| Trigger worker concurrency exceeds the trigger DB pool | Production/staging Trigger.dev only / not locally reproduced | Run the same profile with Trigger.dev enabled. Prove the choke only when Trigger pending/running work grows, `sim-trigger` reaches its pool ceiling of 5, DB/query latency rises, and terminal throughput stays flat. | Size worker concurrency to the scarce downstream resource or add a shared DB-borrow semaphore. | +| Independent Trigger.dev queues contend for one worker resource | Production/staging only / not locally reproduced | Saturate workflow, webhook, and resume traffic separately and together. Compare queue depth, worker running counts, `sim-trigger` connections, and terminal completions. A shared-resource choke requires combined traffic to reduce throughput more than isolated traffic. | Coordinate queue limits against the shared pool; expose queue depth from Trigger.dev rather than treating `async_jobs` as production depth. | +| One tenant can delay another tenant | Production/staging only / fairness gap not reproduced | Send long-running traffic from workspace A and short runs from workspace B at constant B arrival rate. Prove starvation only if B’s completion latency or queue wait worsens as A increases while aggregate throughput looks healthy. | Add per-tenant queue/concurrency keys or weighted fairness after reproducing the effect. | +| Sync and SSE executions pressure the web process outside async admission | Local and production | Compare equal-work async, sync, and SSE runs. Record health failures, request latency, RSS/heap, and terminal throughput. A mode-specific choke requires sync/SSE to degrade the web process while async remains healthy at similar completed work. | Apply separate web-process admission or route long work through the worker path. | +| Executor fan-out and retained loop/payload state consume heap | Local and production workers | Sweep parallel branch count, loop iterations, and output size independently. Record peak RSS/heap and terminal status. Use a diagnostic heap cap to amplify the failure, but require the uncapped trend to support the claim. | Bound fan-out and retained iteration/output state; use durable references or per-execution memory budgets. | +| Terminal logging, storage, or Redis fallback dominates completion | Local and production | Compare small outputs with outputs above the large-value threshold, and Redis healthy with Redis degraded. Prove the path only when upload/JSONB/error latency rises with flat terminal throughput and matching logs. | Batch or reduce terminal writes, preserve Redis health, and bound storage/materialization work. | +| External tools, provider rate limits, or retries dominate workflow time | Local synthetic dependency and production | Use a deterministic external stub with fixed latency, 429s, and 5xx responses. Correlate block duration/retry logs and provider errors with terminal throughput. A dependency choke requires the dependency signal to move with the throughput plateau. | Tune bounded retries/timeouts, provider concurrency, and backpressure; do not “fix” ingress based on `202` latency alone. | + +## Shared measurement queries + +Use event time (`ended_at`), not query time, when calculating completion rates. + +```sql +-- Success throughput +SELECT date_trunc('minute', ended_at) AS minute, count(*) AS completed +FROM workflow_execution_logs +WHERE ended_at >= now() - interval '15 minutes' + AND status = 'completed' + AND level = 'info' +GROUP BY 1 +ORDER BY 1; + +-- All terminal outcomes +SELECT date_trunc('minute', ended_at) AS minute, status, count(*) AS executions +FROM workflow_execution_logs +WHERE ended_at >= now() - interval '15 minutes' + AND status IN ('completed', 'failed', 'cancelled') +GROUP BY 1, 2 +ORDER BY 1, 2; + +-- Cross-pod running executions +SELECT count(*) AS running +FROM workflow_execution_logs +WHERE status = 'running'; + +-- Database pressure +SELECT application_name, state, count(*) AS connections, + max(now() - query_start) AS oldest_query +FROM pg_stat_activity +WHERE application_name IN ('sim-app', 'sim-trigger') +GROUP BY application_name, state +ORDER BY application_name, state; +``` + +## Experiments to run next + +Run these in local or staging with a fixed workflow fixture, tenant mix, replica +count, and drain period. Record per-pod/per-worker results so autoscaling is not +mistaken for a code-level improvement. + +1. **Arrival and duration sweep:** `300 ms` to `2 s` deterministic work, with + arrival rates below and above the admission cap. +2. **Backend comparison:** identical load through the database backend and + Trigger.dev; compare web versus trigger pools and external queue depth. +3. **Fairness isolation:** long-running workspace A versus short-running + workspace B at constant B traffic. +4. **Mode pressure:** async versus sync versus SSE, including health and RSS. +5. **Memory/fan-out:** loop retention, parallel branch count, and large output + thresholds as independent variables. +6. **Dependency failure:** Redis degradation, storage uploads, and deterministic + external-tool latency/rate-limit responses. + +## Existing load profile and interpretation + +The baseline profile is +[workflow-concurrency.yml](../apps/sim/scripts/load/workflow-concurrency.yml). +`bun run load:workflow:baseline` defaults to a 10-second warmup, ramps from +`2` to `8` requests/sec, and holds for 20 seconds. The profile asserts `202`; +completion counts still require the terminal-log queries above. + +The local PR evidence used a synthetic `Start → Function` workflow and is +stability evidence, not a production capacity number. Production claims require +the Trigger.dev path, worker metrics, DB/Redis telemetry, and a fixed comparison +against an identical baseline.