fix(execution): stop the event buffer retaining a run-length backlog

The Redis byte-budget branch in doFlush requeued the rejected batch and
rethrew, skipping the MAX_PENDING_EVENTS trim every other failure path
applies. The backlog then grew for the rest of the run and each retry
re-serialized it, so a wide parallel fan-out drove multi-GB heap growth
and enough event-loop stall to push ALB p90 past 16s.

Drop rejected chunks instead of requeueing, pace retries through the
existing backoff, and split batches that exceed the single-write cap so
an oversized batch can make progress instead of stalling forever.
Terminal status is now writer-scoped, since a concurrent scheduled flush
can be the loop that drains the final chunk.

Record terminal stream meta when the terminal event cannot be buffered,
so reconnecting readers stop polling an active stream until their
deadline. Drop the unused reserve/release budget helpers.

Author: waleedlatif1

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

@@ -41,6 +41,7 @@ const {
   mockHandlePostExecutionPauseState,
   mockHasDurableExecutionOwner,
   mockInitializeExecutionStreamMeta,
+  mockSetExecutionMeta,
   mockReleaseExecutionIdClaim,
   mockReleaseExecutionSlot,
   mockReleaseWorkflowToolExecutionClaim,
@@ -66,6 +67,7 @@ const {
   mockHandlePostExecutionPauseState: vi.fn(),
   mockHasDurableExecutionOwner: vi.fn(),
   mockInitializeExecutionStreamMeta: vi.fn(),
+  mockSetExecutionMeta: vi.fn(),
   mockReleaseExecutionIdClaim: vi.fn(),
   mockReleaseExecutionSlot: vi.fn(),
   mockReleaseWorkflowToolExecutionClaim: vi.fn(),
@@ -128,6 +130,7 @@ vi.mock('@/lib/execution/event-buffer', () => ({
   createExecutionEventWriter: mockCreateExecutionEventWriter,
   flushExecutionStreamReplayBuffer: mockFlushExecutionStreamReplayBuffer,
   initializeExecutionStreamMeta: mockInitializeExecutionStreamMeta,
+  setExecutionMeta: mockSetExecutionMeta,
   LIVE_ONLY_EXECUTION_EVENT_TYPES: new Set(),
 }))
 
@@ -403,6 +406,7 @@ describe('workflow execute async route', () => {
     })
     mockHandlePostExecutionPauseState.mockResolvedValue(undefined)
     mockInitializeExecutionStreamMeta.mockReset().mockResolvedValue(true)
+    mockSetExecutionMeta.mockReset().mockResolvedValue(true)
     mockFlushExecutionStreamReplayBuffer.mockReset().mockResolvedValue(true)
     mockCreateExecutionEventWriter.mockReset().mockReturnValue({
       write: vi.fn(async (event: unknown) => ({ event, eventId: '1' })),
@@ -456,6 +460,31 @@ describe('workflow execute async route', () => {
     expect(body).toContain('execution:completed')
   })
 
+  /**
+   * A terminal event the replay buffer rejected leaves the stream meta on
+   * `active`, so a reconnecting reader polls until its deadline and then errors.
+   * Recording the status directly is the only signal it gets.
+   */
+  it('records terminal stream meta when the replay buffer rejects the terminal event', async () => {
+    mockCreateExecutionEventWriter.mockReturnValue({
+      write: vi.fn(async (event: unknown) => ({ event, eventId: '1' })),
+      writeTerminal: vi.fn(async () => {
+        throw new Error('Execution memory limit exceeded. Reduce payload size and try again.')
+      }),
+      close: vi.fn().mockResolvedValue(undefined),
+    })
+
+    const response = await POST(createBoundCopilotExecutionRequest(), {
+      params: Promise.resolve({ id: 'workflow-1' }),
+    })
+    const body = await response.text()
+
+    expect(response.status).toBe(200)
+    // The live client still receives the terminal event over SSE.
+    expect(body).toContain('execution:completed')
+    expect(mockSetExecutionMeta).toHaveBeenCalledWith('execution-123', { status: 'complete' })
+  })
+
   it('rejects a competing Copilot workflow execution before logging starts', async () => {
     mockClaimWorkflowToolExecution.mockResolvedValueOnce(null)
 

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

@@ -56,6 +56,7 @@ import {
   createExecutionEventWriter,
   flushExecutionStreamReplayBuffer,
   initializeExecutionStreamMeta,
+  setExecutionMeta,
   type TerminalExecutionStreamStatus,
 } from '@/lib/execution/event-buffer'
 import { processInputFileFields } from '@/lib/execution/files'
@@ -1755,6 +1756,7 @@ async function handleExecutePost(
         ) => {
           const isBuffered = !LIVE_ONLY_EXECUTION_EVENT_TYPES.has(event.type)
           let eventToSend = event
+          let terminalBufferWriteFailed = false
           if (isBuffered) {
             try {
               const entry = terminalStatus
@@ -1776,6 +1778,7 @@ async function handleExecutePost(
                 terminal: Boolean(terminalStatus),
                 error: toError(e).message,
               })
+              terminalBufferWriteFailed = Boolean(terminalStatus)
               terminalEventPublished ||= Boolean(terminalStatus)
             }
           }
@@ -1786,6 +1789,24 @@ async function handleExecutePost(
               isStreamClosed = true
             }
           }
+          if (terminalBufferWriteFailed && terminalStatus) {
+            // The terminal event never reached the replay buffer, so a reconnecting
+            // reader would poll an `active` stream until its deadline. Record the
+            // terminal status on the stream meta directly — a plain HSET that bypasses
+            // the byte budget which rejected the event — so the reconnect route sees an
+            // ended run and closes cleanly. Runs after the live enqueue above: Redis is
+            // the most likely reason we are in this branch at all, and a slow best-effort
+            // durability write must never delay the primary delivery path.
+            const metaPersisted = await setExecutionMeta(executionId, {
+              status: terminalStatus,
+            })
+            if (!metaPersisted) {
+              reqLogger.error(
+                'Failed to record terminal execution meta after buffer write failure',
+                { executionId, status: terminalStatus }
+              )
+            }
+          }
         }
 
         try {

apps/sim/lib/execution/event-buffer.test.ts

@@ -368,6 +368,214 @@ describe('execution event buffer', () => {
     expect(persistedEntries).toEqual([])
   })
 
+  /**
+   * Requeueing a batch the budget rejected is what grew `pending` for a whole
+   * run, each retry re-serializing an ever-larger array — the multi-GB heap
+   * growth seen in production. Rejected bytes must be dropped, not retained.
+   */
+  it('drops rejected batches instead of growing a backlog when the Redis budget is exhausted', async () => {
+    mockRedis.incrby.mockResolvedValue(100000)
+    let budgetExhausted = true
+    mockRedis.eval.mockImplementation(async (script: string, ...args: unknown[]) => {
+      if (isFlushScript(script)) {
+        if (budgetExhausted) return [0, 'execution_redis_bytes', 64 * 1024 * 1024]
+        const { zaddArgs } = parseFlushEvalArgs(args)
+        for (let i = 0; i < zaddArgs.length; i += 2) {
+          persistedEntries.push(JSON.parse(zaddArgs[i + 1] as string) as ExecutionEventEntry)
+        }
+        return [1, 1, 0]
+      }
+      return [1, 'ok', 0, 0]
+    })
+
+    const writer = createExecutionEventWriter('exec-1')
+
+    for (let i = 0; i < 2500; i++) {
+      await writer.write(makeEvent(`block-${i}`)).catch(() => {})
+    }
+
+    // Once the budget frees the writer recovers, but only whatever accumulated
+    // since the last rejection — never a run-length backlog.
+    budgetExhausted = false
+    await writer.flush()
+
+    expect(persistedEntries.length).toBeLessThanOrEqual(200)
+  })
+
+  /**
+   * Individual events are capped well below the single-write limit, but a burst
+   * of large ones coalesces into a batch above it. Splitting is the only way the
+   * buffer makes progress: no retry can shrink a batch it keeps whole.
+   */
+  it('splits a batch that exceeds the single-write cap instead of stalling on it', async () => {
+    mockRedis.incrby.mockResolvedValue(100)
+    // Each event stays under the 8MiB per-event cap; two of them do not.
+    const bigPayload = 'x'.repeat(2_500_000)
+
+    const writer = createExecutionEventWriter('exec-1')
+    await writer.write(makeEvent(bigPayload))
+    await writer.write(makeEvent(bigPayload))
+    await writer.flush()
+
+    expect(persistedEntries).toHaveLength(2)
+    expect(
+      mockRedis.eval.mock.calls.filter(([script]) => isFlushScript(script as string))
+    ).toHaveLength(2)
+  })
+
+  it('drops the terminal entry rather than leaving it queued when the budget is exhausted', async () => {
+    mockRedis.incrby.mockResolvedValue(100)
+    let budgetExhausted = true
+    mockRedis.eval.mockImplementation(async (script: string, ...args: unknown[]) => {
+      if (isFlushScript(script)) {
+        if (budgetExhausted) return [0, 'execution_redis_bytes', 64 * 1024 * 1024]
+        const { zaddArgs } = parseFlushEvalArgs(args)
+        for (let i = 0; i < zaddArgs.length; i += 2) {
+          persistedEntries.push(JSON.parse(zaddArgs[i + 1] as string) as ExecutionEventEntry)
+        }
+        return [1, 1, 0]
+      }
+      return [1, 'ok', 0, 0]
+    })
+
+    const writer = createExecutionEventWriter('exec-1')
+
+    await expect(writer.writeTerminal(makeEvent('terminal'), 'complete')).rejects.toThrow(
+      'Execution memory limit exceeded'
+    )
+
+    // The failed terminal write stays surfaced through flush(), but its entry must
+    // not linger in the backlog and reappear once the budget frees up.
+    budgetExhausted = false
+    await writer.flush().catch(() => {})
+
+    expect(persistedEntries).toEqual([])
+  })
+
+  /**
+   * A timer-driven flush carries no terminal status of its own. If it is the
+   * loop that drains the final chunk, the terminal event lands without a status
+   * and readers poll an `active` stream forever — while `writeTerminal` reports
+   * success, so nothing degrades.
+   */
+  it('applies terminal status even when a concurrent scheduled flush drains the final chunk', async () => {
+    mockRedis.incrby.mockResolvedValue(100)
+    const observedTerminalStatuses: string[] = []
+    let releaseFirstFlush: (() => void) | undefined
+    const firstFlushStarted = new Promise<void>((resolveStarted) => {
+      let started = false
+      mockRedis.eval.mockImplementation(async (script: string, ...args: unknown[]) => {
+        if (!isFlushScript(script)) return [1, 'ok', 0, 0]
+        const { terminalStatus, zaddArgs } = parseFlushEvalArgs(args)
+        observedTerminalStatuses.push(terminalStatus)
+        if (!started) {
+          started = true
+          resolveStarted()
+          await new Promise<void>((resolve) => {
+            releaseFirstFlush = resolve
+          })
+        }
+        for (let i = 0; i < zaddArgs.length; i += 2) {
+          persistedEntries.push(JSON.parse(zaddArgs[i + 1] as string) as ExecutionEventEntry)
+        }
+        return [1, 1, 0]
+      })
+    })
+
+    const writer = createExecutionEventWriter('exec-1')
+    await writer.write(makeEvent('first'))
+    await firstFlushStarted
+
+    const terminalWrite = writer.writeTerminal(makeEvent('terminal'), 'complete')
+    // Let writeTerminal's queued body actually enqueue its entry before the
+    // in-flight flush resolves — otherwise the scheduled loop finds nothing left
+    // to drain and the race under test never forms.
+    await new Promise((resolve) => setTimeout(resolve, 5))
+    releaseFirstFlush?.()
+    await terminalWrite
+
+    expect(observedTerminalStatuses).toContain('complete')
+  })
+
+  /**
+   * A terminal publish that threw must not be resurrected. Leaving the status
+   * armed would let the next flush stamp the stream terminal for an event that
+   * was discarded — telling readers the run ended cleanly while the caller was
+   * told it failed.
+   */
+  it('does not stamp terminal status on a later flush after the terminal publish failed', async () => {
+    mockRedis.incrby.mockResolvedValue(100)
+    const observedTerminalStatuses: string[] = []
+    let failNextFlush = false
+    mockRedis.eval.mockImplementation(async (script: string, ...args: unknown[]) => {
+      if (!isFlushScript(script)) return [1, 'ok', 0, 0]
+      if (failNextFlush) throw new Error('redis unavailable')
+      const { terminalStatus, zaddArgs } = parseFlushEvalArgs(args)
+      observedTerminalStatuses.push(terminalStatus)
+      for (let i = 0; i < zaddArgs.length; i += 2) {
+        persistedEntries.push(JSON.parse(zaddArgs[i + 1] as string) as ExecutionEventEntry)
+      }
+      return [1, 1, 0]
+    })
+
+    const writer = createExecutionEventWriter('exec-1')
+    await writer.write(makeEvent('a'))
+
+    failNextFlush = true
+    await expect(writer.writeTerminal(makeEvent('terminal'), 'complete')).rejects.toThrow()
+
+    // flush() still surfaces the earlier terminal failure; what matters is that
+    // the events it drains are not stamped terminal.
+    failNextFlush = false
+    await writer.flush().catch(() => {})
+
+    expect(observedTerminalStatuses).toEqual([''])
+    expect(
+      persistedEntries.map((entry) => (entry.event.data as { blockId: string }).blockId)
+    ).toEqual(['a'])
+  })
+
+  /**
+   * A budget rejection must not colour a later, unrelated failure: reporting a
+   * Redis outage as "reduce payload size" sends the user after the wrong thing.
+   */
+  it('reports the generic failure, not a stale budget rejection, on the terminal path', async () => {
+    mockRedis.incrby.mockResolvedValue(100)
+    let mode: 'budget' | 'outage' = 'budget'
+    mockRedis.eval.mockImplementation(async (script: string) => {
+      if (!isFlushScript(script)) return [1, 'ok', 0, 0]
+      if (mode === 'budget') return [0, 'execution_redis_bytes', 64 * 1024 * 1024]
+      throw new Error('redis unavailable')
+    })
+
+    const writer = createExecutionEventWriter('exec-1')
+    for (let i = 0; i < 200; i++) {
+      await writer.write(makeEvent(`block-${i}`)).catch(() => {})
+    }
+
+    mode = 'outage'
+    await expect(writer.writeTerminal(makeEvent('terminal'), 'complete')).rejects.toThrow(
+      'Failed to flush terminal execution event'
+    )
+  })
+
+  it('settles a scheduled flush that hits the budget instead of rejecting later callers', async () => {
+    mockRedis.incrby.mockResolvedValue(100)
+    mockRedis.eval.mockImplementation(async (script: string) => {
+      if (isFlushScript(script)) {
+        return [0, 'execution_redis_bytes', 64 * 1024 * 1024]
+      }
+      return [1, 'ok', 0, 0]
+    })
+
+    const writer = createExecutionEventWriter('exec-1')
+    await writer.write(makeEvent('a'))
+
+    await new Promise((resolve) => setTimeout(resolve, 60))
+
+    await expect(writer.flush()).resolves.toBeUndefined()
+  })
+
   it('preserves requested UserFile base64 when buffering terminal events', async () => {
     mockRedis.incrby.mockResolvedValue(100)
     const base64 = Buffer.from('hello').toString('base64')