Skip to content

Commit 7c1e727

Browse files
icecrasher321icecrasher321
committed
feat(timeouts): execution timeout limits
1 parent a9b7d75 commit 7c1e727

File tree

29 files changed

+390
-127
lines changed

29 files changed

+390
-127
lines changed

‎apps/sim/app/api/mcp/serve/[serverId]/route.ts‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@ import { and, eq } from 'drizzle-orm'
2121
import { type NextRequest, NextResponse } from 'next/server'
2222
import { checkHybridAuth } from '@/lib/auth/hybrid'
2323
import { generateInternalToken } from '@/lib/auth/internal'
24+
import { getMaxExecutionTimeout } from '@/lib/core/execution-limits'
2425
import { getBaseUrl } from '@/lib/core/utils/urls'
2526

2627
const logger = createLogger('WorkflowMcpServeAPI')
@@ -264,7 +265,7 @@ async function handleToolsCall(
264265
method: 'POST',
265266
headers,
266267
body: JSON.stringify({ input: params.arguments || {}, triggerType: 'mcp' }),
267-
signal: AbortSignal.timeout(600000), // 10 minute timeout
268+
signal: AbortSignal.timeout(getMaxExecutionTimeout()),
268269
})
269270

270271
const executeResult = await response.json()

‎apps/sim/app/api/mcp/tools/execute/route.ts‎

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,14 @@
11
import { createLogger } from '@sim/logger'
22
import type { NextRequest } from 'next/server'
3+
import { getHighestPrioritySubscription } from '@/lib/billing/core/plan'
4+
import { getExecutionTimeout } from '@/lib/core/execution-limits'
35
import { getParsedBody, withMcpAuth } from '@/lib/mcp/middleware'
46
import { mcpService } from '@/lib/mcp/service'
57
import type { McpTool, McpToolCall, McpToolResult } from '@/lib/mcp/types'
68
import {
79
categorizeError,
810
createMcpErrorResponse,
911
createMcpSuccessResponse,
10-
MCP_CONSTANTS,
1112
validateStringParam,
1213
} from '@/lib/mcp/utils'
1314

@@ -171,13 +172,13 @@ export const POST = withMcpAuth('read')(
171172
arguments: args,
172173
}
173174

175+
const userSubscription = await getHighestPrioritySubscription(userId)
176+
const executionTimeout = getExecutionTimeout(userSubscription?.plan, 'sync')
177+
174178
const result = await Promise.race([
175179
mcpService.executeTool(userId, serverId, toolCall, workspaceId),
176180
new Promise<never>((_, reject) =>
177-
setTimeout(
178-
() => reject(new Error('Tool execution timeout')),
179-
MCP_CONSTANTS.EXECUTION_TIMEOUT
180-
)
181+
setTimeout(() => reject(new Error('Tool execution timeout')), executionTimeout)
181182
),
182183
])
183184

‎apps/sim/app/api/workflows/[id]/execute-from-block/route.ts‎

Lines changed: 56 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ import { type NextRequest, NextResponse } from 'next/server'
33
import { v4 as uuidv4 } from 'uuid'
44
import { z } from 'zod'
55
import { checkHybridAuth } from '@/lib/auth/hybrid'
6+
import { getTimeoutErrorMessage, isTimeoutError } from '@/lib/core/execution-limits'
67
import { generateRequestId } from '@/lib/core/utils/request'
78
import { SSE_HEADERS } from '@/lib/core/utils/sse'
89
import { markExecutionCancelled } from '@/lib/execution/cancellation'
@@ -116,6 +117,16 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
116117
const loggingSession = new LoggingSession(workflowId, executionId, 'manual', requestId)
117118
const abortController = new AbortController()
118119
let isStreamClosed = false
120+
let isTimedOut = false
121+
122+
const syncTimeout = preprocessResult.executionTimeout?.sync
123+
let timeoutId: NodeJS.Timeout | undefined
124+
if (syncTimeout) {
125+
timeoutId = setTimeout(() => {
126+
isTimedOut = true
127+
abortController.abort()
128+
}, syncTimeout)
129+
}
119130

