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

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)