From 4ac8c3ac1a55edf0c43c323fddb4488adfa76e3c Mon Sep 17 00:00:00 2001 From: akshay326 Date: Tue, 21 Jul 2026 01:00:36 +0000 Subject: [PATCH 1/5] fix(execution): bound async workflow admission Co-authored-by: Cursor --- .../[id]/execute/route.async.test.ts | 54 ++++++- .../app/api/workflows/[id]/execute/route.ts | 85 +++++++--- apps/sim/blocks/utils.test.ts | 26 ++- apps/sim/blocks/utils.ts | 8 + apps/sim/lib/api-key/byok.ts | 14 ++ apps/sim/lib/core/admission/gate.ts | 10 +- apps/sim/lib/core/config/env-flags.ts | 10 ++ apps/sim/lib/core/config/env.ts | 5 +- apps/sim/lib/core/telemetry.ts | 7 + .../workflows/executor/execution-metrics.ts | 68 ++++++++ apps/sim/package.json | 2 +- .../sim/scripts/load/workflow-concurrency.yml | 2 + bun.lock | 13 +- docs/workflow-throughput-one-pager.md | 148 ++++++++++++++++++ packages/testing/src/mocks/env-flags.mock.ts | 1 + 15 files changed, 418 insertions(+), 35 deletions(-) create mode 100644 apps/sim/lib/workflows/executor/execution-metrics.ts create mode 100644 docs/workflow-throughput-one-pager.md 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..803d114d018 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 @@ -147,6 +147,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 +164,10 @@ const billingAttribution = { payerSubscription: null, } -function createSessionReplayRequest(executionId: string): NextRequest { +function createSessionReplayRequest( + executionId: string, + executionMode: 'async' | 'sync' = 'async' +): NextRequest { return createMockRequest( 'POST', { @@ -173,7 +177,8 @@ function createSessionReplayRequest(executionId: string): NextRequest { }, { 'Content-Type': 'application/json', - 'X-Execution-Mode': 'async', + Cookie: 'session=value', + ...(executionMode === 'async' ? { 'X-Execution-Mode': 'async' } : {}), } ) } @@ -387,6 +392,51 @@ describe('workflow execute async route', () => { ) }) + 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..3b8427e076e 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 { @@ -281,6 +285,17 @@ type AsyncExecutionParams = { interface AsyncExecutionResult { response: NextResponse retainExecutionClaim: boolean + /** + * When true, the caller must not release the admission ticket — ownership + * transferred to the in-process inline runner which releases on completion. + */ + retainAdmissionTicket: boolean +} + +/** Mutable bag so async inline work can retain the HTTP admission ticket. */ +interface AdmissionLease { + ticket: AdmissionTicket + retained: boolean } type ValidatedPreprocessContext = { @@ -319,7 +334,10 @@ function requirePreprocessedExecutionContext( } } -async function handleAsyncExecution(params: AsyncExecutionParams): Promise { +async function handleAsyncExecution( + params: AsyncExecutionParams, + admission?: AdmissionLease +): Promise { const { requestId, workflowId, @@ -373,6 +391,7 @@ async function handleAsyncExecution(params: AsyncExecutionParams): Promise { let workerOwnsReservation = false try { @@ -471,6 +502,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 +551,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 +1097,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/blocks/utils.test.ts b/apps/sim/blocks/utils.test.ts index 3dc571b9dd7..911188bb62e 100644 --- a/apps/sim/blocks/utils.test.ts +++ b/apps/sim/blocks/utils.test.ts @@ -3,11 +3,13 @@ */ import { beforeEach, describe, expect, it, vi } from 'vitest' -const { mockIsHosted, mockIsAzureConfigured, mockIsOllamaConfigured } = vi.hoisted(() => ({ - mockIsHosted: { value: false }, - mockIsAzureConfigured: { value: false }, - mockIsOllamaConfigured: { value: false }, -})) +const { mockIsHosted, mockIsAzureConfigured, mockIsOllamaConfigured, mockIsOpenRouterConfigured } = + vi.hoisted(() => ({ + mockIsHosted: { value: false }, + mockIsAzureConfigured: { value: false }, + mockIsOllamaConfigured: { value: false }, + mockIsOpenRouterConfigured: { value: false }, + })) const { mockGetHostedModels, @@ -44,6 +46,12 @@ vi.mock('@/lib/core/config/env-flags', () => ({ get isOllamaConfigured() { return mockIsOllamaConfigured.value }, + get isOpenRouterConfigured() { + return mockIsOpenRouterConfigured.value + }, + get isCohereConfigured() { + return false + }, })) vi.mock('@/providers/models', () => ({ @@ -106,6 +114,7 @@ describe('getApiKeyCondition / shouldRequireApiKeyForModel', () => { mockIsHosted.value = false mockIsAzureConfigured.value = false mockIsOllamaConfigured.value = false + mockIsOpenRouterConfigured.value = false mockProviders.value = { base: { models: [], isLoading: false }, ollama: { models: [], isLoading: false }, @@ -210,6 +219,13 @@ describe('getApiKeyCondition / shouldRequireApiKeyForModel', () => { expect(evaluateCondition('openrouter/anthropic/claude')).toBe(true) }) + it('does not require API key for openrouter/ models when OpenRouter is configured', () => { + mockIsOpenRouterConfigured.value = true + expect(evaluateCondition('openrouter/deepseek/deepseek-v4-flash')).toBe(false) + mockProviders.value.openrouter.models = ['openrouter/anthropic/claude'] + expect(evaluateCondition('openrouter/anthropic/claude')).toBe(false) + }) + it('is case-insensitive for store lookup', () => { mockProviders.value.ollama.models = ['Llama3:Latest'] expect(evaluateCondition('llama3:latest')).toBe(false) diff --git a/apps/sim/blocks/utils.ts b/apps/sim/blocks/utils.ts index f70ab7b06ee..1027ec371e9 100644 --- a/apps/sim/blocks/utils.ts +++ b/apps/sim/blocks/utils.ts @@ -4,6 +4,7 @@ import { isCohereConfigured, isHosted, isOllamaConfigured, + isOpenRouterConfigured, } from '@/lib/core/config/env-flags' import { getScopesForService } from '@/lib/oauth/utils' import { buildCanonicalIndex } from '@/lib/workflows/subblocks/visibility' @@ -199,6 +200,13 @@ function shouldRequireApiKeyForModel(model: string): boolean { ) { return false } + if ( + isOpenRouterConfigured && + (normalizedModel.startsWith('openrouter/') || + getProviderFromStore(normalizedModel) === 'openrouter') + ) { + return false + } if (normalizedModel.startsWith('vllm/') || normalizedModel.startsWith('litellm/')) { return false } diff --git a/apps/sim/lib/api-key/byok.ts b/apps/sim/lib/api-key/byok.ts index aa456962264..8fb2ae171f8 100644 --- a/apps/sim/lib/api-key/byok.ts +++ b/apps/sim/lib/api-key/byok.ts @@ -187,6 +187,20 @@ export async function getApiKeyWithBYOK( throw new Error(`API key is required for Ollama Cloud ${model}`) } + const isOpenRouterModel = + provider === 'openrouter' || + model.startsWith('openrouter/') || + useProvidersStore.getState().providers.openrouter.models.includes(model) + if (isOpenRouterModel) { + if (userProvidedKey) { + return { apiKey: userProvidedKey, isBYOK: false } + } + if (env.OPENROUTER_API_KEY) { + return { apiKey: env.OPENROUTER_API_KEY, isBYOK: false } + } + throw new Error(`API key is required for OpenRouter ${model}`) + } + const isBedrockModel = provider === 'bedrock' || model.startsWith('bedrock/') if (isBedrockModel) { return { apiKey: PROVIDER_PLACEHOLDER_KEY, isBYOK: false } 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-flags.ts b/apps/sim/lib/core/config/env-flags.ts index eaeeaabcac8..12e80ee9850 100644 --- a/apps/sim/lib/core/config/env-flags.ts +++ b/apps/sim/lib/core/config/env-flags.ts @@ -261,6 +261,16 @@ export const isAzureConfigured = isTruthy(getEnv('NEXT_PUBLIC_AZURE_CONFIGURED') */ export const isCohereConfigured = isTruthy(getEnv('NEXT_PUBLIC_COHERE_CONFIGURED')) +/** + * Whether an OpenRouter API key is pre-configured (`OPENROUTER_API_KEY`). + * When true, Agent/Router/Evaluator blocks hide the API Key field for OpenRouter models + * and resolve the key from the environment at execution time. + * Client bundles also honor `NEXT_PUBLIC_OPENROUTER_CONFIGURED=true` so the field stays + * hidden in the editor (server-only `OPENROUTER_API_KEY` never reaches the browser). + */ +export const isOpenRouterConfigured = + Boolean(env.OPENROUTER_API_KEY) || isTruthy(getEnv('NEXT_PUBLIC_OPENROUTER_CONFIGURED')) + /** * Are invitations disabled globally * When true, workspace invitations are disabled for all users diff --git a/apps/sim/lib/core/config/env.ts b/apps/sim/lib/core/config/env.ts index c86eb9e7ff4..646a76fa9d9 100644 --- a/apps/sim/lib/core/config/env.ts +++ b/apps/sim/lib/core/config/env.ts @@ -171,6 +171,7 @@ export const env = createEnv({ LITELLM_API_KEY: z.string().optional(), // Optional bearer token for LiteLLM FIREWORKS_API_KEY: z.string().optional(), // Optional Fireworks AI API key for model listing TOGETHER_API_KEY: z.string().optional(), // Optional Together AI API key for model listing and inference + OPENROUTER_API_KEY: z.string().optional(), // Optional OpenRouter API key for model listing and inference BASETEN_API_KEY: z.string().optional(), // Optional Baseten API key for model listing and inference COHERE_API_KEY: z.string().min(1).optional(), // Cohere API key for reranker (rerank-v4.0-pro, rerank-v4.0-fast, rerank-v3.5) COHERE_API_KEY_1: z.string().min(1).optional(), // Primary Cohere API key for rotation @@ -294,7 +295,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) @@ -540,6 +541,7 @@ export const env = createEnv({ NEXT_PUBLIC_BEDROCK_DEFAULT_CREDENTIALS: z.string().optional(), // Hide Bedrock credential fields when deployment uses AWS default credential chain (IAM roles, instance profiles, ECS task roles, IRSA) NEXT_PUBLIC_AZURE_CONFIGURED: z.string().optional(), // Hide Azure credential fields when endpoint/key/version are pre-configured server-side NEXT_PUBLIC_COHERE_CONFIGURED: z.string().optional(), // Hide Cohere API key field on Knowledge block when COHERE_API_KEY is pre-configured server-side + NEXT_PUBLIC_OPENROUTER_CONFIGURED: z.string().optional(), // Hide OpenRouter API key field when OPENROUTER_API_KEY is pre-configured server-side NEXT_PUBLIC_COPILOT_TRAINING_ENABLED: z.string().optional(), NEXT_PUBLIC_ENABLE_PLAYGROUND: z.string().optional(), // Enable component playground at /playground NEXT_PUBLIC_DOCUMENTATION_URL: z.string().url().optional(), // Custom documentation URL @@ -613,6 +615,7 @@ export const env = createEnv({ NEXT_PUBLIC_BEDROCK_DEFAULT_CREDENTIALS: process.env.NEXT_PUBLIC_BEDROCK_DEFAULT_CREDENTIALS, NEXT_PUBLIC_AZURE_CONFIGURED: process.env.NEXT_PUBLIC_AZURE_CONFIGURED, NEXT_PUBLIC_COHERE_CONFIGURED: process.env.NEXT_PUBLIC_COHERE_CONFIGURED, + NEXT_PUBLIC_OPENROUTER_CONFIGURED: process.env.NEXT_PUBLIC_OPENROUTER_CONFIGURED, NEXT_PUBLIC_COPILOT_TRAINING_ENABLED: process.env.NEXT_PUBLIC_COPILOT_TRAINING_ENABLED, NEXT_PUBLIC_ENABLE_PLAYGROUND: process.env.NEXT_PUBLIC_ENABLE_PLAYGROUND, NEXT_PUBLIC_POSTHOG_ENABLED: process.env.NEXT_PUBLIC_POSTHOG_ENABLED, 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/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/bun.lock b/bun.lock index f3ad75a78dd..7e81dd4365f 100644 --- a/bun.lock +++ b/bun.lock @@ -1,5 +1,6 @@ { "lockfileVersion": 1, + "configVersion": 0, "workspaces": { "": { "name": "simstudio", @@ -3085,7 +3086,7 @@ "lru.min": ["lru.min@1.1.4", "", {}, "sha512-DqC6n3QQ77zdFpCMASA1a3Jlb64Hv2N2DciFGkO/4L9+q/IpIAuRlKOvCXabtRW6cQf8usbmM6BE/TOPysCdIA=="], - "lucide-react": ["lucide-react@1.23.0", "", { "peerDependencies": { "react": "^16.5.1 || ^17.0.0 || ^18.0.0 || ^19.0.0" } }, "sha512-38BpJcD0JhFosxHApP/BYsBetLpQFRoTRzEzstM/XCc3jsAG7wqaY1lgVwxiUe3xqYE+lNxo2PkCmYwXWrwwIw=="], + "lucide-react": ["lucide-react@0.479.0", "", { "peerDependencies": { "react": "^16.5.1 || ^17.0.0 || ^18.0.0 || ^19.0.0" } }, "sha512-aBhNnveRhorBOK7uA4gDjgaf+YlHMdMhQ/3cupk6exM10hWlEU+2QtWYOfhXhjAsmdb6LeKR+NZnow4UxRRiTQ=="], "magic-string": ["magic-string@0.30.21", "", { "dependencies": { "@jridgewell/sourcemap-codec": "^1.5.5" } }, "sha512-vd2F4YUyEXKGcLHoq+TEyCjxueSeHnFxyyjNp80yg0XV4vUhnDer/lvvlqM/arB5bXQN5K2/3oinyCRyx8T2CQ=="], @@ -4461,10 +4462,6 @@ "@sim/db/@types/node": ["@types/node@22.19.21", "", { "dependencies": { "undici-types": "~6.21.0" } }, "sha512-VMeFBSCKQKmm2swI2kW51SFusDqekC6q9trBCvJ/JliDchFSuoYYKN7yVNjPthP1HKZcx3U1gI/wTcEBjEFKTA=="], - "@sim/emcn/lucide-react": ["lucide-react@0.479.0", "", { "peerDependencies": { "react": "^16.5.1 || ^17.0.0 || ^18.0.0 || ^19.0.0" } }, "sha512-aBhNnveRhorBOK7uA4gDjgaf+YlHMdMhQ/3cupk6exM10hWlEU+2QtWYOfhXhjAsmdb6LeKR+NZnow4UxRRiTQ=="], - - "@sim/workflow-renderer/lucide-react": ["lucide-react@0.479.0", "", { "peerDependencies": { "react": "^16.5.1 || ^17.0.0 || ^18.0.0 || ^19.0.0" } }, "sha512-aBhNnveRhorBOK7uA4gDjgaf+YlHMdMhQ/3cupk6exM10hWlEU+2QtWYOfhXhjAsmdb6LeKR+NZnow4UxRRiTQ=="], - "@smithy/middleware-compression/fflate": ["fflate@0.8.1", "", {}, "sha512-/exOvEuc+/iaUm105QIiOt4LpBdMTWsXxqR0HDF35vx3fmaKzw7354gTilCh5rkzEt8WYyG//ku3h3nRmd7CHQ=="], "@socket.io/redis-adapter/debug": ["debug@4.3.7", "", { "dependencies": { "ms": "^2.1.3" } }, "sha512-Er2nc/H7RrMXZBFCEim6TCmMk02Z8vLC2Rbi1KEBggpo0fS6l0S1nnapwmIi3yW/+GOJap1Krg4w0Hg80oCqgQ=="], @@ -4663,10 +4660,14 @@ "fumadocs-openapi/ajv": ["ajv@8.20.0", "", { "dependencies": { "fast-deep-equal": "^3.1.3", "fast-uri": "^3.0.1", "json-schema-traverse": "^1.0.0", "require-from-string": "^2.0.2" } }, "sha512-Thbli+OlOj+iMPYFBVBfJ3OmCAnaSyNn4M1vz9T6Gka5Jt9ba/HIR56joy65tY6kx/FCF5VXNB819Y7/GUrBGA=="], + "fumadocs-openapi/lucide-react": ["lucide-react@1.23.0", "", { "peerDependencies": { "react": "^16.5.1 || ^17.0.0 || ^18.0.0 || ^19.0.0" } }, "sha512-38BpJcD0JhFosxHApP/BYsBetLpQFRoTRzEzstM/XCc3jsAG7wqaY1lgVwxiUe3xqYE+lNxo2PkCmYwXWrwwIw=="], + "fumadocs-openapi/tailwind-merge": ["tailwind-merge@3.6.0", "", {}, "sha512-uxL7qAVQriqRQPAyK3pj66VqskWqoZ37PW94jwOTwNfq/z9oyu1V+eqrZqtR2+fCiXdYOZe/Modt8GtvqNzu+w=="], "fumadocs-ui/@radix-ui/react-slot": ["@radix-ui/react-slot@1.3.0", "", { "dependencies": { "@radix-ui/react-compose-refs": "1.1.3" }, "peerDependencies": { "@types/react": "*", "react": "^16.8 || ^17.0 || ^18.0 || ^19.0 || ^19.0.0-rc" }, "optionalPeers": ["@types/react"] }, "sha512-MojKku4U/miO8Av4Dkb+ctMAQx7JmY96LmtDQlAarCRtd7rN52QCSzBF+XAvr5S6coSVj9HEPBgHAHKEJVk/WA=="], + "fumadocs-ui/lucide-react": ["lucide-react@1.23.0", "", { "peerDependencies": { "react": "^16.5.1 || ^17.0.0 || ^18.0.0 || ^19.0.0" } }, "sha512-38BpJcD0JhFosxHApP/BYsBetLpQFRoTRzEzstM/XCc3jsAG7wqaY1lgVwxiUe3xqYE+lNxo2PkCmYwXWrwwIw=="], + "fumadocs-ui/tailwind-merge": ["tailwind-merge@3.6.0", "", {}, "sha512-uxL7qAVQriqRQPAyK3pj66VqskWqoZ37PW94jwOTwNfq/z9oyu1V+eqrZqtR2+fCiXdYOZe/Modt8GtvqNzu+w=="], "gaxios/node-fetch": ["node-fetch@2.7.0", "", { "dependencies": { "whatwg-url": "^5.0.0" }, "peerDependencies": { "encoding": "^0.1.0" }, "optionalPeers": ["encoding"] }, "sha512-c4FRfUm/dbcWZ7U+1Wq0AwCyFL+3nt2bEw05wfxSz+DWpWsitgmSgYmy2dQdWyKC1694ELPqMs/YzUSNozLt8A=="], @@ -4821,8 +4822,6 @@ "sharp/semver": ["semver@7.8.4", "", { "bin": { "semver": "bin/semver.js" } }, "sha512-rUCObTnP32Q08R2uuIrt7r9PlEonuTmtuXYcW6s5kjdlj3xbnwe+21yXptAUYcMAABLkYYTtnmzb3w3EDZfueA=="], - "sim/lucide-react": ["lucide-react@0.479.0", "", { "peerDependencies": { "react": "^16.5.1 || ^17.0.0 || ^18.0.0 || ^19.0.0" } }, "sha512-aBhNnveRhorBOK7uA4gDjgaf+YlHMdMhQ/3cupk6exM10hWlEU+2QtWYOfhXhjAsmdb6LeKR+NZnow4UxRRiTQ=="], - "sim/tailwindcss": ["tailwindcss@3.4.19", "", { "dependencies": { "@alloc/quick-lru": "^5.2.0", "arg": "^5.0.2", "chokidar": "^3.6.0", "didyoumean": "^1.2.2", "dlv": "^1.1.3", "fast-glob": "^3.3.2", "glob-parent": "^6.0.2", "is-glob": "^4.0.3", "jiti": "^1.21.7", "lilconfig": "^3.1.3", "micromatch": "^4.0.8", "normalize-path": "^3.0.0", "object-hash": "^3.0.0", "picocolors": "^1.1.1", "postcss": "^8.4.47", "postcss-import": "^15.1.0", "postcss-js": "^4.0.1", "postcss-load-config": "^4.0.2 || ^5.0 || ^6.0", "postcss-nested": "^6.2.0", "postcss-selector-parser": "^6.1.2", "resolve": "^1.22.8", "sucrase": "^3.35.0" }, "bin": { "tailwind": "lib/cli.js", "tailwindcss": "lib/cli.js" } }, "sha512-3ofp+LL8E+pK/JuPLPggVAIaEuhvIz4qNcf3nA1Xn2o/7fb7s/TYpHhwGDv1ZU3PkBluUVaF8PyCHcm48cKLWQ=="], "simstudio/@types/node": ["@types/node@20.19.43", "", { "dependencies": { "undici-types": "~6.21.0" } }, "sha512-6oYBAi5ikg4Pl+kGsoYtawUMBT2zZMCvPNF7pVLnHZfd1zf38DRiWn/gT01RYCdUqkv7Fhr+C9ot4/tb+2sVvA=="], diff --git a/docs/workflow-throughput-one-pager.md b/docs/workflow-throughput-one-pager.md new file mode 100644 index 00000000000..d09f1581c8d --- /dev/null +++ b/docs/workflow-throughput-one-pager.md @@ -0,0 +1,148 @@ +# Sim Workflow Execution Throughput + +## Executive summary + +The measured local failure was not “the API cannot return fast enough.” It was: + +> Sim returned `202 Accepted` faster than workflows finished. The admission +> permit was released when the HTTP response returned, while the local async +> execution continued in the same process. Active work accumulated until the +> process became slow and unreachable. + +Throughput is **completed workflows per second**, not `202` responses per +second. This work improves overload behavior and protects the process; it does +not yet claim a production throughput increase. + +Scope: local database async backend, synthetic workflow, one web process. These +results are a falsifiable local control-flow finding, not a production capacity +number. + +## Before and after + +### Before: `202` released capacity too early + +```mermaid +sequenceDiagram + participant C as Caller + participant R as ExecuteRoute + participant G as AdmissionGate + participant Q as DatabaseBackend + participant E as InlineExecutor + + 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 + Note over E: More requests can enter
while old runs are still active + E-->>Q: complete +``` + +### After: permit represents real in-process work + +```mermaid +sequenceDiagram + participant C as Caller + participant R as ExecuteRoute + participant G as AdmissionGate + participant Q as DatabaseBackend + participant E as InlineExecutor + + 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 + E-->>Q: complete + Q->>G: release permit + else capacity full + G-->>R: no permit + R-->>C: 429 Retry-After + end +``` + +The same admission behavior now applies to cookie-backed **async** requests, +which represent the normal authenticated UI path. Session-backed sync/SSE +requests retain their existing response/stream lifecycle. + +## Evidence + +Synthetic workflow: `Start → Function`, with 300ms of deterministic CPU work. +Load profile: async `POST /api/workflows/{id}/execute`, ramp to 8 requests/sec, +then hold for 60 seconds. + +- Before the fix: approximately 218 successful `202` responses, 225 + `ECONNREFUSED` failures, and roughly 2 completed workflows/sec. The process + became unreachable. +- At a more aggressive 30 requests/sec probe: 351 requests returned `202`, + approximately one completion was observed in the window, and more than 280 + executions remained running. +- With `ADMISSION_GATE_MAX_INFLIGHT=5`: 207 `429` responses and repeated + admission-rejection logs confirmed that the gate rejects at capacity. +- After the fix: the 8 requests/sec rerun produced explicit `202`/`429` + backpressure and did not reproduce the previous connection-refused cascade. + One transient health check missed, so this is stability evidence, not a + claim that the process was perfectly healthy every second. + +## Changes in this branch + +- Hold the admission ticket through local inline async execution: + [execute route](../apps/sim/app/api/workflows/[id]/execute/route.ts) +- Apply the gate to session-backed async requests while preserving sync/SSE: + [execute route](../apps/sim/app/api/workflows/[id]/execute/route.ts) +- Lower the default in-process admission limit from 500 to 10: + [admission gate](../apps/sim/lib/core/admission/gate.ts) +- Add a 4 GB Node heap ceiling to the production start script: + [Sim package scripts](../apps/sim/package.json) +- Add workflow execution count/duration telemetry: + [execution metrics](../apps/sim/lib/workflows/executor/execution-metrics.ts) + +The limit of 10 is a safety starting point, not an optimized production +setting. It bounds damage; it does not make execution faster. + +## Other chokepoints found or tested + +### Proven, not fixed yet + +- **Execution cost:** CPU/isolated-runtime work determines how long a permit + remains occupied. More expensive workflows reduce completion rate and cause + honest `429` backpressure. +- **Web database pool:** local execution shares the web process and its + `primaryMax=10` pool. The sustainable limit depends on queries per execution. +- **Trigger worker pool:** with `SIM_DB_ROLE=trigger`, the pool is + `primaryMax=5`. A 12-way sleep probe took about 6.06 seconds versus 4.04 + seconds on the web pool, with peak active connections of 5. +- **Sync/SSE process pressure:** session-backed sync/SSE requests intentionally + remain outside this async admission change. Their execution and streaming + behavior needs a separate test because they run through the web process. + +### Tested but inconclusive or not run + +- A 20-request synchronous blast completed successfully; it did not reproduce + an OOM. +- Trigger.dev’s 75/75/50 workflow, webhook, and resume queue contention was + not tested in this session. +- No production workflow bodies were available. The synthetic graph proves the + backlog mechanism, not production workload capacity. + +## Next falsifiable measurement + +Run the same profile through both API-key and authenticated-cookie async +requests. Record: + +1. terminal completions/sec from `workflow_execution_logs.ended_at`; +2. `202` versus `429` responses; +3. running execution count; +4. process health and RSS; +5. `pg_stat_activity` for `sim-app`. + +Then vary deterministic workflow work from 300ms to 2 seconds. A valid next +finding is: completion rate falls and `429` rate rises while the process stays +healthy. That would identify execution cost as the next ceiling without +pretending to know production input distributions. diff --git a/packages/testing/src/mocks/env-flags.mock.ts b/packages/testing/src/mocks/env-flags.mock.ts index 590a5b9fde6..7f1c552cae9 100644 --- a/packages/testing/src/mocks/env-flags.mock.ts +++ b/packages/testing/src/mocks/env-flags.mock.ts @@ -32,6 +32,7 @@ export const envFlagsMock = { isE2BDocEnabled: false, isOllamaConfigured: false, isAzureConfigured: false, + isOpenRouterConfigured: false, isInvitationsDisabled: false, isPublicApiDisabled: false, isGoogleAuthDisabled: false, From cd571586a9c7f0f827a68504e1dfed3a92ba3426 Mon Sep 17 00:00:00 2001 From: akshay326 Date: Tue, 21 Jul 2026 01:06:50 +0000 Subject: [PATCH 2/5] docs(execution): ground throughput one-pager in Sim paths Co-authored-by: Cursor --- docs/workflow-throughput-one-pager.md | 174 ++++++++++++++++++++------ 1 file changed, 137 insertions(+), 37 deletions(-) diff --git a/docs/workflow-throughput-one-pager.md b/docs/workflow-throughput-one-pager.md index d09f1581c8d..12a6e960b94 100644 --- a/docs/workflow-throughput-one-pager.md +++ b/docs/workflow-throughput-one-pager.md @@ -1,21 +1,36 @@ -# Sim Workflow Execution Throughput +# Sim Workflow Execution: Async, Inline, and Database ## Executive summary -The measured local failure was not “the API cannot return fast enough.” It was: +The local failure was not that the execute route could not return `202` +quickly. It was: > Sim returned `202 Accepted` faster than workflows finished. The admission > permit was released when the HTTP response returned, while the local async > execution continued in the same process. Active work accumulated until the > process became slow and unreachable. -Throughput is **completed workflows per second**, not `202` responses per -second. This work improves overload behavior and protects the process; it does -not yet claim a production throughput increase. +The execute route has three existing execution modes: -Scope: local database async backend, synthetic workflow, one web process. These -results are a falsifiable local control-flow finding, not a production capacity -number. +- `sync`: the request waits for `executeWorkflowCore` and returns JSON. +- `stream`: the request returns an SSE response while `executeWorkflow` runs. +- `async`: the request queues a `workflow-execution` job and returns `202`. + +`async` does not by itself mean that execution leaves the web process. The +existing async backend decides that: + +- The **database backend** calls `shouldExecuteInline() === true` and runs + `executeWorkflowJob` in the same process after inserting the job. +- **Trigger.dev** calls `shouldExecuteInline() === false` and runs the task in + the Trigger.dev worker. + +Throughput is completed workflow executions per second, not `202` responses +per second. This work improves overload behavior and protects the process; it +does not claim a production throughput increase. + +Scope: local database backend, synthetic workflow, one web process. The +control-flow finding is proven by the route and backend code below. The local +load numbers are stability evidence, not a production capacity number. ## Before and after @@ -26,8 +41,8 @@ sequenceDiagram participant C as Caller participant R as ExecuteRoute participant G as AdmissionGate - participant Q as DatabaseBackend - participant E as InlineExecutor + participant Q as DatabaseJobQueue + participant E as executeWorkflowJob C->>R: POST /execute R->>G: tryAdmit() @@ -48,8 +63,8 @@ sequenceDiagram participant C as Caller participant R as ExecuteRoute participant G as AdmissionGate - participant Q as DatabaseBackend - participant E as InlineExecutor + participant Q as DatabaseJobQueue + participant E as executeWorkflowJob C->>R: POST /execute R->>G: tryAdmit() @@ -67,15 +82,83 @@ sequenceDiagram end ``` -The same admission behavior now applies to cookie-backed **async** requests, -which represent the normal authenticated UI path. Session-backed sync/SSE -requests retain their existing response/stream lifecycle. +The same admission behavior now applies to cookie-backed `async` requests, +which represent the normal authenticated UI path. Session-backed `sync` and +SSE requests retain their existing response/stream lifecycle. + +## Database in the current setup + +The repository proves that the database can be a shared capacity limit, but it +does not prove that database latency caused the local connection-refused +cascade. The evidence is: + +- `packages/db/db.ts` sets `primaryMax=10` for the web role and `primaryMax=5` + for the Trigger.dev role. The same file documents that one run can need + three or more simultaneous connections because of parallel queries and + overlapping logging writes. +- `DatabaseJobQueue.enqueue` inserts into `async_jobs`. The inline runner then + updates the job to `processing`, runs `executeWorkflowJob`, and updates the + job to `completed` or `failed` + ([database backend](../apps/sim/lib/core/async-jobs/backends/database.ts)). +- `executeWorkflowJob` runs preprocessing, `executeWorkflowCore`, logging + finalization, and post-execution work + ([workflow execution](../apps/sim/background/workflow-execution.ts)). +- `executeWorkflowCore` persists block-start and block-complete lifecycle + markers through `LoggingSession`. The durable fallback is a + `workflow_execution_logs` JSONB update + ([execution core](../apps/sim/lib/workflows/executor/execution-core.ts), + [logging session](../apps/sim/lib/logs/execution/logging-session.ts)). +- Completion reads and updates `workflow_execution_logs`, may externalize + execution data, and reconciles usage in a transaction + ([execution logger](../apps/sim/lib/logs/execution/logger.ts)). +- `workflow_execution_logs.execution_id` and the relevant `async_jobs` status + indexes exist in `packages/db/schema.ts`. Indexes establish the intended + access paths; they do not establish query latency under load. + +Therefore, the DB distinction is: + +- With the **database backend**, web-process execution and its `async_jobs`, + execution-log, and usage queries share the web pool. +- With **Trigger.dev**, the web request mostly pays for enqueueing; the + Trigger.dev task still performs execution-log and usage queries, but those + queries use the Trigger.dev pool and the task queue limit. +- With `sync` and SSE, the web request itself waits on the same execution and + database work. With `async` plus the database backend, the request returns + before that work finishes, but the work still occupies the web process and + web pool. + +No repository-only claim should call the DB the measured root cause without +`pg_stat_activity`, pool-wait, query-duration, or equivalent production +measurements. + +## Async and inline in Sim + +These are not opposite endpoint modes: + +1. `sync` calls `executeWorkflowCore` from the execute route; SSE calls + `executeWorkflow` through `createStreamingResponse`. Both keep the + response/stream lifecycle attached to the run. +2. `async` calls `getJobQueue().enqueue('workflow-execution', ...)` and returns + `202`. +3. The database backend starts the job and calls `executeWorkflowJob` in the + web process. The admission ticket must remain held until that call finishes. +4. Trigger.dev starts the task outside the web process. The route releases its + ticket after the enqueue response; Trigger.dev applies + `WORKFLOW_EXECUTION_CONCURRENCY_LIMIT`, whose default is `75`. + +The current branch fixes only the database-backend case: it retains the +admission ticket for in-process `async` execution and releases it in the +inline runner's `finally` block. It does not change the database pool size, +Trigger.dev concurrency, or execution cost. ## Evidence Synthetic workflow: `Start → Function`, with 300ms of deterministic CPU work. Load profile: async `POST /api/workflows/{id}/execute`, ramp to 8 requests/sec, -then hold for 60 seconds. +then hold for 60 seconds +([Artillery profile](../apps/sim/scripts/load/workflow-concurrency.yml)). +The profile asserts the enqueue response is `202`; completion counts require +the execution-log query below. - Before the fix: approximately 218 successful `202` responses, 225 `ECONNREFUSED` failures, and roughly 2 completed workflows/sec. The process @@ -84,7 +167,8 @@ then hold for 60 seconds. approximately one completion was observed in the window, and more than 280 executions remained running. - With `ADMISSION_GATE_MAX_INFLIGHT=5`: 207 `429` responses and repeated - admission-rejection logs confirmed that the gate rejects at capacity. + admission-rejection logs confirmed that the admission gate rejects at + capacity. - After the fix: the 8 requests/sec rerun produced explicit `202`/`429` backpressure and did not reproduce the previous connection-refused cascade. One transient health check missed, so this is stability evidence, not a @@ -106,21 +190,24 @@ then hold for 60 seconds. The limit of 10 is a safety starting point, not an optimized production setting. It bounds damage; it does not make execution faster. -## Other chokepoints found or tested - -### Proven, not fixed yet - -- **Execution cost:** CPU/isolated-runtime work determines how long a permit - remains occupied. More expensive workflows reduce completion rate and cause - honest `429` backpressure. -- **Web database pool:** local execution shares the web process and its - `primaryMax=10` pool. The sustainable limit depends on queries per execution. -- **Trigger worker pool:** with `SIM_DB_ROLE=trigger`, the pool is - `primaryMax=5`. A 12-way sleep probe took about 6.06 seconds versus 4.04 - seconds on the web pool, with peak active connections of 5. -- **Sync/SSE process pressure:** session-backed sync/SSE requests intentionally - remain outside this async admission change. Their execution and streaming - behavior needs a separate test because they run through the web process. +## Other limits found or tested + +### Proven by code, not measured as the local root cause + +- **Execution cost:** CPU/isolated-runtime work determines how long an + admission ticket remains occupied. More expensive workflow executions reduce + completion rate and cause honest `429` backpressure. +- **Web database pool:** database-backend execution shares the web process and + its `primaryMax=10` pool. The sustainable rate depends on the DB work per + execution. +- **Trigger.dev task and DB limits:** the task definition sets a default + concurrency limit of `75`, while the Trigger.dev DB role has a + `primaryMax=5` pool. Queue concurrency and DB connection availability are + separate limits. +- **Sync/SSE process pressure:** session-backed `sync` and SSE requests + intentionally remain outside this async admission change. Their execution + and streaming behavior needs a separate test because they run through the + web process. ### Tested but inconclusive or not run @@ -136,13 +223,26 @@ setting. It bounds damage; it does not make execution faster. Run the same profile through both API-key and authenticated-cookie async requests. Record: -1. terminal completions/sec from `workflow_execution_logs.ended_at`; +1. completed workflow executions/sec from + `workflow_execution_logs.ended_at`; 2. `202` versus `429` responses; 3. running execution count; 4. process health and RSS; -5. `pg_stat_activity` for `sim-app`. +5. `pg_stat_activity` for `sim-app` and `sim-trigger`. + +The following query makes the DB part falsifiable during the run: + +```sql +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; +``` Then vary deterministic workflow work from 300ms to 2 seconds. A valid next -finding is: completion rate falls and `429` rate rises while the process stays -healthy. That would identify execution cost as the next ceiling without -pretending to know production input distributions. +finding is: completed workflow executions/sec falls and `429` rate rises +while the process stays healthy. A DB finding requires the same run to show +pool saturation or query wait/duration growth, rather than inferring it from +`202` latency alone. From 0085e687b5ab210a64dcd2f8d840e54e79ffba10 Mon Sep 17 00:00:00 2001 From: akshay326 Date: Tue, 21 Jul 2026 01:08:15 +0000 Subject: [PATCH 3/5] test(execution): cover inline admission lease Co-authored-by: Cursor --- .../[id]/execute/route.async.test.ts | 52 +++++++++++++++++-- .../app/api/workflows/[id]/execute/route.ts | 9 ---- apps/sim/scripts/load/README.md | 6 +-- 3 files changed, 50 insertions(+), 17 deletions(-) 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 803d114d018..1ddab4a3ddd 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', () => ({ @@ -288,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}`, @@ -392,6 +404,36 @@ 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', expect.anything()) + }) + it('applies admission backpressure to session-backed async executions', async () => { hybridAuthMockFns.mockHasExternalApiCredentials.mockReturnValue(false) const heldTickets = Array.from({ length: getAdmissionGateStatus().maxInflight }, () => diff --git a/apps/sim/app/api/workflows/[id]/execute/route.ts b/apps/sim/app/api/workflows/[id]/execute/route.ts index 3b8427e076e..d3604092f2b 100644 --- a/apps/sim/app/api/workflows/[id]/execute/route.ts +++ b/apps/sim/app/api/workflows/[id]/execute/route.ts @@ -285,11 +285,6 @@ type AsyncExecutionParams = { interface AsyncExecutionResult { response: NextResponse retainExecutionClaim: boolean - /** - * When true, the caller must not release the admission ticket — ownership - * transferred to the in-process inline runner which releases on completion. - */ - retainAdmissionTicket: boolean } /** Mutable bag so async inline work can retain the HTTP admission ticket. */ @@ -391,7 +386,6 @@ async function handleAsyncExecution( return { response: NextResponse.json({ error: 'Failed to queue async execution' }, { status: 500 }), retainExecutionClaim: false, - retainAdmissionTicket: false, } } @@ -443,7 +437,6 @@ async function handleAsyncExecution( return { response: NextResponse.json({ error: 'Failed to queue async execution' }, { status: 500 }), retainExecutionClaim: false, - retainAdmissionTicket: false, } } @@ -457,7 +450,6 @@ async function handleAsyncExecution( { status: 503, headers: { [WORKFLOW_EXECUTION_ID_HEADER]: executionId } } ), retainExecutionClaim: true, - retainAdmissionTicket: false, } } @@ -521,7 +513,6 @@ async function handleAsyncExecution( { status: 202 } ), retainExecutionClaim: true, - retainAdmissionTicket: executeInline, } } diff --git a/apps/sim/scripts/load/README.md b/apps/sim/scripts/load/README.md index 43a08976431..5e111824e6b 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= \ @@ -103,5 +103,5 @@ 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` +- 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 From 4ff34ea64d8f37814dc7b5885d0c5e15df0ac789 Mon Sep 17 00:00:00 2001 From: akshay326 Date: Tue, 21 Jul 2026 01:08:53 +0000 Subject: [PATCH 4/5] test(execution): match completed job payload Co-authored-by: Cursor --- apps/sim/app/api/workflows/[id]/execute/route.async.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 1ddab4a3ddd..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 @@ -431,7 +431,7 @@ describe('workflow execute async route', () => { await vi.waitFor(() => { expect(getAdmissionGateStatus().inflight).toBe(0) }) - expect(mockCompleteJob).toHaveBeenCalledWith('job-123', expect.anything()) + expect(mockCompleteJob).toHaveBeenCalledWith('job-123', undefined) }) it('applies admission backpressure to session-backed async executions', async () => { From e7398192a434ec42807f6310e2e91328c3e5fcd5 Mon Sep 17 00:00:00 2001 From: akshay326 Date: Tue, 21 Jul 2026 01:38:49 +0000 Subject: [PATCH 5/5] docs(execution): tighten throughput evidence Co-authored-by: Cursor --- apps/sim/blocks/utils.test.ts | 26 +- apps/sim/blocks/utils.ts | 8 - apps/sim/lib/api-key/byok.ts | 14 - apps/sim/lib/core/config/env-flags.ts | 10 - apps/sim/lib/core/config/env.ts | 3 - apps/sim/scripts/load/README.md | 2 +- bun.lock | 13 +- docs/workflow-throughput-one-pager.md | 336 ++++++++----------- packages/testing/src/mocks/env-flags.mock.ts | 1 - 9 files changed, 153 insertions(+), 260 deletions(-) diff --git a/apps/sim/blocks/utils.test.ts b/apps/sim/blocks/utils.test.ts index 911188bb62e..3dc571b9dd7 100644 --- a/apps/sim/blocks/utils.test.ts +++ b/apps/sim/blocks/utils.test.ts @@ -3,13 +3,11 @@ */ import { beforeEach, describe, expect, it, vi } from 'vitest' -const { mockIsHosted, mockIsAzureConfigured, mockIsOllamaConfigured, mockIsOpenRouterConfigured } = - vi.hoisted(() => ({ - mockIsHosted: { value: false }, - mockIsAzureConfigured: { value: false }, - mockIsOllamaConfigured: { value: false }, - mockIsOpenRouterConfigured: { value: false }, - })) +const { mockIsHosted, mockIsAzureConfigured, mockIsOllamaConfigured } = vi.hoisted(() => ({ + mockIsHosted: { value: false }, + mockIsAzureConfigured: { value: false }, + mockIsOllamaConfigured: { value: false }, +})) const { mockGetHostedModels, @@ -46,12 +44,6 @@ vi.mock('@/lib/core/config/env-flags', () => ({ get isOllamaConfigured() { return mockIsOllamaConfigured.value }, - get isOpenRouterConfigured() { - return mockIsOpenRouterConfigured.value - }, - get isCohereConfigured() { - return false - }, })) vi.mock('@/providers/models', () => ({ @@ -114,7 +106,6 @@ describe('getApiKeyCondition / shouldRequireApiKeyForModel', () => { mockIsHosted.value = false mockIsAzureConfigured.value = false mockIsOllamaConfigured.value = false - mockIsOpenRouterConfigured.value = false mockProviders.value = { base: { models: [], isLoading: false }, ollama: { models: [], isLoading: false }, @@ -219,13 +210,6 @@ describe('getApiKeyCondition / shouldRequireApiKeyForModel', () => { expect(evaluateCondition('openrouter/anthropic/claude')).toBe(true) }) - it('does not require API key for openrouter/ models when OpenRouter is configured', () => { - mockIsOpenRouterConfigured.value = true - expect(evaluateCondition('openrouter/deepseek/deepseek-v4-flash')).toBe(false) - mockProviders.value.openrouter.models = ['openrouter/anthropic/claude'] - expect(evaluateCondition('openrouter/anthropic/claude')).toBe(false) - }) - it('is case-insensitive for store lookup', () => { mockProviders.value.ollama.models = ['Llama3:Latest'] expect(evaluateCondition('llama3:latest')).toBe(false) diff --git a/apps/sim/blocks/utils.ts b/apps/sim/blocks/utils.ts index 1027ec371e9..f70ab7b06ee 100644 --- a/apps/sim/blocks/utils.ts +++ b/apps/sim/blocks/utils.ts @@ -4,7 +4,6 @@ import { isCohereConfigured, isHosted, isOllamaConfigured, - isOpenRouterConfigured, } from '@/lib/core/config/env-flags' import { getScopesForService } from '@/lib/oauth/utils' import { buildCanonicalIndex } from '@/lib/workflows/subblocks/visibility' @@ -200,13 +199,6 @@ function shouldRequireApiKeyForModel(model: string): boolean { ) { return false } - if ( - isOpenRouterConfigured && - (normalizedModel.startsWith('openrouter/') || - getProviderFromStore(normalizedModel) === 'openrouter') - ) { - return false - } if (normalizedModel.startsWith('vllm/') || normalizedModel.startsWith('litellm/')) { return false } diff --git a/apps/sim/lib/api-key/byok.ts b/apps/sim/lib/api-key/byok.ts index 8fb2ae171f8..aa456962264 100644 --- a/apps/sim/lib/api-key/byok.ts +++ b/apps/sim/lib/api-key/byok.ts @@ -187,20 +187,6 @@ export async function getApiKeyWithBYOK( throw new Error(`API key is required for Ollama Cloud ${model}`) } - const isOpenRouterModel = - provider === 'openrouter' || - model.startsWith('openrouter/') || - useProvidersStore.getState().providers.openrouter.models.includes(model) - if (isOpenRouterModel) { - if (userProvidedKey) { - return { apiKey: userProvidedKey, isBYOK: false } - } - if (env.OPENROUTER_API_KEY) { - return { apiKey: env.OPENROUTER_API_KEY, isBYOK: false } - } - throw new Error(`API key is required for OpenRouter ${model}`) - } - const isBedrockModel = provider === 'bedrock' || model.startsWith('bedrock/') if (isBedrockModel) { return { apiKey: PROVIDER_PLACEHOLDER_KEY, isBYOK: false } diff --git a/apps/sim/lib/core/config/env-flags.ts b/apps/sim/lib/core/config/env-flags.ts index 12e80ee9850..eaeeaabcac8 100644 --- a/apps/sim/lib/core/config/env-flags.ts +++ b/apps/sim/lib/core/config/env-flags.ts @@ -261,16 +261,6 @@ export const isAzureConfigured = isTruthy(getEnv('NEXT_PUBLIC_AZURE_CONFIGURED') */ export const isCohereConfigured = isTruthy(getEnv('NEXT_PUBLIC_COHERE_CONFIGURED')) -/** - * Whether an OpenRouter API key is pre-configured (`OPENROUTER_API_KEY`). - * When true, Agent/Router/Evaluator blocks hide the API Key field for OpenRouter models - * and resolve the key from the environment at execution time. - * Client bundles also honor `NEXT_PUBLIC_OPENROUTER_CONFIGURED=true` so the field stays - * hidden in the editor (server-only `OPENROUTER_API_KEY` never reaches the browser). - */ -export const isOpenRouterConfigured = - Boolean(env.OPENROUTER_API_KEY) || isTruthy(getEnv('NEXT_PUBLIC_OPENROUTER_CONFIGURED')) - /** * Are invitations disabled globally * When true, workspace invitations are disabled for all users diff --git a/apps/sim/lib/core/config/env.ts b/apps/sim/lib/core/config/env.ts index 646a76fa9d9..cf202b68aca 100644 --- a/apps/sim/lib/core/config/env.ts +++ b/apps/sim/lib/core/config/env.ts @@ -171,7 +171,6 @@ export const env = createEnv({ LITELLM_API_KEY: z.string().optional(), // Optional bearer token for LiteLLM FIREWORKS_API_KEY: z.string().optional(), // Optional Fireworks AI API key for model listing TOGETHER_API_KEY: z.string().optional(), // Optional Together AI API key for model listing and inference - OPENROUTER_API_KEY: z.string().optional(), // Optional OpenRouter API key for model listing and inference BASETEN_API_KEY: z.string().optional(), // Optional Baseten API key for model listing and inference COHERE_API_KEY: z.string().min(1).optional(), // Cohere API key for reranker (rerank-v4.0-pro, rerank-v4.0-fast, rerank-v3.5) COHERE_API_KEY_1: z.string().min(1).optional(), // Primary Cohere API key for rotation @@ -541,7 +540,6 @@ export const env = createEnv({ NEXT_PUBLIC_BEDROCK_DEFAULT_CREDENTIALS: z.string().optional(), // Hide Bedrock credential fields when deployment uses AWS default credential chain (IAM roles, instance profiles, ECS task roles, IRSA) NEXT_PUBLIC_AZURE_CONFIGURED: z.string().optional(), // Hide Azure credential fields when endpoint/key/version are pre-configured server-side NEXT_PUBLIC_COHERE_CONFIGURED: z.string().optional(), // Hide Cohere API key field on Knowledge block when COHERE_API_KEY is pre-configured server-side - NEXT_PUBLIC_OPENROUTER_CONFIGURED: z.string().optional(), // Hide OpenRouter API key field when OPENROUTER_API_KEY is pre-configured server-side NEXT_PUBLIC_COPILOT_TRAINING_ENABLED: z.string().optional(), NEXT_PUBLIC_ENABLE_PLAYGROUND: z.string().optional(), // Enable component playground at /playground NEXT_PUBLIC_DOCUMENTATION_URL: z.string().url().optional(), // Custom documentation URL @@ -615,7 +613,6 @@ export const env = createEnv({ NEXT_PUBLIC_BEDROCK_DEFAULT_CREDENTIALS: process.env.NEXT_PUBLIC_BEDROCK_DEFAULT_CREDENTIALS, NEXT_PUBLIC_AZURE_CONFIGURED: process.env.NEXT_PUBLIC_AZURE_CONFIGURED, NEXT_PUBLIC_COHERE_CONFIGURED: process.env.NEXT_PUBLIC_COHERE_CONFIGURED, - NEXT_PUBLIC_OPENROUTER_CONFIGURED: process.env.NEXT_PUBLIC_OPENROUTER_CONFIGURED, NEXT_PUBLIC_COPILOT_TRAINING_ENABLED: process.env.NEXT_PUBLIC_COPILOT_TRAINING_ENABLED, NEXT_PUBLIC_ENABLE_PLAYGROUND: process.env.NEXT_PUBLIC_ENABLE_PLAYGROUND, NEXT_PUBLIC_POSTHOG_ENABLED: process.env.NEXT_PUBLIC_POSTHOG_ENABLED, diff --git a/apps/sim/scripts/load/README.md b/apps/sim/scripts/load/README.md index 5e111824e6b..3dc871a2952 100644 --- a/apps/sim/scripts/load/README.md +++ b/apps/sim/scripts/load/README.md @@ -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 +- 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/bun.lock b/bun.lock index 7e81dd4365f..f3ad75a78dd 100644 --- a/bun.lock +++ b/bun.lock @@ -1,6 +1,5 @@ { "lockfileVersion": 1, - "configVersion": 0, "workspaces": { "": { "name": "simstudio", @@ -3086,7 +3085,7 @@ "lru.min": ["lru.min@1.1.4", "", {}, "sha512-DqC6n3QQ77zdFpCMASA1a3Jlb64Hv2N2DciFGkO/4L9+q/IpIAuRlKOvCXabtRW6cQf8usbmM6BE/TOPysCdIA=="], - "lucide-react": ["lucide-react@0.479.0", "", { "peerDependencies": { "react": "^16.5.1 || ^17.0.0 || ^18.0.0 || ^19.0.0" } }, "sha512-aBhNnveRhorBOK7uA4gDjgaf+YlHMdMhQ/3cupk6exM10hWlEU+2QtWYOfhXhjAsmdb6LeKR+NZnow4UxRRiTQ=="], + "lucide-react": ["lucide-react@1.23.0", "", { "peerDependencies": { "react": "^16.5.1 || ^17.0.0 || ^18.0.0 || ^19.0.0" } }, "sha512-38BpJcD0JhFosxHApP/BYsBetLpQFRoTRzEzstM/XCc3jsAG7wqaY1lgVwxiUe3xqYE+lNxo2PkCmYwXWrwwIw=="], "magic-string": ["magic-string@0.30.21", "", { "dependencies": { "@jridgewell/sourcemap-codec": "^1.5.5" } }, "sha512-vd2F4YUyEXKGcLHoq+TEyCjxueSeHnFxyyjNp80yg0XV4vUhnDer/lvvlqM/arB5bXQN5K2/3oinyCRyx8T2CQ=="], @@ -4462,6 +4461,10 @@ "@sim/db/@types/node": ["@types/node@22.19.21", "", { "dependencies": { "undici-types": "~6.21.0" } }, "sha512-VMeFBSCKQKmm2swI2kW51SFusDqekC6q9trBCvJ/JliDchFSuoYYKN7yVNjPthP1HKZcx3U1gI/wTcEBjEFKTA=="], + "@sim/emcn/lucide-react": ["lucide-react@0.479.0", "", { "peerDependencies": { "react": "^16.5.1 || ^17.0.0 || ^18.0.0 || ^19.0.0" } }, "sha512-aBhNnveRhorBOK7uA4gDjgaf+YlHMdMhQ/3cupk6exM10hWlEU+2QtWYOfhXhjAsmdb6LeKR+NZnow4UxRRiTQ=="], + + "@sim/workflow-renderer/lucide-react": ["lucide-react@0.479.0", "", { "peerDependencies": { "react": "^16.5.1 || ^17.0.0 || ^18.0.0 || ^19.0.0" } }, "sha512-aBhNnveRhorBOK7uA4gDjgaf+YlHMdMhQ/3cupk6exM10hWlEU+2QtWYOfhXhjAsmdb6LeKR+NZnow4UxRRiTQ=="], + "@smithy/middleware-compression/fflate": ["fflate@0.8.1", "", {}, "sha512-/exOvEuc+/iaUm105QIiOt4LpBdMTWsXxqR0HDF35vx3fmaKzw7354gTilCh5rkzEt8WYyG//ku3h3nRmd7CHQ=="], "@socket.io/redis-adapter/debug": ["debug@4.3.7", "", { "dependencies": { "ms": "^2.1.3" } }, "sha512-Er2nc/H7RrMXZBFCEim6TCmMk02Z8vLC2Rbi1KEBggpo0fS6l0S1nnapwmIi3yW/+GOJap1Krg4w0Hg80oCqgQ=="], @@ -4660,14 +4663,10 @@ "fumadocs-openapi/ajv": ["ajv@8.20.0", "", { "dependencies": { "fast-deep-equal": "^3.1.3", "fast-uri": "^3.0.1", "json-schema-traverse": "^1.0.0", "require-from-string": "^2.0.2" } }, "sha512-Thbli+OlOj+iMPYFBVBfJ3OmCAnaSyNn4M1vz9T6Gka5Jt9ba/HIR56joy65tY6kx/FCF5VXNB819Y7/GUrBGA=="], - "fumadocs-openapi/lucide-react": ["lucide-react@1.23.0", "", { "peerDependencies": { "react": "^16.5.1 || ^17.0.0 || ^18.0.0 || ^19.0.0" } }, "sha512-38BpJcD0JhFosxHApP/BYsBetLpQFRoTRzEzstM/XCc3jsAG7wqaY1lgVwxiUe3xqYE+lNxo2PkCmYwXWrwwIw=="], - "fumadocs-openapi/tailwind-merge": ["tailwind-merge@3.6.0", "", {}, "sha512-uxL7qAVQriqRQPAyK3pj66VqskWqoZ37PW94jwOTwNfq/z9oyu1V+eqrZqtR2+fCiXdYOZe/Modt8GtvqNzu+w=="], "fumadocs-ui/@radix-ui/react-slot": ["@radix-ui/react-slot@1.3.0", "", { "dependencies": { "@radix-ui/react-compose-refs": "1.1.3" }, "peerDependencies": { "@types/react": "*", "react": "^16.8 || ^17.0 || ^18.0 || ^19.0 || ^19.0.0-rc" }, "optionalPeers": ["@types/react"] }, "sha512-MojKku4U/miO8Av4Dkb+ctMAQx7JmY96LmtDQlAarCRtd7rN52QCSzBF+XAvr5S6coSVj9HEPBgHAHKEJVk/WA=="], - "fumadocs-ui/lucide-react": ["lucide-react@1.23.0", "", { "peerDependencies": { "react": "^16.5.1 || ^17.0.0 || ^18.0.0 || ^19.0.0" } }, "sha512-38BpJcD0JhFosxHApP/BYsBetLpQFRoTRzEzstM/XCc3jsAG7wqaY1lgVwxiUe3xqYE+lNxo2PkCmYwXWrwwIw=="], - "fumadocs-ui/tailwind-merge": ["tailwind-merge@3.6.0", "", {}, "sha512-uxL7qAVQriqRQPAyK3pj66VqskWqoZ37PW94jwOTwNfq/z9oyu1V+eqrZqtR2+fCiXdYOZe/Modt8GtvqNzu+w=="], "gaxios/node-fetch": ["node-fetch@2.7.0", "", { "dependencies": { "whatwg-url": "^5.0.0" }, "peerDependencies": { "encoding": "^0.1.0" }, "optionalPeers": ["encoding"] }, "sha512-c4FRfUm/dbcWZ7U+1Wq0AwCyFL+3nt2bEw05wfxSz+DWpWsitgmSgYmy2dQdWyKC1694ELPqMs/YzUSNozLt8A=="], @@ -4822,6 +4821,8 @@ "sharp/semver": ["semver@7.8.4", "", { "bin": { "semver": "bin/semver.js" } }, "sha512-rUCObTnP32Q08R2uuIrt7r9PlEonuTmtuXYcW6s5kjdlj3xbnwe+21yXptAUYcMAABLkYYTtnmzb3w3EDZfueA=="], + "sim/lucide-react": ["lucide-react@0.479.0", "", { "peerDependencies": { "react": "^16.5.1 || ^17.0.0 || ^18.0.0 || ^19.0.0" } }, "sha512-aBhNnveRhorBOK7uA4gDjgaf+YlHMdMhQ/3cupk6exM10hWlEU+2QtWYOfhXhjAsmdb6LeKR+NZnow4UxRRiTQ=="], + "sim/tailwindcss": ["tailwindcss@3.4.19", "", { "dependencies": { "@alloc/quick-lru": "^5.2.0", "arg": "^5.0.2", "chokidar": "^3.6.0", "didyoumean": "^1.2.2", "dlv": "^1.1.3", "fast-glob": "^3.3.2", "glob-parent": "^6.0.2", "is-glob": "^4.0.3", "jiti": "^1.21.7", "lilconfig": "^3.1.3", "micromatch": "^4.0.8", "normalize-path": "^3.0.0", "object-hash": "^3.0.0", "picocolors": "^1.1.1", "postcss": "^8.4.47", "postcss-import": "^15.1.0", "postcss-js": "^4.0.1", "postcss-load-config": "^4.0.2 || ^5.0 || ^6.0", "postcss-nested": "^6.2.0", "postcss-selector-parser": "^6.1.2", "resolve": "^1.22.8", "sucrase": "^3.35.0" }, "bin": { "tailwind": "lib/cli.js", "tailwindcss": "lib/cli.js" } }, "sha512-3ofp+LL8E+pK/JuPLPggVAIaEuhvIz4qNcf3nA1Xn2o/7fb7s/TYpHhwGDv1ZU3PkBluUVaF8PyCHcm48cKLWQ=="], "simstudio/@types/node": ["@types/node@20.19.43", "", { "dependencies": { "undici-types": "~6.21.0" } }, "sha512-6oYBAi5ikg4Pl+kGsoYtawUMBT2zZMCvPNF7pVLnHZfd1zf38DRiWn/gT01RYCdUqkv7Fhr+C9ot4/tb+2sVvA=="], diff --git a/docs/workflow-throughput-one-pager.md b/docs/workflow-throughput-one-pager.md index 12a6e960b94..1f87207ed1e 100644 --- a/docs/workflow-throughput-one-pager.md +++ b/docs/workflow-throughput-one-pager.md @@ -1,48 +1,51 @@ -# Sim Workflow Execution: Async, Inline, and Database +# Sim workflow execution throughput and bottlenecks ## Executive summary -The local failure was not that the execute route could not return `202` -quickly. It was: - -> Sim returned `202 Accepted` faster than workflows finished. The admission -> permit was released when the HTTP response returned, while the local async -> execution continued in the same process. Active work accumulated until the -> process became slow and unreachable. - -The execute route has three existing execution modes: - -- `sync`: the request waits for `executeWorkflowCore` and returns JSON. -- `stream`: the request returns an SSE response while `executeWorkflow` runs. -- `async`: the request queues a `workflow-execution` job and returns `202`. - -`async` does not by itself mean that execution leaves the web process. The -existing async backend decides that: - -- The **database backend** calls `shouldExecuteInline() === true` and runs - `executeWorkflowJob` in the same process after inserting the job. -- **Trigger.dev** calls `shouldExecuteInline() === false` and runs the task in - the Trigger.dev worker. - -Throughput is completed workflow executions per second, not `202` responses -per second. This work improves overload behavior and protects the process; it -does not claim a production throughput increase. - -Scope: local database backend, synthetic workflow, one web process. The -control-flow finding is proven by the route and backend code below. The local -load numbers are stability evidence, not a production capacity number. - -## Before and after - -### Before: `202` released capacity too early +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 DatabaseJobQueue - participant E as executeWorkflowJob + participant Q as DatabaseBackend + participant E as InlineExecution C->>R: POST /execute R->>G: tryAdmit() @@ -51,20 +54,20 @@ sequenceDiagram Q-->>R: job accepted R-->>C: 202 Accepted R->>G: release permit - Q->>E: continue execution - Note over E: More requests can enter
while old runs are still active + Q->>E: continue execution in web process + Note over E: More requests enter while old runs remain active E-->>Q: complete ``` -### After: permit represents real in-process work +### After: the permit represents in-process work ```mermaid sequenceDiagram participant C as Caller participant R as ExecuteRoute participant G as AdmissionGate - participant Q as DatabaseJobQueue - participant E as executeWorkflowJob + participant Q as DatabaseBackend + participant E as InlineExecution C->>R: POST /execute R->>G: tryAdmit() @@ -73,7 +76,7 @@ sequenceDiagram R->>Q: enqueue workflow Q-->>R: job accepted R-->>C: 202 Accepted - Q->>E: execute inline + Q->>E: execute inline in web process E-->>Q: complete Q->>G: release permit else capacity full @@ -82,157 +85,73 @@ sequenceDiagram end ``` -The same admission behavior now applies to cookie-backed `async` requests, -which represent the normal authenticated UI path. Session-backed `sync` and -SSE requests retain their existing response/stream lifecycle. - -## Database in the current setup - -The repository proves that the database can be a shared capacity limit, but it -does not prove that database latency caused the local connection-refused -cascade. The evidence is: - -- `packages/db/db.ts` sets `primaryMax=10` for the web role and `primaryMax=5` - for the Trigger.dev role. The same file documents that one run can need - three or more simultaneous connections because of parallel queries and - overlapping logging writes. -- `DatabaseJobQueue.enqueue` inserts into `async_jobs`. The inline runner then - updates the job to `processing`, runs `executeWorkflowJob`, and updates the - job to `completed` or `failed` - ([database backend](../apps/sim/lib/core/async-jobs/backends/database.ts)). -- `executeWorkflowJob` runs preprocessing, `executeWorkflowCore`, logging - finalization, and post-execution work - ([workflow execution](../apps/sim/background/workflow-execution.ts)). -- `executeWorkflowCore` persists block-start and block-complete lifecycle - markers through `LoggingSession`. The durable fallback is a - `workflow_execution_logs` JSONB update - ([execution core](../apps/sim/lib/workflows/executor/execution-core.ts), - [logging session](../apps/sim/lib/logs/execution/logging-session.ts)). -- Completion reads and updates `workflow_execution_logs`, may externalize - execution data, and reconciles usage in a transaction - ([execution logger](../apps/sim/lib/logs/execution/logger.ts)). -- `workflow_execution_logs.execution_id` and the relevant `async_jobs` status - indexes exist in `packages/db/schema.ts`. Indexes establish the intended - access paths; they do not establish query latency under load. - -Therefore, the DB distinction is: - -- With the **database backend**, web-process execution and its `async_jobs`, - execution-log, and usage queries share the web pool. -- With **Trigger.dev**, the web request mostly pays for enqueueing; the - Trigger.dev task still performs execution-log and usage queries, but those - queries use the Trigger.dev pool and the task queue limit. -- With `sync` and SSE, the web request itself waits on the same execution and - database work. With `async` plus the database backend, the request returns - before that work finishes, but the work still occupies the web process and - web pool. - -No repository-only claim should call the DB the measured root cause without -`pg_stat_activity`, pool-wait, query-duration, or equivalent production -measurements. - -## Async and inline in Sim - -These are not opposite endpoint modes: - -1. `sync` calls `executeWorkflowCore` from the execute route; SSE calls - `executeWorkflow` through `createStreamingResponse`. Both keep the - response/stream lifecycle attached to the run. -2. `async` calls `getJobQueue().enqueue('workflow-execution', ...)` and returns - `202`. -3. The database backend starts the job and calls `executeWorkflowJob` in the - web process. The admission ticket must remain held until that call finishes. -4. Trigger.dev starts the task outside the web process. The route releases its - ticket after the enqueue response; Trigger.dev applies - `WORKFLOW_EXECUTION_CONCURRENCY_LIMIT`, whose default is `75`. - -The current branch fixes only the database-backend case: it retains the -admission ticket for in-process `async` execution and releases it in the -inline runner's `finally` block. It does not change the database pool size, -Trigger.dev concurrency, or execution cost. - -## Evidence - -Synthetic workflow: `Start → Function`, with 300ms of deterministic CPU work. -Load profile: async `POST /api/workflows/{id}/execute`, ramp to 8 requests/sec, -then hold for 60 seconds -([Artillery profile](../apps/sim/scripts/load/workflow-concurrency.yml)). -The profile asserts the enqueue response is `202`; completion counts require -the execution-log query below. - -- Before the fix: approximately 218 successful `202` responses, 225 - `ECONNREFUSED` failures, and roughly 2 completed workflows/sec. The process - became unreachable. -- At a more aggressive 30 requests/sec probe: 351 requests returned `202`, - approximately one completion was observed in the window, and more than 280 - executions remained running. -- With `ADMISSION_GATE_MAX_INFLIGHT=5`: 207 `429` responses and repeated - admission-rejection logs confirmed that the admission gate rejects at - capacity. -- After the fix: the 8 requests/sec rerun produced explicit `202`/`429` - backpressure and did not reproduce the previous connection-refused cascade. - One transient health check missed, so this is stability evidence, not a - claim that the process was perfectly healthy every second. - -## Changes in this branch - -- Hold the admission ticket through local inline async execution: - [execute route](../apps/sim/app/api/workflows/[id]/execute/route.ts) -- Apply the gate to session-backed async requests while preserving sync/SSE: - [execute route](../apps/sim/app/api/workflows/[id]/execute/route.ts) -- Lower the default in-process admission limit from 500 to 10: - [admission gate](../apps/sim/lib/core/admission/gate.ts) -- Add a 4 GB Node heap ceiling to the production start script: - [Sim package scripts](../apps/sim/package.json) -- Add workflow execution count/duration telemetry: - [execution metrics](../apps/sim/lib/workflows/executor/execution-metrics.ts) - -The limit of 10 is a safety starting point, not an optimized production -setting. It bounds damage; it does not make execution faster. - -## Other limits found or tested - -### Proven by code, not measured as the local root cause - -- **Execution cost:** CPU/isolated-runtime work determines how long an - admission ticket remains occupied. More expensive workflow executions reduce - completion rate and cause honest `429` backpressure. -- **Web database pool:** database-backend execution shares the web process and - its `primaryMax=10` pool. The sustainable rate depends on the DB work per - execution. -- **Trigger.dev task and DB limits:** the task definition sets a default - concurrency limit of `75`, while the Trigger.dev DB role has a - `primaryMax=5` pool. Queue concurrency and DB connection availability are - separate limits. -- **Sync/SSE process pressure:** session-backed `sync` and SSE requests - intentionally remain outside this async admission change. Their execution - and streaming behavior needs a separate test because they run through the - web process. - -### Tested but inconclusive or not run - -- A 20-request synchronous blast completed successfully; it did not reproduce - an OOM. -- Trigger.dev’s 75/75/50 workflow, webhook, and resume queue contention was - not tested in this session. -- No production workflow bodies were available. The synthetic graph proves the - backlog mechanism, not production workload capacity. - -## Next falsifiable measurement - -Run the same profile through both API-key and authenticated-cookie async -requests. Record: - -1. completed workflow executions/sec from - `workflow_execution_logs.ended_at`; -2. `202` versus `429` responses; -3. running execution count; -4. process health and RSS; -5. `pg_stat_activity` for `sim-app` and `sim-trigger`. - -The following query makes the DB part falsifiable during the run: +### 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 @@ -241,8 +160,33 @@ GROUP BY application_name, state ORDER BY application_name, state; ``` -Then vary deterministic workflow work from 300ms to 2 seconds. A valid next -finding is: completed workflow executions/sec falls and `429` rate rises -while the process stays healthy. A DB finding requires the same run to show -pool saturation or query wait/duration growth, rather than inferring it from -`202` latency alone. +## 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. diff --git a/packages/testing/src/mocks/env-flags.mock.ts b/packages/testing/src/mocks/env-flags.mock.ts index 7f1c552cae9..590a5b9fde6 100644 --- a/packages/testing/src/mocks/env-flags.mock.ts +++ b/packages/testing/src/mocks/env-flags.mock.ts @@ -32,7 +32,6 @@ export const envFlagsMock = { isE2BDocEnabled: false, isOllamaConfigured: false, isAzureConfigured: false, - isOpenRouterConfigured: false, isInvitationsDisabled: false, isPublicApiDisabled: false, isGoogleAuthDisabled: false,