diff --git a/apps/sim/app/api/workflows/[id]/executions/[executionId]/stream/route.test.ts b/apps/sim/app/api/workflows/[id]/executions/[executionId]/stream/route.test.ts index 5ab5775652b..bc1c4b981db 100644 --- a/apps/sim/app/api/workflows/[id]/executions/[executionId]/stream/route.test.ts +++ b/apps/sim/app/api/workflows/[id]/executions/[executionId]/stream/route.test.ts @@ -90,7 +90,7 @@ describe('execution stream reconnect route', () => { expect(mockReadExecutionEventsState).toHaveBeenNthCalledWith(2, 'exec-1', 3) }) - it('errors when terminal metadata has no terminal event to replay', async () => { + it('ends the stream cleanly when terminal metadata has no terminal event to replay', async () => { mockReadExecutionMetaState .mockResolvedValueOnce({ status: 'found', @@ -115,9 +115,7 @@ describe('execution stream reconnect route', () => { }) expect(response.status).toBe(200) - await expect(response.text()).rejects.toThrow( - 'Execution reached terminal metadata without a terminal event' - ) + await expect(response.text()).resolves.toContain('data: [DONE]') }) it('allows replay event id gaps from reserved but unused writer ids', async () => { diff --git a/apps/sim/app/api/workflows/[id]/executions/[executionId]/stream/route.ts b/apps/sim/app/api/workflows/[id]/executions/[executionId]/stream/route.ts index 2be5fed1c37..1e4c1936141 100644 --- a/apps/sim/app/api/workflows/[id]/executions/[executionId]/stream/route.ts +++ b/apps/sim/app/api/workflows/[id]/executions/[executionId]/stream/route.ts @@ -142,9 +142,20 @@ export const GET = withRouteHandler( if (!closed) controller.close() } + /** + * Terminal metadata is the authoritative end-of-run signal. The + * happy path writes it atomically with the terminal event, and a run + * whose terminal event could not be buffered records the status on + * its own — so metadata without a matching event means the buffer + * degraded, not that the run is still going. End the stream cleanly: + * the reader has already received every event that was buffered, and + * failing here would turn a degraded replay into a broken one. + */ const closeAfterTerminalEvent = (events: ExecutionEventEntry[]) => { if (!enqueueEvents(events)) { - throw new Error('Execution reached terminal metadata without a terminal event') + logger.warn('Execution reached terminal metadata without a terminal event', { + executionId, + }) } closeWithDone() } diff --git a/apps/sim/lib/execution/event-buffer.test.ts b/apps/sim/lib/execution/event-buffer.test.ts index 93dbd3bff07..e0287f6d2af 100644 --- a/apps/sim/lib/execution/event-buffer.test.ts +++ b/apps/sim/lib/execution/event-buffer.test.ts @@ -68,6 +68,10 @@ function isResetScript(script: string): boolean { return script.includes('retained_bytes') && script.includes('replayStartEventId') } +function countOccurrences(haystack: string, needle: string): number { + return haystack.split(needle).length - 1 +} + describe('execution event buffer', () => { beforeEach(() => { vi.clearAllMocks() @@ -417,6 +421,37 @@ describe('execution event buffer', () => { ) }) + it('never extends an existing user budget window while flushing events', async () => { + let flushScript = '' + mockRedis.eval.mockImplementation(async (script: string) => { + if (isFlushScript(script)) flushScript = script + return [1, false, 0] + }) + + const writer = createExecutionEventWriter('exec-1', { userId: 'user-1' }) + await writer.writeTerminal(makeEvent('terminal'), 'complete') + + expect(flushScript).not.toBe('') + const userKeyExpires = countOccurrences(flushScript, "redis.call('EXPIRE', KEYS[5]") + const userKeyTtlGuards = countOccurrences(flushScript, "redis.call('TTL', KEYS[5]) < 0") + expect(userKeyExpires).toBeGreaterThan(0) + expect(userKeyTtlGuards).toBe(userKeyExpires) + }) + + it('keeps sliding the execution budget window, which expires with its own data', async () => { + let flushScript = '' + mockRedis.eval.mockImplementation(async (script: string) => { + if (isFlushScript(script)) flushScript = script + return [1, false, 0] + }) + + const writer = createExecutionEventWriter('exec-1', { userId: 'user-1' }) + await writer.writeTerminal(makeEvent('terminal'), 'complete') + + expect(countOccurrences(flushScript, "redis.call('TTL', KEYS[4]) < 0")).toBe(0) + expect(countOccurrences(flushScript, "redis.call('EXPIRE', KEYS[4]")).toBeGreaterThan(0) + }) + it('reports pruned replay buffers before reading incomplete events', async () => { mockRedis.hgetall.mockResolvedValue({ status: 'active', earliestEventId: '10' }) diff --git a/apps/sim/lib/execution/event-buffer.ts b/apps/sim/lib/execution/event-buffer.ts index 5afe0c4ccaf..f2661d21334 100644 --- a/apps/sim/lib/execution/event-buffer.ts +++ b/apps/sim/lib/execution/event-buffer.ts @@ -85,7 +85,9 @@ if net_bytes > 0 then redis.call('EXPIRE', KEYS[4], budget_ttl_seconds) if #KEYS >= 5 then redis.call('INCRBY', KEYS[5], net_bytes) - redis.call('EXPIRE', KEYS[5], budget_ttl_seconds) + if redis.call('TTL', KEYS[5]) < 0 then + redis.call('EXPIRE', KEYS[5], budget_ttl_seconds) + end end elseif net_bytes < 0 then local release_bytes = -net_bytes @@ -99,7 +101,7 @@ elseif net_bytes < 0 then local user_next = redis.call('DECRBY', KEYS[5], release_bytes) if user_next <= 0 then redis.call('DEL', KEYS[5]) - else + elseif redis.call('TTL', KEYS[5]) < 0 then redis.call('EXPIRE', KEYS[5], budget_ttl_seconds) end end @@ -107,7 +109,7 @@ else if redis.call('EXISTS', KEYS[4]) == 1 then redis.call('EXPIRE', KEYS[4], budget_ttl_seconds) end - if #KEYS >= 5 and redis.call('EXISTS', KEYS[5]) == 1 then + if #KEYS >= 5 and redis.call('EXISTS', KEYS[5]) == 1 and redis.call('TTL', KEYS[5]) < 0 then redis.call('EXPIRE', KEYS[5], budget_ttl_seconds) end end @@ -148,7 +150,7 @@ if retained_bytes > 0 then local user_next = redis.call('DECRBY', KEYS[4], retained_bytes) if user_next <= 0 then redis.call('DEL', KEYS[4]) - else + elseif redis.call('TTL', KEYS[4]) < 0 then redis.call('EXPIRE', KEYS[4], tonumber(ARGV[4])) end end diff --git a/apps/sim/lib/execution/redis-budget.server.test.ts b/apps/sim/lib/execution/redis-budget.server.test.ts new file mode 100644 index 00000000000..af2f0338b90 --- /dev/null +++ b/apps/sim/lib/execution/redis-budget.server.test.ts @@ -0,0 +1,76 @@ +/** + * @vitest-environment node + */ +import { describe, expect, it, vi } from 'vitest' +import { + getExecutionRedisBudgetKeys, + reserveExecutionRedisBytes, +} 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', () => { + it('scopes the reservation to the execution, and to the user when one is known', () => { + expect( + getExecutionRedisBudgetKeys({ + executionId: 'exec-1', + category: 'event_buffer', + operation: 'write_events', + bytes: 1, + }) + ).toEqual(['execution:redis-budget:execution:exec-1']) + + expect( + getExecutionRedisBudgetKeys({ + executionId: 'exec-1', + userId: 'user-1', + category: 'event_buffer', + operation: 'write_events', + bytes: 1, + }) + ).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 ddf58b1772d..1b922171d55 100644 --- a/apps/sim/lib/execution/redis-budget.server.ts +++ b/apps/sim/lib/execution/redis-budget.server.ts @@ -12,6 +12,22 @@ 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. + * + * 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 + * bytes it represents in step. + * + * A user key aggregates across every execution that user runs. Sliding its TTL + * on each write keeps it alive indefinitely for any user who stays active, + * while the per-execution data it accounts for keeps expiring underneath it — + * so the counter accrues bytes Redis has already dropped and eventually pins + * the user at their ceiling until they go a full TTL without writing. User + * 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]) @@ -32,7 +48,9 @@ redis.call('INCRBY', KEYS[1], bytes) redis.call('EXPIRE', KEYS[1], ttl_seconds) if #KEYS >= 2 then redis.call('INCRBY', KEYS[2], bytes) - redis.call('EXPIRE', KEYS[2], ttl_seconds) + 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} ` diff --git a/apps/sim/lib/uploads/utils/user-file-base64.server.ts b/apps/sim/lib/uploads/utils/user-file-base64.server.ts index ba0407dd766..d9195d47c2e 100644 --- a/apps/sim/lib/uploads/utils/user-file-base64.server.ts +++ b/apps/sim/lib/uploads/utils/user-file-base64.server.ts @@ -59,7 +59,7 @@ if bytes and bytes > 0 then local user_next = redis.call('DECRBY', KEYS[4], bytes) if user_next <= 0 then redis.call('DEL', KEYS[4]) - else + elseif redis.call('TTL', KEYS[4]) < 0 then redis.call('EXPIRE', KEYS[4], budget_ttl_seconds) end end @@ -129,7 +129,7 @@ if #KEYS >= 4 then redis.call('DEL', KEYS[4]) end end - if redis.call('EXISTS', KEYS[4]) == 1 then + if redis.call('EXISTS', KEYS[4]) == 1 and redis.call('TTL', KEYS[4]) < 0 then redis.call('EXPIRE', KEYS[4], budget_ttl_seconds) end end 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 c6cc5a75c7d..ee495660480 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 @@ -14,6 +14,7 @@ import { flushExecutionStreamReplayBuffer, initializeExecutionStreamMeta, resetExecutionStreamBuffer, + setExecutionMeta, type TerminalExecutionStreamStatus, } from '@/lib/execution/event-buffer' import { @@ -71,7 +72,6 @@ const execDb = dbFor('exec') const logger = createLogger('HumanInTheLoopManager') const RUN_BUFFER_UNAVAILABLE_ERROR = 'Run buffer temporarily unavailable' -const TERMINAL_PUBLISH_ERROR = 'Run buffer terminal event publish failed' const RESUMABLE_PAUSED_STATUSES = ['paused', 'partially_resumed'] as const const CANCELLABLE_PAUSED_STATUSES = ['paused', 'partially_resumed'] as const const AUTOMATIC_RESUME_INTERVENTION_PREFIX = 'Automatic resume requires manual intervention: ' @@ -771,7 +771,7 @@ export class PauseResumeManager { preserveForRetry: true, retryable: error.retryable, }) - } else if (message === RUN_BUFFER_UNAVAILABLE_ERROR || message === TERMINAL_PUBLISH_ERROR) { + } else if (message === RUN_BUFFER_UNAVAILABLE_ERROR) { await PauseResumeManager.markResumeAttemptFailed({ resumeEntryId, pausedExecutionId: pausedExecution.id, @@ -1280,6 +1280,38 @@ export class PauseResumeManager { } let terminalEventPublished = false + let terminalPublishDegraded = false + + /** + * The run buffer is a replay convenience for stream readers; the durable + * execution record is authoritative. A failed terminal publish must not + * decide the outcome of work that already ran — the resume is not + * retryable at this point, so throwing here would strand the execution as + * paused and re-run its side effects on the next attempt. Degrade instead: + * record the terminal status on the stream meta so readers are not left + * polling an 'active' stream forever, and let the resume settle normally. + */ + const degradeTerminalPublish = async ( + terminalStatus: TerminalExecutionStreamStatus, + error: unknown + ) => { + terminalPublishDegraded = true + logger.warn('Failed to publish resume terminal event', { + resumeExecutionId, + status: terminalStatus, + error: toError(error).message, + }) + const metaPersisted = await setExecutionMeta(resumeExecutionId, { + status: terminalStatus, + }).catch(() => false) + if (!metaPersisted) { + logger.warn('Failed to record degraded terminal status on resume stream meta', { + resumeExecutionId, + status: terminalStatus, + }) + } + } + const writeBufferedEvent = async ( event: ExecutionEvent, terminalStatus?: TerminalExecutionStreamStatus @@ -1287,13 +1319,9 @@ export class PauseResumeManager { const isBuffered = !LIVE_ONLY_EXECUTION_EVENT_TYPES.has(event.type) if (isBuffered) { const entry = terminalStatus - ? await eventWriter.writeTerminal(event, terminalStatus).catch((error) => { - logger.warn('Failed to publish resume terminal event', { - resumeExecutionId, - status: terminalStatus, - error: toError(error).message, - }) - throw new Error(TERMINAL_PUBLISH_ERROR) + ? await eventWriter.writeTerminal(event, terminalStatus).catch(async (error) => { + await degradeTerminalPublish(terminalStatus, error) + return { eventId: 0, executionId: resumeExecutionId, event } }) : await eventWriter.write(event) event.eventId = entry.eventId @@ -1686,9 +1714,10 @@ export class PauseResumeManager { status: finalMetaStatus, replayBufferFlushed, }) - if (!executionError) { - executionError = new Error(TERMINAL_PUBLISH_ERROR) - } + await degradeTerminalPublish( + finalMetaStatus, + new Error('Terminal event was never published') + ) } else { await eventWriter.close().catch((error) => { logger.warn('Failed to close resume event writer after terminal publish', { @@ -1707,6 +1736,13 @@ export class PauseResumeManager { */ await loggingSession.waitForPostExecution() + if (terminalPublishDegraded) { + logger.warn('Resume settled with a degraded run buffer', { + resumeExecutionId, + status: finalMetaStatus, + }) + } + if (executionError || !result) { throw executionError ?? new Error('Resume execution did not produce a result') }