From fcdcaed00dfc7dc922a6ca520cdc2e7aac1af79f Mon Sep 17 00:00:00 2001 From: Waleed Date: Wed, 4 Mar 2026 17:46:20 -0800 Subject: [PATCH] fix(memory): add Bun.gc, stream cancellation, and unconsumed fetch drains (#3416) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(memory): add Bun.gc, stream cancellation, and unconsumed fetch drains * fix(memory): await reader.cancel() and use non-blocking Bun.gc * fix(memory): update Bun.gc comment to match non-blocking call * fix(memory): use response.body.cancel() instead of response.text() for drains * fix(executor): flush TextDecoder after streaming loop for multi-byte chars * fix(memory): use text() drain for SecureFetchResponse which lacks body property * fix(chat): prevent premature isExecuting=false from killing chat stream The onExecutionCompleted/Error/Cancelled callbacks were setting isExecuting=false as soon as the server-side SSE stream completed. For chat executions, this triggered a useEffect in chat.tsx that cancelled the client-side stream reader before it finished consuming buffered data — causing empty or partial chat responses. Skip the isExecuting=false in these callbacks for chat executions since the chat's own finally block handles cleanup after the stream is fully consumed. * fix(chat): remove useEffect anti-pattern that killed chat stream on state change The effect reacted to isExecuting becoming false to clean up streams, but this is an anti-pattern per React guidelines — using state changes as a proxy for events. All cleanup cases are already handled by proper event paths: stream done (processStreamingResponse), user cancel (handleStopStreaming), component unmount (cleanup effect), and abort/error (catch block). * fix(servicenow): remove invalid string comparison on numeric offset param * upgrade turborepo --- apps/sim/app/api/tools/stt/route.ts | 1 + apps/sim/app/api/tools/textract/parse/route.ts | 1 + apps/sim/app/api/tools/tts/route.ts | 1 + apps/sim/app/api/tools/vision/analyze/route.ts | 1 + apps/sim/app/api/workflows/[id]/execute/route.ts | 2 +- .../w/[workflowId]/components/chat/chat.tsx | 11 ----------- .../[workflowId]/hooks/use-workflow-execution.ts | 13 +++++++++---- apps/sim/executor/execution/block-executor.ts | 12 +++++++++--- apps/sim/lib/monitoring/memory-telemetry.ts | 10 ++++++++++ apps/sim/lib/webhooks/gmail-polling-service.ts | 1 + apps/sim/next.config.ts | 1 + apps/sim/tools/gmail/utils.ts | 1 + apps/sim/tools/servicenow/read_record.ts | 2 +- bun.lock | 16 ++++++++-------- package.json | 2 +- 15 files changed, 46 insertions(+), 29 deletions(-) diff --git a/apps/sim/app/api/tools/stt/route.ts b/apps/sim/app/api/tools/stt/route.ts index 2d18e19935..aaacadd5a6 100644 --- a/apps/sim/app/api/tools/stt/route.ts +++ b/apps/sim/app/api/tools/stt/route.ts @@ -150,6 +150,7 @@ export async function POST(request: NextRequest) { method: 'GET', }) if (!response.ok) { + await response.text().catch(() => {}) throw new Error(`Failed to download audio from URL: ${response.statusText}`) } diff --git a/apps/sim/app/api/tools/textract/parse/route.ts b/apps/sim/app/api/tools/textract/parse/route.ts index ca1e4a540c..c1986a4de9 100644 --- a/apps/sim/app/api/tools/textract/parse/route.ts +++ b/apps/sim/app/api/tools/textract/parse/route.ts @@ -135,6 +135,7 @@ async function fetchDocumentBytes(url: string): Promise<{ bytes: string; content method: 'GET', }) if (!response.ok) { + await response.text().catch(() => {}) throw new Error(`Failed to fetch document: ${response.statusText}`) } diff --git a/apps/sim/app/api/tools/tts/route.ts b/apps/sim/app/api/tools/tts/route.ts index f8e1065540..153925c407 100644 --- a/apps/sim/app/api/tools/tts/route.ts +++ b/apps/sim/app/api/tools/tts/route.ts @@ -65,6 +65,7 @@ export async function POST(request: NextRequest) { }) if (!response.ok) { + await response.body?.cancel().catch(() => {}) logger.error(`Failed to generate TTS: ${response.status} ${response.statusText}`) return NextResponse.json( { error: `Failed to generate TTS: ${response.status} ${response.statusText}` }, diff --git a/apps/sim/app/api/tools/vision/analyze/route.ts b/apps/sim/app/api/tools/vision/analyze/route.ts index 684094b2bd..08071e9f63 100644 --- a/apps/sim/app/api/tools/vision/analyze/route.ts +++ b/apps/sim/app/api/tools/vision/analyze/route.ts @@ -184,6 +184,7 @@ export async function POST(request: NextRequest) { method: 'GET', }) if (!response.ok) { + await response.text().catch(() => {}) return NextResponse.json( { success: false, error: 'Failed to fetch image for Gemini' }, { status: 400 } diff --git a/apps/sim/app/api/workflows/[id]/execute/route.ts b/apps/sim/app/api/workflows/[id]/execute/route.ts index ea7c792bb1..5207f77c01 100644 --- a/apps/sim/app/api/workflows/[id]/execute/route.ts +++ b/apps/sim/app/api/workflows/[id]/execute/route.ts @@ -964,7 +964,7 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id: logger.error(`[${requestId}] Error streaming block content:`, error) } finally { try { - reader.releaseLock() + await reader.cancel().catch(() => {}) } catch {} } } diff --git a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/chat/chat.tsx b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/chat/chat.tsx index a199d13707..35ec02c352 100644 --- a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/chat/chat.tsx +++ b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/chat/chat.tsx @@ -501,17 +501,6 @@ export function Chat() { } }, []) - useEffect(() => { - if (!isExecuting && isStreaming) { - const lastMessage = workflowMessages[workflowMessages.length - 1] - if (lastMessage?.isStreaming) { - streamReaderRef.current?.cancel() - streamReaderRef.current = null - finalizeMessageStream(lastMessage.id) - } - } - }, [isExecuting, isStreaming, workflowMessages, finalizeMessageStream]) - const handleStopStreaming = useCallback(() => { streamReaderRef.current?.cancel() streamReaderRef.current = null diff --git a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/hooks/use-workflow-execution.ts b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/hooks/use-workflow-execution.ts index 3624f455e2..21f46be3c7 100644 --- a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/hooks/use-workflow-execution.ts +++ b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/hooks/use-workflow-execution.ts @@ -1495,8 +1495,13 @@ export function useWorkflowExecution() { : null if (activeWorkflowId && !workflowExecState?.isDebugging) { setExecutionResult(executionResult) - setIsExecuting(activeWorkflowId, false) - setActiveBlocks(activeWorkflowId, new Set()) + // For chat executions, don't set isExecuting=false here — the chat's + // client-side stream wrapper still has buffered data to deliver. + // The chat's finally block handles cleanup after the stream is fully consumed. + if (!isExecutingFromChat) { + setIsExecuting(activeWorkflowId, false) + setActiveBlocks(activeWorkflowId, new Set()) + } setTimeout(() => { queryClient.invalidateQueries({ queryKey: subscriptionKeys.all }) }, 1000) @@ -1536,7 +1541,7 @@ export function useWorkflowExecution() { isPreExecutionError, }) - if (activeWorkflowId) { + if (activeWorkflowId && !isExecutingFromChat) { setIsExecuting(activeWorkflowId, false) setIsDebugging(activeWorkflowId, false) setActiveBlocks(activeWorkflowId, new Set()) @@ -1562,7 +1567,7 @@ export function useWorkflowExecution() { durationMs: data?.duration, }) - if (activeWorkflowId) { + if (activeWorkflowId && !isExecutingFromChat) { setIsExecuting(activeWorkflowId, false) setIsDebugging(activeWorkflowId, false) setActiveBlocks(activeWorkflowId, new Set()) diff --git a/apps/sim/executor/execution/block-executor.ts b/apps/sim/executor/execution/block-executor.ts index 46b2e554d2..9c54d2bd99 100644 --- a/apps/sim/executor/execution/block-executor.ts +++ b/apps/sim/executor/execution/block-executor.ts @@ -618,6 +618,8 @@ export class BlockExecutor { await ctx.onStream?.(clientStreamingExec) } catch (error) { logger.error('Error in onStream callback', { blockId, error }) + // Cancel the client stream to release the tee'd buffer + await processedClientStream.cancel().catch(() => {}) } })() @@ -646,6 +648,7 @@ export class BlockExecutor { }) } catch (error) { logger.error('Error in onStream callback', { blockId, error }) + await processedStream.cancel().catch(() => {}) } } @@ -657,22 +660,25 @@ export class BlockExecutor { ): Promise { const reader = stream.getReader() const decoder = new TextDecoder() - let fullContent = '' + const chunks: string[] = [] try { while (true) { const { done, value } = await reader.read() if (done) break - fullContent += decoder.decode(value, { stream: true }) + chunks.push(decoder.decode(value, { stream: true })) } + const tail = decoder.decode() + if (tail) chunks.push(tail) } catch (error) { logger.error('Error reading executor stream for block', { blockId, error }) } finally { try { - reader.releaseLock() + await reader.cancel().catch(() => {}) } catch {} } + const fullContent = chunks.join('') if (!fullContent) { return } diff --git a/apps/sim/lib/monitoring/memory-telemetry.ts b/apps/sim/lib/monitoring/memory-telemetry.ts index d9383c6411..ef7867ce84 100644 --- a/apps/sim/lib/monitoring/memory-telemetry.ts +++ b/apps/sim/lib/monitoring/memory-telemetry.ts @@ -23,6 +23,16 @@ export function startMemoryTelemetry(intervalMs = 60_000) { started = true const timer = setInterval(() => { + // Trigger opportunistic (non-blocking) garbage collection if running on Bun. + // This signals JSC GC + mimalloc page purge without blocking the event loop, + // helping reclaim RSS that mimalloc otherwise retains under sustained load. + const bunGlobal = (globalThis as Record).Bun as + | { gc?: (force: boolean) => void } + | undefined + if (typeof bunGlobal?.gc === 'function') { + bunGlobal.gc(false) + } + const mem = process.memoryUsage() const heap = v8.getHeapStatistics() diff --git a/apps/sim/lib/webhooks/gmail-polling-service.ts b/apps/sim/lib/webhooks/gmail-polling-service.ts index 3415dce362..5d4af90634 100644 --- a/apps/sim/lib/webhooks/gmail-polling-service.ts +++ b/apps/sim/lib/webhooks/gmail-polling-service.ts @@ -759,6 +759,7 @@ async function markEmailAsRead(accessToken: string, messageId: string) { }) if (!response.ok) { + await response.body?.cancel().catch(() => {}) throw new Error( `Failed to mark email ${messageId} as read: ${response.status} ${response.statusText}` ) diff --git a/apps/sim/next.config.ts b/apps/sim/next.config.ts index 583c4bc893..f453dc52d9 100644 --- a/apps/sim/next.config.ts +++ b/apps/sim/next.config.ts @@ -95,6 +95,7 @@ const nextConfig: NextConfig = { optimizeCss: true, turbopackSourceMaps: false, turbopackFileSystemCacheForDev: true, + preloadEntriesOnStart: false, }, ...(isDev && { allowedDevOrigins: [ diff --git a/apps/sim/tools/gmail/utils.ts b/apps/sim/tools/gmail/utils.ts index 7da950b7dd..4d856db1d5 100644 --- a/apps/sim/tools/gmail/utils.ts +++ b/apps/sim/tools/gmail/utils.ts @@ -239,6 +239,7 @@ export async function downloadAttachments( ) if (!attachmentResponse.ok) { + await attachmentResponse.body?.cancel().catch(() => {}) continue } diff --git a/apps/sim/tools/servicenow/read_record.ts b/apps/sim/tools/servicenow/read_record.ts index bc44622c7c..347c7c0c6c 100644 --- a/apps/sim/tools/servicenow/read_record.ts +++ b/apps/sim/tools/servicenow/read_record.ts @@ -109,7 +109,7 @@ export const readRecordTool: ToolConfig