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, - }) - } -}