From 2e6bcd59a87b05b8bfcc11c6d7ab701c33212d3b Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Mon, 3 Aug 2026 16:30:52 -0700 Subject: [PATCH 1/4] fix(execution): offload buffered event values under budget pressure MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- apps/sim/lib/execution/event-buffer.test.ts | 49 +++++++++++++++++++ apps/sim/lib/execution/event-buffer.ts | 39 +++++++++++++++ .../executor/human-in-the-loop-manager.ts | 13 ++++- 3 files changed, 100 insertions(+), 1 deletion(-) diff --git a/apps/sim/lib/execution/event-buffer.test.ts b/apps/sim/lib/execution/event-buffer.test.ts index f548a194230..4bf5cb2135d 100644 --- a/apps/sim/lib/execution/event-buffer.test.ts +++ b/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,54 @@ 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) + }) + it('preserves requested UserFile base64 when buffering terminal events', async () => { mockRedis.incrby.mockResolvedValue(100) const base64 = Buffer.from('hello').toString('base64') diff --git a/apps/sim/lib/execution/event-buffer.ts b/apps/sim/lib/execution/event-buffer.ts index 07472823674..12cfbf054a7 100644 --- a/apps/sim/lib/execution/event-buffer.ts +++ b/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,24 @@ export function createExecutionEventWriter( let maxReservedId = 0 let flushTimer: ReturnType | null = null let consecutiveFlushFailures = 0 + /** + * Bytes this execution has successfully buffered. Counted gross rather than + * net of ring-buffer pruning, so it reaches the pressure mark 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 @@ -911,6 +947,7 @@ export function createExecutionEventWriter( } consecutiveFlushFailures = 0 lastResourceLimitError = null + bufferedBytes += batchBytes if (chunkTerminalStatus) pendingTerminalStatus = undefined return true } catch (error) { @@ -992,6 +1029,7 @@ export function createExecutionEventWriter( ...context, executionId, requireDurablePayloads: true, + valueThresholdBytes: getValueThresholdBytes(), }) const entry: ExecutionEventEntry = { eventId, executionId, event: compactEvent } pending.push(entry) @@ -1038,6 +1076,7 @@ export function createExecutionEventWriter( ...context, executionId, requireDurablePayloads: true, + valueThresholdBytes: getValueThresholdBytes(), }) const entry: ExecutionEventEntry = { eventId, executionId, event: compactEvent } pending.push(entry) diff --git a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts index ee495660480..ed7acf2ece2 100644 --- a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts +++ b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts @@ -1323,7 +1323,18 @@ export class PauseResumeManager { await degradeTerminalPublish(terminalStatus, error) return { eventId: 0, executionId: resumeExecutionId, event } }) - : await eventWriter.write(event) + : 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 { eventId: 0, executionId: resumeExecutionId, event } + }) event.eventId = entry.eventId terminalEventPublished ||= Boolean(terminalStatus) } From 01f34e55e45fb75caa55ecd4f08df73b2d55af6b Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Mon, 3 Aug 2026 16:47:33 -0700 Subject: [PATCH 2/4] fix(execution): measure pressure at write time and keep terminal status last MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- apps/sim/lib/execution/event-buffer.test.ts | 71 +++++++++++++++++++++ apps/sim/lib/execution/event-buffer.ts | 21 ++++-- 2 files changed, 87 insertions(+), 5 deletions(-) diff --git a/apps/sim/lib/execution/event-buffer.test.ts b/apps/sim/lib/execution/event-buffer.test.ts index 4bf5cb2135d..d64039db5d2 100644 --- a/apps/sim/lib/execution/event-buffer.test.ts +++ b/apps/sim/lib/execution/event-buffer.test.ts @@ -671,6 +671,77 @@ describe('execution event buffer', () => { 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) + }) + it('preserves requested UserFile base64 when buffering terminal events', async () => { mockRedis.incrby.mockResolvedValue(100) const base64 = Buffer.from('hello').toString('base64') diff --git a/apps/sim/lib/execution/event-buffer.ts b/apps/sim/lib/execution/event-buffer.ts index 12cfbf054a7..885b23302fd 100644 --- a/apps/sim/lib/execution/event-buffer.ts +++ b/apps/sim/lib/execution/event-buffer.ts @@ -765,9 +765,12 @@ export function createExecutionEventWriter( let flushTimer: ReturnType | null = null let consecutiveFlushFailures = 0 /** - * Bytes this execution has successfully buffered. Counted gross rather than - * net of ring-buffer pruning, so it reaches the pressure mark early — erring - * toward offloading sooner is the safe direction. + * 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 @@ -947,7 +950,6 @@ export function createExecutionEventWriter( } consecutiveFlushFailures = 0 lastResourceLimitError = null - bufferedBytes += batchBytes if (chunkTerminalStatus) pendingTerminalStatus = undefined return true } catch (error) { @@ -1032,6 +1034,7 @@ export function createExecutionEventWriter( valueThresholdBytes: getValueThresholdBytes(), }) const entry: ExecutionEventEntry = { eventId, executionId, event: compactEvent } + bufferedBytes += getJsonSize(entry) ?? 0 pending.push(entry) if (pending.length >= FLUSH_MAX_BATCH) { await flushPending() @@ -1079,6 +1082,7 @@ export function createExecutionEventWriter( valueThresholdBytes: getValueThresholdBytes(), }) const entry: ExecutionEventEntry = { eventId, executionId, event: compactEvent } + bufferedBytes += getJsonSize(entry) ?? 0 pending.push(entry) let ok = false try { @@ -1091,9 +1095,16 @@ export function createExecutionEventWriter( // 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) + // Drain what is queued ahead of the terminal event first. Terminal + // status is the reader's end-of-run signal: stamping it while lower + // event ids are still queued strands them behind a stream the reader + // has already drained and closed. + if (remaining.length > 0) { + pending = remaining + await flushPending(false) + } pending = [entry] ok = await flushPending(false) - pending = pending.concat(remaining) } } catch (error) { discardTerminalEntry(entry) From 84d9164baca17124f1d3b39d0ed8abd560e17fab Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Mon, 3 Aug 2026 16:57:31 -0700 Subject: [PATCH 3/4] fix(execution): do not lose the backlog or publish terminal status early MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- apps/sim/lib/execution/event-buffer.test.ts | 43 +++++++++++++++++++ apps/sim/lib/execution/event-buffer.ts | 27 ++++++++---- .../executor/human-in-the-loop-manager.ts | 7 ++- 3 files changed, 66 insertions(+), 11 deletions(-) diff --git a/apps/sim/lib/execution/event-buffer.test.ts b/apps/sim/lib/execution/event-buffer.test.ts index d64039db5d2..d11b52b5bd2 100644 --- a/apps/sim/lib/execution/event-buffer.test.ts +++ b/apps/sim/lib/execution/event-buffer.test.ts @@ -742,6 +742,49 @@ describe('execution event buffer', () => { 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) + }) + it('preserves requested UserFile base64 when buffering terminal events', async () => { mockRedis.incrby.mockResolvedValue(100) const base64 = Buffer.from('hello').toString('base64') diff --git a/apps/sim/lib/execution/event-buffer.ts b/apps/sim/lib/execution/event-buffer.ts index 885b23302fd..0a125928aea 100644 --- a/apps/sim/lib/execution/event-buffer.ts +++ b/apps/sim/lib/execution/event-buffer.ts @@ -1094,17 +1094,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) - // Drain what is queued ahead of the terminal event first. Terminal - // status is the reader's end-of-run signal: stamping it while lower - // event ids are still queued strands them behind a stream the reader - // has already drained and closed. - if (remaining.length > 0) { - pending = 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) } - pending = [entry] - ok = await flushPending(false) } } catch (error) { discardTerminalEntry(entry) diff --git a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts index ed7acf2ece2..5f004d49d2e 100644 --- a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts +++ b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts @@ -1333,9 +1333,12 @@ export class PauseResumeManager { eventType: event.type, error: toError(error).message, }) - return { eventId: 0, executionId: resumeExecutionId, event } + return null }) - event.eventId = entry.eventId + // 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) From e2e4138dac082a6f01026f5547fd1a2185f9f4a5 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Mon, 3 Aug 2026 17:06:58 -0700 Subject: [PATCH 4/4] fix(execution): keep an event in the buffer when a pressure offload fails MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- apps/sim/lib/execution/event-buffer.test.ts | 20 +++++++++++ apps/sim/lib/execution/event-buffer.ts | 39 ++++++++++++++------- 2 files changed, 47 insertions(+), 12 deletions(-) diff --git a/apps/sim/lib/execution/event-buffer.test.ts b/apps/sim/lib/execution/event-buffer.test.ts index d11b52b5bd2..041fda1c85e 100644 --- a/apps/sim/lib/execution/event-buffer.test.ts +++ b/apps/sim/lib/execution/event-buffer.test.ts @@ -785,6 +785,26 @@ describe('execution event buffer', () => { 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') diff --git a/apps/sim/lib/execution/event-buffer.ts b/apps/sim/lib/execution/event-buffer.ts index 0a125928aea..ebc7d2ec628 100644 --- a/apps/sim/lib/execution/event-buffer.ts +++ b/apps/sim/lib/execution/event-buffer.ts @@ -1022,17 +1022,37 @@ 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 => { if (nextEventId === 0 || nextEventId > maxReservedId) { await reserveIds(1) } const eventId = nextEventId++ - const compactEvent = await compactEventForBuffer(event, { - ...context, - executionId, - requireDurablePayloads: true, - valueThresholdBytes: getValueThresholdBytes(), - }) + const compactEvent = await compactForBuffer(event) const entry: ExecutionEventEntry = { eventId, executionId, event: compactEvent } bufferedBytes += getJsonSize(entry) ?? 0 pending.push(entry) @@ -1075,12 +1095,7 @@ export function createExecutionEventWriter( await reserveIds(1) } const eventId = nextEventId++ - const compactEvent = await compactEventForBuffer(event, { - ...context, - executionId, - requireDurablePayloads: true, - valueThresholdBytes: getValueThresholdBytes(), - }) + const compactEvent = await compactForBuffer(event) const entry: ExecutionEventEntry = { eventId, executionId, event: compactEvent } bufferedBytes += getJsonSize(entry) ?? 0 pending.push(entry)