From 7c396f7c1a0e2a30dbe3b56bd47513936c3a44c0 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Mon, 3 Aug 2026 15:55:35 -0700 Subject: [PATCH] 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 could drive unbounded heap growth and stall the event loop. 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, and a terminal event whose batch was dropped is retried on its own rather than lost with it. 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. --- .../[id]/execute/route.async.test.ts | 29 ++ .../app/api/workflows/[id]/execute/route.ts | 19 ++ apps/sim/lib/execution/event-buffer.test.ts | 254 ++++++++++++++++++ apps/sim/lib/execution/event-buffer.ts | 158 +++++++++-- .../lib/execution/redis-budget.server.test.ts | 54 +--- apps/sim/lib/execution/redis-budget.server.ts | 112 +------- 6 files changed, 444 insertions(+), 182 deletions(-) diff --git a/apps/sim/app/api/workflows/[id]/execute/route.async.test.ts b/apps/sim/app/api/workflows/[id]/execute/route.async.test.ts index 6da3b5368fc..476e600112c 100644 --- a/apps/sim/app/api/workflows/[id]/execute/route.async.test.ts +++ b/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) diff --git a/apps/sim/app/api/workflows/[id]/execute/route.ts b/apps/sim/app/api/workflows/[id]/execute/route.ts index 73854085f0c..466aab44d48 100644 --- a/apps/sim/app/api/workflows/[id]/execute/route.ts +++ b/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,22 @@ async function handleExecutePost( isStreamClosed = true } } + if (terminalBufferWriteFailed && terminalStatus) { + // Without this the reconnect route polls an `active` stream until its + // deadline. The meta write is a plain HSET, so it bypasses the byte budget + // that rejected the event. Runs after the live enqueue because Redis is the + // likely reason we are here at all, and a slow best-effort durability write + // must not 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 { diff --git a/apps/sim/lib/execution/event-buffer.test.ts b/apps/sim/lib/execution/event-buffer.test.ts index e0287f6d2af..f548a194230 100644 --- a/apps/sim/lib/execution/event-buffer.test.ts +++ b/apps/sim/lib/execution/event-buffer.test.ts @@ -2,6 +2,7 @@ * @vitest-environment node */ 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 type { ExecutionEvent } from '@/lib/workflows/executor/execution-events' @@ -368,6 +369,259 @@ 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. 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) + // Built from many modest fields rather than one huge one: compaction offloads + // individual values over its threshold, so a single large string would leave a + // tiny ref behind and never reach the batch cap. Each event stays under the + // 8MiB per-event cap; two of them do not. + const chunk = 'x'.repeat(100_000) + const wideEvent = () => { + const event = makeEvent('wide') + const data = event.data as Record + for (let i = 0; i < 45; i++) data[`field${i}`] = chunk + return event + } + + const writer = createExecutionEventWriter('exec-1') + await writer.write(wideEvent()) + await writer.write(wideEvent()) + 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((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((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 sleep(5) + releaseFirstFlush?.() + await terminalWrite + + expect(observedTerminalStatuses).toContain('complete') + }) + + /** + * The backlog ahead of a terminal event can exceed the budget while the + * terminal event itself still fits. Discarding it alongside the backlog would + * leave readers without the final status for a run that could have published + * one. + */ + it('still publishes the terminal event when the backlog ahead of it is dropped', async () => { + mockRedis.incrby.mockResolvedValue(100) + const observedTerminalStatuses: string[] = [] + mockRedis.eval.mockImplementation(async (script: string, ...args: unknown[]) => { + if (!isFlushScript(script)) return [1, 'ok', 0, 0] + const { terminalStatus, zaddArgs } = parseFlushEvalArgs(args) + // Reject anything but a lone entry, standing in for a budget with only + // enough headroom left for one small write. + if (zaddArgs.length > 2) return [0, 'execution_redis_bytes', 64 * 1024 * 1024] + 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') + for (let i = 0; i < 5; i++) { + await writer.write(makeEvent(`block-${i}`)).catch(() => {}) + } + + await expect(writer.writeTerminal(makeEvent('terminal'), 'complete')).resolves.toMatchObject({ + executionId: 'exec-1', + }) + expect(observedTerminalStatuses).toContain('complete') + expect( + persistedEntries.map((entry) => (entry.event.data as { blockId: string }).blockId) + ).toContain('terminal') + }) + + /** + * 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 sleep(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') diff --git a/apps/sim/lib/execution/event-buffer.ts b/apps/sim/lib/execution/event-buffer.ts index f2661d21334..07472823674 100644 --- a/apps/sim/lib/execution/event-buffer.ts +++ b/apps/sim/lib/execution/event-buffer.ts @@ -184,10 +184,6 @@ function getJsonSize(value: unknown): number | null { } } -function getExecutionEventEntryJson(entry: ExecutionEventEntry): string { - return JSON.stringify(entry) -} - function getFlushScriptResult(value: unknown): { allowed: boolean resource?: string @@ -764,7 +760,12 @@ export function createExecutionEventWriter( if (flushTimer) return flushTimer = setTimeout(() => { flushTimer = null - void flushPending() + flushPending().catch((error) => { + logger.warn('Scheduled execution event flush failed', { + executionId, + error: toError(error).message, + }) + }) }, delayMs) } @@ -780,33 +781,94 @@ export function createExecutionEventWriter( let flushPromise: Promise | null = null let closed = false + /** + * Why the most recent flush was rejected, cleared by the next success. + * + * Budget rejection is recoverable — the execution counter falls as the ring + * buffer prunes, and the user counter is shared with the caller's other runs — + * so it must not latch the writer off. This only preserves the specific reason + * so a failed terminal flush can report it instead of a generic message. + */ + let lastResourceLimitError: ExecutionResourceLimitError | null = null + /** + * Terminal status awaiting its chunk, scoped to the writer rather than to the + * `flushPending` call that requested it. A concurrent timer-driven flush can + * be the loop that drains the final chunk, and it carries no status of its + * own — reading it from here keeps the run from being written without one. + */ + let pendingTerminalStatus: TerminalExecutionStreamStatus | undefined let writeQueue: Promise = Promise.resolve() const inflightWrites = new Set>() let writeFailure: Error | null = null - const doFlush = async (terminalStatus?: TerminalExecutionStreamStatus): Promise => { + /** + * Largest prefix of `pending` that fits one Redis write, always at least one + * entry. A burst of large events can otherwise build a batch above the + * single-write cap that no retry can ever shrink, stalling the buffer for the + * rest of the run. + */ + const takeFlushChunk = (maxBytes: number) => { + const zaddArgs: (string | number)[] = [] + let bytes = 0 + let count = 0 + // Serialize before detaching anything: an entry that cannot be stringified + // must leave `pending` untouched so the batch is still retried, not silently + // dropped on the floor. + while (count < pending.length) { + const entry = pending[count] + const entryJson = JSON.stringify(entry) + const entryBytes = Buffer.byteLength(entryJson, 'utf8') + if (count > 0 && bytes + entryBytes > maxBytes) break + zaddArgs.push(entry.eventId, entryJson) + bytes += entryBytes + count += 1 + if (bytes > maxBytes) break + } + const entries = pending.slice(0, count) + pending = pending.slice(count) + return { entries, zaddArgs, bytes } + } + + /** + * Abandon a terminal entry whose publish failed. The status must be dropped + * with it: leaving it armed would let a later `flush()`/`close()` stamp the + * stream terminal for an event that was discarded, contradicting the failure + * just reported to the caller. + */ + const discardTerminalEntry = (entry: ExecutionEventEntry) => { + pending = pending.filter((pendingEntry) => pendingEntry !== entry) + pendingTerminalStatus = undefined + } + + /** + * Resolves `false` on every failure and never rejects, so a detached flush + * cannot surface as an unhandled rejection. Callers keep their own guards as + * defence in depth for that invariant. + */ + const doFlush = async (): Promise => { if (pending.length === 0) return true - const batch = pending - pending = [] + let batch: ExecutionEventEntry[] = [] + let batchBytes = 0 try { + const limits = getExecutionRedisBudgetLimits() + const chunk = takeFlushChunk(limits.maxSingleWriteBytes) + batch = chunk.entries + batchBytes = chunk.bytes + const { zaddArgs } = chunk + // Only authoritative once the final chunk lands. + const chunkTerminalStatus = pending.length === 0 ? pendingTerminalStatus : undefined const key = getEventsKey(executionId) - const zaddArgs: (string | number)[] = [] - let batchBytes = 0 - for (const entry of batch) { - const entryJson = getExecutionEventEntryJson(entry) - batchBytes += Buffer.byteLength(entryJson, 'utf8') - zaddArgs.push(entry.eventId, entryJson) - } const budgetReservation: ExecutionRedisBudgetReservation = { executionId, userId: context.userId, category: 'event_buffer', - operation: terminalStatus ? 'write_terminal_events' : 'write_events', + operation: chunkTerminalStatus ? 'write_terminal_events' : 'write_events', bytes: batchBytes, logger, } - const limits = getExecutionRedisBudgetLimits() if (batchBytes > limits.maxSingleWriteBytes) { + // A single entry above the cap can never be written; dropping it is the + // only way the rest of the buffer makes progress. throw new ExecutionResourceLimitError({ resource: 'redis_key_bytes', attemptedBytes: batchBytes, @@ -825,7 +887,7 @@ export function createExecutionEventWriter( TTL_SECONDS, EVENT_LIMIT, new Date().toISOString(), - terminalStatus ?? '', + chunkTerminalStatus ?? '', batchBytes, limits.maxExecutionBytes, limits.maxUserBytes, @@ -848,13 +910,31 @@ export function createExecutionEventWriter( }) } consecutiveFlushFailures = 0 + lastResourceLimitError = null + if (chunkTerminalStatus) pendingTerminalStatus = undefined return true } catch (error) { if (isExecutionResourceLimitError(error)) { - pending = batch.concat(pending) - throw error + // Requeueing is what let `pending` grow for a whole run: the batch was + // restored and retried, each attempt re-serializing an ever-larger array. + // These bytes cannot be persisted, so drop them and let backoff pace the + // retries — the budget frees as the ring buffer prunes and as the + // caller's other executions finish. + consecutiveFlushFailures += 1 + lastResourceLimitError = error instanceof ExecutionResourceLimitError ? error : null + logger.warn('Dropped execution events that exceeded the Redis byte budget', { + executionId, + droppedEvents: batch.length, + droppedBytes: batchBytes, + consecutiveFailures: consecutiveFlushFailures, + resource: (error as Partial).resource, + }) + return false } consecutiveFlushFailures += 1 + // Only report a budget rejection when it was the most recent cause; a + // stale one would surface a Redis outage as a bogus 413. + lastResourceLimitError = null logger.warn('Failed to flush execution events', { executionId, batchSize: batch.length, @@ -880,6 +960,7 @@ export function createExecutionEventWriter( scheduleOnFailure = true, terminalStatus?: TerminalExecutionStreamStatus ): Promise => { + if (terminalStatus) pendingTerminalStatus = terminalStatus while (true) { if (flushPromise) { const ok = await flushPromise @@ -888,7 +969,7 @@ export function createExecutionEventWriter( } if (pending.length === 0) return true - flushPromise = doFlush(terminalStatus) + flushPromise = doFlush() let ok = false try { ok = await flushPromise @@ -960,10 +1041,37 @@ export function createExecutionEventWriter( }) const entry: ExecutionEventEntry = { eventId, executionId, event: compactEvent } pending.push(entry) - const ok = await flushPending(false, status) - if (!ok) { - pending = pending.filter((pendingEntry) => pendingEntry !== entry) - throw new Error(`Failed to flush terminal execution event for ${executionId}`) + let ok = false + try { + ok = await flushPending(false, status) + if (!ok && lastResourceLimitError && pendingTerminalStatus) { + // The batch carrying the terminal event exceeded the budget and was + // dropped. Those bytes are gone either way, and the terminal event is + // small enough to plausibly fit on its own — so give it one attempt + // 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) + } + } catch (error) { + discardTerminalEntry(entry) + throw error + } + // `pendingTerminalStatus` still being set means the status never reached + // Redis even though the events did, which would leave readers polling an + // `active` stream. Treat it as a failed terminal publish so the caller + // degrades instead of reporting a success that readers cannot observe. + if (!ok || pendingTerminalStatus) { + discardTerminalEntry(entry) + // Report why when the budget was the cause, so the run surfaces the + // actionable "reduce payload size" message rather than a generic failure. + throw ( + lastResourceLimitError ?? + new Error(`Failed to flush terminal execution event for ${executionId}`) + ) } closed = true return entry diff --git a/apps/sim/lib/execution/redis-budget.server.test.ts b/apps/sim/lib/execution/redis-budget.server.test.ts index af2f0338b90..f9456fc3fae 100644 --- a/apps/sim/lib/execution/redis-budget.server.test.ts +++ b/apps/sim/lib/execution/redis-budget.server.test.ts @@ -1,37 +1,10 @@ /** * @vitest-environment node */ -import { describe, expect, it, vi } from 'vitest' -import { - getExecutionRedisBudgetKeys, - reserveExecutionRedisBytes, -} from '@/lib/execution/redis-budget.server' +import { describe, expect, it } from 'vitest' +import { getExecutionRedisBudgetKeys } from '@/lib/execution/redis-budget.server' -function countOccurrences(haystack: string, needle: string): number { - return haystack.split(needle).length - 1 -} - -async function captureReserveScript(userId?: string): Promise { - let script = '' - const redis = { - eval: vi.fn(async (source: string) => { - script = source - return [1, 'ok', 0, 0] - }), - } - - await reserveExecutionRedisBytes(redis as never, { - executionId: 'exec-1', - userId, - category: 'event_buffer', - operation: 'write_events', - bytes: 128, - }) - - return script -} - -describe('reserveExecutionRedisBytes', () => { +describe('getExecutionRedisBudgetKeys', () => { it('scopes the reservation to the execution, and to the user when one is known', () => { expect( getExecutionRedisBudgetKeys({ @@ -52,25 +25,4 @@ describe('reserveExecutionRedisBytes', () => { }) ).toEqual(['execution:redis-budget:execution:exec-1', 'execution:redis-budget:user:user-1']) }) - - /** - * A user key aggregates across every execution that user runs, so extending - * its TTL on each write keeps it alive indefinitely while the per-execution - * data it accounts for expires underneath it. The window must be fixed. - */ - it('never extends an existing user budget window', async () => { - const script = await captureReserveScript('user-1') - - const userKeyExpires = countOccurrences(script, "redis.call('EXPIRE', KEYS[2]") - const userKeyTtlGuards = countOccurrences(script, "redis.call('TTL', KEYS[2]) < 0") - expect(userKeyExpires).toBeGreaterThan(0) - expect(userKeyTtlGuards).toBe(userKeyExpires) - }) - - it('keeps sliding the execution budget window, which expires with its own data', async () => { - const script = await captureReserveScript('user-1') - - expect(countOccurrences(script, "redis.call('TTL', KEYS[1]) < 0")).toBe(0) - expect(countOccurrences(script, "redis.call('EXPIRE', KEYS[1]")).toBeGreaterThan(0) - }) }) diff --git a/apps/sim/lib/execution/redis-budget.server.ts b/apps/sim/lib/execution/redis-budget.server.ts index 1b922171d55..daf5822fb4c 100644 --- a/apps/sim/lib/execution/redis-budget.server.ts +++ b/apps/sim/lib/execution/redis-budget.server.ts @@ -1,19 +1,15 @@ -import { createLogger, type Logger } from '@sim/logger' -import { toError } from '@sim/utils/errors' -import type { getRedisClient } from '@/lib/core/config/redis' -import { ExecutionResourceLimitError } from '@/lib/execution/resource-errors' +import type { Logger } from '@sim/logger' -type RedisClient = NonNullable> - -const logger = createLogger('ExecutionRedisBudget') const REDIS_BUDGET_PREFIX = 'execution:redis-budget:' const MAX_SINGLE_REDIS_WRITE_BYTES = 8 * 1024 * 1024 const MAX_EXECUTION_REDIS_BYTES = 64 * 1024 * 1024 const MAX_USER_REDIS_BYTES = 256 * 1024 * 1024 -const REDIS_BUDGET_TTL_SECONDS = 60 * 60 /** - * Execution and user budget keys expire differently on purpose. + * Window applied to both budget keys, but extended differently on purpose by + * every Lua script that enforces them — `FLUSH_EVENTS_SCRIPT` and + * `RESET_STREAM_SCRIPT` in `event-buffer.ts`, and the base64 cache pair in + * `lib/uploads/utils/user-file-base64.server.ts`. * * An execution key accounts for data that is refreshed on the same schedule as * the key itself, so sliding its TTL on every write keeps the counter and the @@ -27,44 +23,7 @@ const REDIS_BUDGET_TTL_SECONDS = 60 * 60 * keys therefore get a fixed window: the TTL is set when the key is created * and never extended. */ - -const RESERVE_REDIS_BYTES_SCRIPT = ` -local bytes = tonumber(ARGV[1]) -local execution_limit = tonumber(ARGV[2]) -local user_limit = tonumber(ARGV[3]) -local ttl_seconds = tonumber(ARGV[4]) -local execution_current = tonumber(redis.call('GET', KEYS[1]) or '0') -if execution_limit > 0 and execution_current + bytes > execution_limit then - return {0, 'execution_redis_bytes', execution_current} -end -local user_current = 0 -if #KEYS >= 2 then - user_current = tonumber(redis.call('GET', KEYS[2]) or '0') - if user_limit > 0 and user_current + bytes > user_limit then - return {0, 'user_redis_bytes', user_current} - end -end -redis.call('INCRBY', KEYS[1], bytes) -redis.call('EXPIRE', KEYS[1], ttl_seconds) -if #KEYS >= 2 then - redis.call('INCRBY', KEYS[2], bytes) - if redis.call('TTL', KEYS[2]) < 0 then - redis.call('EXPIRE', KEYS[2], ttl_seconds) - end -end -return {1, 'ok', execution_current + bytes, user_current + bytes} -` - -const RELEASE_REDIS_BYTES_SCRIPT = ` -local bytes = tonumber(ARGV[1]) -for i = 1, #KEYS do - local next_value = redis.call('DECRBY', KEYS[i], bytes) - if next_value <= 0 then - redis.call('DEL', KEYS[i]) - end -end -return 1 -` +const REDIS_BUDGET_TTL_SECONDS = 60 * 60 export type ExecutionRedisBudgetCategory = 'event_buffer' | 'base64_cache' @@ -95,62 +54,3 @@ export function getExecutionRedisBudgetKeys( } return keys } - -export async function reserveExecutionRedisBytes( - redis: RedisClient, - reservation: ExecutionRedisBudgetReservation -): Promise { - if (reservation.bytes <= 0) return - - const limits = getExecutionRedisBudgetLimits() - if (reservation.bytes > limits.maxSingleWriteBytes) { - throw new ExecutionResourceLimitError({ - resource: 'redis_key_bytes', - attemptedBytes: reservation.bytes, - limitBytes: limits.maxSingleWriteBytes, - }) - } - - const keys = getExecutionRedisBudgetKeys(reservation) - const result = (await redis.eval( - RESERVE_REDIS_BYTES_SCRIPT, - keys.length, - ...keys, - reservation.bytes, - limits.maxExecutionBytes, - limits.maxUserBytes, - limits.ttlSeconds - )) as [number, string, number | string | null] - - const [allowed, resource, current] = result - if (allowed === 1) return - - throw new ExecutionResourceLimitError({ - resource: resource === 'user_redis_bytes' ? 'user_redis_bytes' : 'execution_redis_bytes', - attemptedBytes: reservation.bytes, - currentBytes: Number(current ?? 0), - limitBytes: resource === 'user_redis_bytes' ? limits.maxUserBytes : limits.maxExecutionBytes, - }) -} - -export async function releaseExecutionRedisBytes( - redis: RedisClient, - reservation: ExecutionRedisBudgetReservation -): Promise { - if (reservation.bytes <= 0) return - - try { - const keys = getExecutionRedisBudgetKeys(reservation) - await redis.eval(RELEASE_REDIS_BYTES_SCRIPT, keys.length, ...keys, reservation.bytes) - } catch (error) { - const log = reservation.logger ?? logger - log.warn('Failed to release execution Redis budget reservation', { - executionId: reservation.executionId, - userId: reservation.userId, - category: reservation.category, - operation: reservation.operation, - bytes: reservation.bytes, - error: toError(error).message, - }) - } -}