diff --git a/apps/sim/lib/execution/event-buffer.test.ts b/apps/sim/lib/execution/event-buffer.test.ts index f548a194230..041fda1c85e 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,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') diff --git a/apps/sim/lib/execution/event-buffer.ts b/apps/sim/lib/execution/event-buffer.ts index 07472823674..ebc7d2ec628 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,27 @@ export function createExecutionEventWriter( let maxReservedId = 0 let flushTimer: ReturnType | 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 => { 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) 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..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 @@ -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)