fix(execution): offload buffered event values under budget pressure (#6231)

* fix(execution): offload buffered event values under budget pressure

An execution buffers EVENT_LIMIT events inside a per-execution byte budget, so
a full ring only fits if events average under budget/EVENT_LIMIT. Values were
only offloaded to object storage at the shared 8 MiB cap, far above that, so a
run emitting large block outputs exhausted its budget within a few dozen events
and stayed pinned at its ceiling for the rest of its life.

Applying that ceiling to every run would be worse than the problem: the SSE
stream carries the compacted event and the terminal renders a ref only as a
preview, so ordinary block outputs would stop being readable live, and every
value would cost an object-storage write on the hot path. Engage the tight
ceiling only once a run has actually buffered past half its budget. A short run
keeps full-fidelity output and pays nothing; a runaway one stops accumulating.

Both bounds derive from the existing budget rather than being asserted, and
preserved UserFile base64 is exempt — it is an explicit request for inline
delivery, already bounded by its own cap and the strip-and-recompact fallback.

Also stop a failed resume-path buffer write from failing the run: it was
awaited bare, so the rejection propagated into the executor callback and failed
work that had already completed. The buffer only backs reconnect replay, so
degrade to live-only delivery the way the execute route does.

* fix(execution): measure pressure at write time and keep terminal status last

Pressure was read from bytes counted once a flush succeeded, but a burst is
compacted long before the scheduled flush runs — so the very batch that
exhausts the budget went through at the loose ceiling and was dropped instead
of offloaded. Count bytes as each event is compacted.

Separately, the terminal-alone retry stamped terminal status while entries
queued ahead of it were still unwritten. Terminal status is the reader's
end-of-run signal: a reconnecting client drains what is in Redis and closes, so
those entries were stranded behind a stream it had already finished with. Drain
the backlog first, then publish the terminal event.

* fix(execution): do not lose the backlog or publish terminal status early

Draining the backlog ahead of the terminal event left the terminal status armed,
so whichever chunk emptied the queue stamped the run complete before its
terminal event was written — the inverse of the ordering the drain was added to
guarantee. Disarm the status for the drain and restore it afterwards.

The drain's result was also discarded: a transient Redis failure requeues its
batch, and the unconditional reassignment that followed dropped those events
even though the budget never rejected them. Keep whatever could not be
persisted, and publish the terminal event alone only once nothing earlier is
still queued — failing otherwise lets the caller degrade, which records the
status without claiming the missing events arrived.

Leave eventId unset on a failed resume-path write. Assigning 0 was persisted by
clients as a reconnect cursor and rewound them to the start of the run.

* fix(execution): keep an event in the buffer when a pressure offload fails

Durable compaction runs before an event is queued, so a storage or metadata
failure dropped it from replay entirely — a reconnecting client would never see
it, even though the live path carried on. Offloading under pressure is only an
optimization that keeps a heavy run from exhausting its budget, so when the
value cannot be persisted, fall back to buffering it inline: exactly what the
run would have done before pressure engaged.

Author: waleedlatif1

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

@@ -5,6 +5,7 @@ import { redisConfigMockFns, resetRedisConfigMock } from '@sim/testing'
 import { sleep } from '@sim/utils/helpers'
 import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest'
 import type { ExecutionEventEntry } from '@/lib/execution/event-buffer'
+import { LARGE_VALUE_REF_MARKER } from '@/lib/execution/payloads/large-value-ref'
 import type { ExecutionEvent } from '@/lib/workflows/executor/execution-events'
 
 const { mockRedis, persistedEntries } = vi.hoisted(() => {
@@ -622,6 +623,188 @@ describe('execution event buffer', () => {
     await expect(writer.flush()).resolves.toBeUndefined()
   })
 
+  /**
+   * A short run must keep full-fidelity output: the SSE stream carries the
+   * compacted event, and the terminal renders a ref only as a preview, so
+   * offloading ordinary block outputs would make them unreadable live.
+   */
+  it('keeps values inline while the execution is below the offload pressure mark', async () => {
+    mockRedis.incrby.mockResolvedValue(100)
+    const payload = 'x'.repeat(512 * 1024)
+
+    const writer = createExecutionEventWriter('exec-1', {
+      workspaceId: 'ws-1',
+      workflowId: 'wf-1',
+    })
+    await writer.write(makeEvent(payload))
+    await writer.flush()
+
+    const persisted = JSON.stringify(persistedEntries[0])
+    expect(persisted).toContain(payload)
+    expect(persisted).not.toContain(LARGE_VALUE_REF_MARKER)
+  })
+
+  /**
+   * Once a run has buffered its way into the danger zone the tight ceiling
+   * engages, so it stops accumulating against its budget instead of pinning
+   * itself at the ceiling for the rest of its life.
+   */
+  it('offloads values once the execution crosses the offload pressure mark', async () => {
+    mockRedis.incrby.mockResolvedValue(100000)
+    const payload = 'x'.repeat(2 * 1024 * 1024)
+
+    const writer = createExecutionEventWriter('exec-1', {
+      workspaceId: 'ws-1',
+      workflowId: 'wf-1',
+    })
+    // Push past half the per-execution budget so the next write is under pressure.
+    for (let i = 0; i < 17; i++) {
+      await writer.write(makeEvent(payload))
+      await writer.flush()
+    }
+    persistedEntries.length = 0
+    await writer.write(makeEvent(payload))
+    await writer.flush()
+
+    const persisted = JSON.stringify(persistedEntries[0])
+    expect(persisted).toContain(LARGE_VALUE_REF_MARKER)
+    expect(persisted).not.toContain(payload)
+  })
+
+  /**
+   * Terminal status is the reader's end-of-run signal: once it lands, a
+   * reconnecting client drains what is in Redis and closes. Stamping it while
+   * lower event ids are still queued strands those events behind a stream the
+   * reader has already finished with.
+   *
+   * Needs a backlog past the single-write cap so chunking leaves a remainder
+   * behind the terminal entry — the only shape where that ordering can invert.
+   */
+  it('does not stamp terminal status while earlier events are still queued', async () => {
+    mockRedis.incrby.mockResolvedValue(100000)
+    const idsAtStamp: number[] = []
+    mockRedis.eval.mockImplementation(async (script: string, ...args: unknown[]) => {
+      if (!isFlushScript(script)) return [1, 'ok', 0, 0]
+      const { terminalStatus, zaddArgs } = parseFlushEvalArgs(args)
+      // Reject any multi-entry batch, forcing the terminal-alone retry path.
+      if (zaddArgs.length > 2) return [0, 'execution_redis_bytes', 64 * 1024 * 1024]
+      for (let i = 0; i < zaddArgs.length; i += 2) {
+        persistedEntries.push(JSON.parse(zaddArgs[i + 1] as string) as ExecutionEventEntry)
+      }
+      if (terminalStatus && idsAtStamp.length === 0) {
+        idsAtStamp.push(...persistedEntries.map((e) => e.eventId))
+      }
+      return [1, 1, 0]
+    })
+
+    // ~3MB per event, so three of them exceed the 8MiB single-write cap and the
+    // chunk boundary leaves a remainder queued behind the terminal entry.
+    const payload = 'x'.repeat(1_500_000)
+    const writer = createExecutionEventWriter('exec-1', {
+      workspaceId: 'ws-1',
+      workflowId: 'wf-1',
+    })
+    for (let i = 0; i < 3; i++) {
+      await writer.write(makeEvent(payload)).catch(() => {})
+    }
+    await writer.writeTerminal(makeEvent('terminal'), 'complete').catch(() => {})
+    await writer.close().catch(() => {})
+
+    const terminalId = Math.max(...persistedEntries.map((e) => e.eventId))
+    const strandedAtStamp = persistedEntries
+      .map((e) => e.eventId)
+      .filter((id) => id < terminalId && !idsAtStamp.includes(id))
+    expect(strandedAtStamp).toEqual([])
+  })
+
+  /**
+   * Pressure has to be measured as events are produced, not once a flush
+   * succeeds. A burst is compacted long before the scheduled flush runs, so
+   * flush-time accounting would let the very batch that exhausts the budget
+   * through at the loose ceiling and drop it instead of offloading it.
+   */
+  it('engages pressure within a burst that has not flushed yet', async () => {
+    mockRedis.incrby.mockResolvedValue(100000)
+    const payload = 'x'.repeat(2 * 1024 * 1024)
+
+    const writer = createExecutionEventWriter('exec-1', {
+      workspaceId: 'ws-1',
+      workflowId: 'wf-1',
+    })
+    // No flush between writes: everything stays pending while the burst builds.
+    for (let i = 0; i < 20; i++) {
+      await writer.write(makeEvent(payload)).catch(() => {})
+    }
+    await writer.flush().catch(() => {})
+
+    // The later events in the burst must have been offloaded, not left inline.
+    const persisted = JSON.stringify(persistedEntries)
+    expect(persisted).toContain(LARGE_VALUE_REF_MARKER)
+  })
+
+  /**
+   * A transient failure while draining the backlog must not cost events, and
+   * must not let the run be marked terminal. Overwriting the queue would drop
+   * entries the budget never rejected, and the drain's final chunk would
+   * otherwise stamp the status before the terminal event is written.
+   */
+  it('retains the backlog and withholds terminal status when the drain fails transiently', async () => {
+    mockRedis.incrby.mockResolvedValue(100000)
+    const stamped: string[] = []
+    let failDrain = true
+    mockRedis.eval.mockImplementation(async (script: string, ...args: unknown[]) => {
+      if (!isFlushScript(script)) return [1, 'ok', 0, 0]
+      const { terminalStatus, zaddArgs } = parseFlushEvalArgs(args)
+      // Reject any multi-entry batch so the terminal-alone retry path is taken.
+      if (zaddArgs.length > 2) return [0, 'execution_redis_bytes', 64 * 1024 * 1024]
+      // The backlog drain hits a transient outage rather than a budget rejection.
+      if (failDrain) {
+        failDrain = false
+        throw new Error('redis unavailable')
+      }
+      for (let i = 0; i < zaddArgs.length; i += 2) {
+        persistedEntries.push(JSON.parse(zaddArgs[i + 1] as string) as ExecutionEventEntry)
+      }
+      if (terminalStatus) stamped.push(terminalStatus)
+      return [1, 1, 0]
+    })
+
+    const payload = 'x'.repeat(1_500_000)
+    const writer = createExecutionEventWriter('exec-1', {
+      workspaceId: 'ws-1',
+      workflowId: 'wf-1',
+    })
+    for (let i = 0; i < 3; i++) {
+      await writer.write(makeEvent(payload)).catch(() => {})
+    }
+    await expect(writer.writeTerminal(makeEvent('terminal'), 'complete')).rejects.toThrow()
+
+    // The transiently-failed backlog is still queued, so it is not lost.
+    expect(stamped).toEqual([])
+    await writer.close().catch(() => {})
+    expect(persistedEntries.length).toBeGreaterThan(0)
+  })
+
+  /**
+   * Offloading under pressure is an optimization. If the value cannot be
+   * persisted durably, the event must still reach the replay buffer inline —
+   * dropping it would leave a reconnecting client permanently missing it.
+   */
+  it('buffers the event inline when a pressure offload cannot be persisted', async () => {
+    mockRedis.incrby.mockResolvedValue(100000)
+    const payload = 'x'.repeat(2 * 1024 * 1024)
+
+    // No workspace/workflow ids, so durable persistence of an offloaded value
+    // fails the way a storage outage would.
+    const writer = createExecutionEventWriter('exec-1')
+    for (let i = 0; i < 20; i++) {
+      await writer.write(makeEvent(payload)).catch(() => {})
+    }
+    await writer.flush().catch(() => {})
+
+    expect(persistedEntries).toHaveLength(20)
+  })
+
   it('preserves requested UserFile base64 when buffering terminal events', async () => {
     mockRedis.incrby.mockResolvedValue(100)
     const base64 = Buffer.from('hello').toString('base64')

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

@@ -27,6 +27,21 @@ const FLUSH_INTERVAL_MS = 15
 const FLUSH_MAX_RETRY_INTERVAL_MS = 1000
 const FLUSH_MAX_BATCH = 200
 const MAX_PENDING_EVENTS = 1000
+/**
+ * Bytes a single execution may buffer before its events start offloading
+ * aggressively, and the per-value threshold applied once it does.
+ *
+ * The buffer holds `EVENT_LIMIT` events inside the per-execution byte budget,
+ * so a full ring only fits if events average under budget/EVENT_LIMIT. Applying
+ * that ceiling to every run would offload ordinary block outputs into refs the
+ * terminal cannot display — the SSE stream carries the compacted event, and a
+ * ref renders only as a preview. Instead the tight ceiling engages only once a
+ * run has actually buffered its way into the danger zone, so a short run keeps
+ * full-fidelity output and a runaway one stops accumulating.
+ */
+const EXECUTION_EVENT_OFFLOAD_PRESSURE_BYTES = getExecutionRedisBudgetLimits().maxExecutionBytes / 2
+const EXECUTION_EVENT_PRESSURE_VALUE_BYTES =
+  getExecutionRedisBudgetLimits().maxExecutionBytes / EVENT_LIMIT
 const ACTIVE_META_ATTEMPTS = 3
 const FINALIZE_FLUSH_ATTEMPTS = 2
 const FLUSH_EVENTS_SCRIPT = `
@@ -282,6 +297,8 @@ export interface ExecutionEventWriter {
 export interface ExecutionEventWriterContext extends LargeValueStoreContext {
   requireDurablePayloads?: boolean
   preserveUserFileBase64?: boolean
+  /** Offload ceiling for individual values; defaults to the shared large-value cap. */
+  valueThresholdBytes?: number
 }
 
 async function compactEventForBuffer(
@@ -297,6 +314,7 @@ async function compactEventForBuffer(
     executionId: context.executionId ?? event.executionId,
     requireDurable: context.requireDurablePayloads,
     preserveRoot: true,
+    thresholdBytes: context.valueThresholdBytes,
   }
 
   let compactedData = await compactExecutionPayload(event.data, {
@@ -746,6 +764,27 @@ export function createExecutionEventWriter(
   let maxReservedId = 0
   let flushTimer: ReturnType<typeof setTimeout> | null = null
   let consecutiveFlushFailures = 0
+  /**
+   * Bytes this execution has produced, counted as each event is compacted
+   * rather than once a flush succeeds. A burst can be compacted long before the
+   * scheduled flush runs, so flush-time accounting would let the very batch that
+   * exhausts the budget through at the loose ceiling. Counted gross of
+   * ring-buffer pruning too, so the mark is reached early — erring toward
+   * offloading sooner is the safe direction.
+   */
+  let bufferedBytes = 0
+
+  /**
+   * Preserved base64 is an explicit request for inline delivery and is already
+   * bounded by its own cap and the strip-and-recompact fallback, so pressure
+   * never rewrites it into a ref the caller cannot read.
+   */
+  const getValueThresholdBytes = () => {
+    if (context.preserveUserFileBase64) return undefined
+    return bufferedBytes >= EXECUTION_EVENT_OFFLOAD_PRESSURE_BYTES
+      ? EXECUTION_EVENT_PRESSURE_VALUE_BYTES
+      : undefined
+  }
 
   const getFlushDelayMs = () => {
     if (consecutiveFlushFailures === 0) return FLUSH_INTERVAL_MS
@@ -983,17 +1022,39 @@ export function createExecutionEventWriter(
     }
   }
 
+  /**
+   * Compact an event for the buffer, degrading if pressure offloading fails.
+   *
+   * Offloading under pressure is an optimization: it keeps a heavy run from
+   * exhausting its budget. When durable storage rejects the write, losing the
+   * event from replay entirely is a worse outcome than carrying it inline, so
+   * fall back to the shared cap — exactly what the run would have done before
+   * pressure engaged.
+   */
+  const compactForBuffer = async (event: ExecutionEvent) => {
+    const valueThresholdBytes = getValueThresholdBytes()
+    const options = { ...context, executionId, requireDurablePayloads: true }
+    if (valueThresholdBytes === undefined) return compactEventForBuffer(event, options)
+    try {
+      return await compactEventForBuffer(event, { ...options, valueThresholdBytes })
+    } catch (error) {
+      logger.warn('Pressure offload failed; buffering the event inline instead', {
+        executionId,
+        eventType: event.type,
+        error: toError(error).message,
+      })
+      return compactEventForBuffer(event, options)
+    }
+  }
+
   const writeCore = async (event: ExecutionEvent): Promise<ExecutionEventEntry> => {
     if (nextEventId === 0 || nextEventId > maxReservedId) {
       await reserveIds(1)
     }
     const eventId = nextEventId++
-    const compactEvent = await compactEventForBuffer(event, {
-      ...context,
-      executionId,
-      requireDurablePayloads: true,
-    })
+    const compactEvent = await compactForBuffer(event)
     const entry: ExecutionEventEntry = { eventId, executionId, event: compactEvent }
+    bufferedBytes += getJsonSize(entry) ?? 0
     pending.push(entry)
     if (pending.length >= FLUSH_MAX_BATCH) {
       await flushPending()
@@ -1034,12 +1095,9 @@ export function createExecutionEventWriter(
         await reserveIds(1)
       }
       const eventId = nextEventId++
-      const compactEvent = await compactEventForBuffer(event, {
-        ...context,
-        executionId,
-        requireDurablePayloads: true,
-      })
+      const compactEvent = await compactForBuffer(event)
       const entry: ExecutionEventEntry = { eventId, executionId, event: compactEvent }
+      bufferedBytes += getJsonSize(entry) ?? 0
       pending.push(entry)
       let ok = false
       try {
@@ -1051,10 +1109,26 @@ export function createExecutionEventWriter(
           // alone rather than losing the run's final status with them. Gated on a
           // budget rejection specifically: a transient Redis error leaves the batch
           // queued for retry, and clearing it here would turn that into data loss.
-          const remaining = pending.filter((pendingEntry) => pendingEntry !== entry)
-          pending = [entry]
-          ok = await flushPending(false)
-          pending = pending.concat(remaining)
+          const terminalStatus = pendingTerminalStatus
+          pending = pending.filter((pendingEntry) => pendingEntry !== entry)
+          if (pending.length > 0) {
+            // Drain what is queued ahead of the terminal event first, with the
+            // status disarmed: `doFlush` stamps it on whichever chunk empties
+            // `pending`, so leaving it armed would mark the run complete before
+            // its terminal event is written. Whatever this cannot persist stays
+            // queued — it must not be overwritten.
+            pendingTerminalStatus = undefined
+            await flushPending(false)
+            pendingTerminalStatus = terminalStatus
+          }
+          if (pending.length === 0) {
+            // Only publish alone once nothing earlier is still queued. Doing so
+            // over a surviving backlog would signal end-of-run to a reader that
+            // has not received those events; failing instead lets the caller
+            // degrade, which records the status without claiming they arrived.
+            pending = [entry]
+            ok = await flushPending(false)
+          }
         }
       } catch (error) {
         discardTerminalEntry(entry)

apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts

@@ -1323,8 +1323,22 @@ export class PauseResumeManager {
               await degradeTerminalPublish(terminalStatus, error)
               return { eventId: 0, executionId: resumeExecutionId, event }
             })
-          : await eventWriter.write(event)
-        event.eventId = entry.eventId
+          : await eventWriter.write(event).catch((error) => {
+              // The buffer only backs reconnect replay; the live stream is the
+              // primary delivery path. Awaiting this bare let a failed write
+              // propagate into the executor callback and fail work that had
+              // already run, so degrade the same way the execute route does.
+              logger.warn('Resume event buffer write failed; delivering live only', {
+                resumeExecutionId,
+                eventType: event.type,
+                error: toError(error).message,
+              })
+              return null
+            })
+        // Leave `eventId` unset when the write failed, matching the execute
+        // route. Assigning 0 here would be persisted as a reconnect cursor and
+        // rewind the client to the start of the run.
+        if (entry) event.eventId = entry.eventId
         terminalEventPublished ||= Boolean(terminalStatus)
       }
       sendEvent?.(event)