120131
const stream = new ReadableStream<Uint8Array>({
121132
async start(controller) {
@@ -167,13 +178,33 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
167178
})
168179

169180
if (result.status === 'cancelled') {
170-
sendEvent({
171-
type: 'execution:cancelled',
172-
timestamp: new Date().toISOString(),
173-
executionId,
174-
workflowId,
175-
data: { duration: result.metadata?.duration || 0 },
176-
})
181+
if (isTimedOut && syncTimeout) {
182+
const timeoutErrorMessage = getTimeoutErrorMessage(null, syncTimeout)
183+
logger.info(`[${requestId}] Run-from-block execution timed out`, {
184+
timeoutMs: syncTimeout,
185+
})
186+
187+
await loggingSession.markAsFailed(timeoutErrorMessage)
188+
189+
sendEvent({
190+
type: 'execution:error',
191+
timestamp: new Date().toISOString(),
192+
executionId,
193+
workflowId,
194+
data: {
195+
error: timeoutErrorMessage,
196+
duration: result.metadata?.duration || 0,
197+
},
198+
})
199+
} else {
200+
sendEvent({
201+
type: 'execution:cancelled',
202+
timestamp: new Date().toISOString(),
203+
executionId,
204+
workflowId,
205+
data: { duration: result.metadata?.duration || 0 },
206+
})
207+
}
177208
} else {
178209
sendEvent({
179210
type: 'execution:completed',
@@ -190,11 +221,25 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
190221
})
191222
}
192223
} catch (error: unknown) {
193-
const errorMessage = error instanceof Error ? error.message : 'Unknown error'
194-
logger.error(`[${requestId}] Run-from-block execution failed: ${errorMessage}`)
224+
const isTimeout = isTimeoutError(error) || isTimedOut
225+
const errorMessage = isTimeout
226+
? getTimeoutErrorMessage(error, syncTimeout)
227+
: error instanceof Error
228+
? error.message
229+
: 'Unknown error'
230+
231+
logger.error(`[${requestId}] Run-from-block execution failed: ${errorMessage}`, {
232+
isTimeout,
233+
})
195234

196235
const executionResult = hasExecutionResult(error) ? error.executionResult : undefined
197236

237+
await loggingSession.safeCompleteWithError({
238+
totalDurationMs: executionResult?.metadata?.duration,
239+
error: { message: errorMessage },
240+
traceSpans: executionResult?.logs as any,
241+
})
242+
198243
sendEvent({
199244
type: 'execution:error',
200245
timestamp: new Date().toISOString(),
@@ -206,6 +251,7 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
206251
},
207252
})
208253
} finally {
254+
if (timeoutId) clearTimeout(timeoutId)
209255
if (!isStreamClosed) {
210256
try {
211257
controller.enqueue(new TextEncoder().encode('data: [DONE]\n\n'))
@@ -216,6 +262,7 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
216262
},
217263
cancel() {
218264
isStreamClosed = true
265+
if (timeoutId) clearTimeout(timeoutId)
219266
abortController.abort()
220267
markExecutionCancelled(executionId).catch(() => {})
221268
},

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

Lines changed: 74 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import { validate as uuidValidate, v4 as uuidv4 } from 'uuid'
55
import { z } from 'zod'
66
import { checkHybridAuth } from '@/lib/auth/hybrid'
77
import { isTriggerDevEnabled } from '@/lib/core/config/feature-flags'
8+
import { getTimeoutErrorMessage, isTimeoutError } from '@/lib/core/execution-limits'
89
import { generateRequestId } from '@/lib/core/utils/request'
910
import { SSE_HEADERS } from '@/lib/core/utils/sse'
1011
import { getBaseUrl } from '@/lib/core/utils/urls'
@@ -120,10 +121,6 @@ type AsyncExecutionParams = {
120121
triggerType: CoreTriggerType
121122
}
122123

123-
/**
124-
* Handles async workflow execution by queueing a background job.
125-
* Returns immediately with a 202 Accepted response containing the job ID.
126-
*/
127124
async function handleAsyncExecution(params: AsyncExecutionParams): Promise<NextResponse> {
128125
const { requestId, workflowId, userId, input, triggerType } = params
129126

@@ -405,6 +402,7 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
405402

406403
if (!enableSSE) {
407404
logger.info(`[${requestId}] Using non-SSE execution (direct JSON response)`)
405+
const syncTimeout = preprocessResult.executionTimeout?.sync
408406
try {
409407
const metadata: ExecutionMetadata = {
410408
requestId,
@@ -438,6 +436,7 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
438436
includeFileBase64,
439437
base64MaxBytes,
440438
stopAfterBlockId,
439+
abortSignal: syncTimeout ? AbortSignal.timeout(syncTimeout) : undefined,
441440
})
442441

443442
const outputWithBase64 = includeFileBase64
@@ -473,11 +472,23 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
473472

474473
return NextResponse.json(filteredResult)
475474
} catch (error: unknown) {
476-
const errorMessage = error instanceof Error ? error.message : 'Unknown error'
477-
logger.error(`[${requestId}] Non-SSE execution failed: ${errorMessage}`)
475+
const isTimeout = isTimeoutError(error)
476+
const errorMessage = isTimeout
477+
? getTimeoutErrorMessage(error, syncTimeout)
478+
: error instanceof Error
479+
? error.message
480+
: 'Unknown error'
481+
482+
logger.error(`[${requestId}] Non-SSE execution failed: ${errorMessage}`, { isTimeout })
478483

479484
const executionResult = hasExecutionResult(error) ? error.executionResult : undefined
480485

486+
await loggingSession.safeCompleteWithError({
487+
totalDurationMs: executionResult?.metadata?.duration,
488+
error: { message: errorMessage },
489+
traceSpans: executionResult?.logs as any,
490+
})
491+
481492
return NextResponse.json(
482493
{
483494
success: false,
@@ -491,7 +502,7 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
491502
}
492503
: undefined,
493504
},
494-
{ status: 500 }
505+
{ status: isTimeout ? 408 : 500 }
495506
)
496507
}
497508
}
@@ -537,6 +548,16 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
537548
const encoder = new TextEncoder()
538549
const abortController = new AbortController()
539550
let isStreamClosed = false
551+
let isTimedOut = false
552+
553+
const syncTimeout = preprocessResult.executionTimeout?.sync
554+
let timeoutId: NodeJS.Timeout | undefined
555+
if (syncTimeout) {
556+
timeoutId = setTimeout(() => {
557+
isTimedOut = true
558+
abortController.abort()
559+
}, syncTimeout)
560+
}
540561

541562
const stream = new ReadableStream<Uint8Array>({
542563
async start(controller) {
@@ -763,16 +784,35 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
763784
}
764785

765786
if (result.status === 'cancelled') {
766-
logger.info(`[${requestId}] Workflow execution was cancelled`)
767-
sendEvent({
768-
type: 'execution:cancelled',
769-
timestamp: new Date().toISOString(),
770-
executionId,
771-
workflowId,
772-
data: {
773-
duration: result.metadata?.duration || 0,
774-
},
775-
})
787+
if (isTimedOut && syncTimeout) {
788+
const timeoutErrorMessage = getTimeoutErrorMessage(null, syncTimeout)
789+
logger.info(`[${requestId}] Workflow execution timed out`, { timeoutMs: syncTimeout })
790+
791+
await loggingSession.markAsFailed(timeoutErrorMessage)
792+
793+
sendEvent({
794+
type: 'execution:error',
795+
timestamp: new Date().toISOString(),
796+
executionId,
797+
workflowId,
798+
data: {
799+
error: timeoutErrorMessage,
800+
duration: result.metadata?.duration || 0,
801+
},
802+
})
803+
} else {
804+
logger.info(`[${requestId}] Workflow execution was cancelled`)
805+
806+
sendEvent({
807+
type: 'execution:cancelled',
808+
timestamp: new Date().toISOString(),
809+
executionId,
810+
workflowId,
811+
data: {
812+
duration: result.metadata?.duration || 0,
813+
},
814+
})
815+
}
776816
return
777817
}
778818

@@ -799,11 +839,23 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
799839
// Cleanup base64 cache for this execution
800840
await cleanupExecutionBase64Cache(executionId)
801841
} catch (error: unknown) {
802-
const errorMessage = error instanceof Error ? error.message : 'Unknown error'
803-
logger.error(`[${requestId}] SSE execution failed: ${errorMessage}`)
842+
const isTimeout = isTimeoutError(error) || isTimedOut
843+
const errorMessage = isTimeout
844+
? getTimeoutErrorMessage(error, syncTimeout)
845+
: error instanceof Error
846+
? error.message
847+
: 'Unknown error'
848+
849+
logger.error(`[${requestId}] SSE execution failed: ${errorMessage}`, { isTimeout })
804850

805851
const executionResult = hasExecutionResult(error) ? error.executionResult : undefined
806852

853+
await loggingSession.safeCompleteWithError({
854+
totalDurationMs: executionResult?.metadata?.duration,
855+
error: { message: errorMessage },
856+
traceSpans: executionResult?.logs as any,
857+
})
858+
807859
sendEvent({
808860
type: 'execution:error',
809861
timestamp: new Date().toISOString(),
@@ -815,18 +867,18 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
815867
},
816868
})
817869
} finally {
870+
if (timeoutId) clearTimeout(timeoutId)
818871
if (!isStreamClosed) {
819872
try {
820873
controller.enqueue(encoder.encode('data: [DONE]\n\n'))
821874
controller.close()
822-
} catch {
823-
// Stream already closed - nothing to do
824-
}
875+
} catch {}
825876
}
826877
}
827878
},
828879
cancel() {
829880
isStreamClosed = true
881+
if (timeoutId) clearTimeout(timeoutId)
830882
logger.info(`[${requestId}] Client aborted SSE stream, signalling cancellation`)
831883
abortController.abort()
832884
markExecutionCancelled(executionId).catch(() => {})

0 commit comments

Comments
 (0)