From 20d6f49bf51fee29e3a60b19fb2817864df673ea Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sat, 1 Aug 2026 20:54:14 -0700 Subject: [PATCH 1/4] fix(resume): stop a failed run-buffer publish from stranding a resumed execution MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The run buffer is a replay convenience for stream readers; the durable execution record is authoritative. A failed terminal event publish was deciding the outcome of work that had already run: it threw, the resume was marked failed, and the paused execution was left paused forever. That path is also not retryable — the workflow had already executed — so the next resume attempt re-ran its side effects. Degrade instead. Record the terminal status on the stream meta so readers are not left polling an 'active' stream, log the failure, and let the resume settle on its real result. The same treatment applies when the terminal event is never published at all, which previously synthesized an error for the same stranding effect. Removes the now-unreachable TERMINAL_PUBLISH_ERROR constant and its branch. Pre-execution buffer failures stay fatal and retryable, since no work has happened yet. --- .../executor/human-in-the-loop-manager.ts | 60 +++++++++++++++---- 1 file changed, 48 insertions(+), 12 deletions(-) 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') } From 22f03db64857f6ab49fddf6a33038a2cceab3cd0 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sat, 1 Aug 2026 20:54:14 -0700 Subject: [PATCH 2/4] fix(execution): give per-user Redis budget keys a fixed window MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The per-user byte budget refreshed its TTL on every accepted write, so for any user who never went a full TTL without writing, the key never expired. The per-execution data it accounted for kept expiring underneath it, so the counter accrued bytes Redis had already dropped and drifted toward the ceiling. On reaching it, every subsequent write for that user was rejected until they stopped writing for a full TTL — and since a rejected write does not refresh the TTL, it recovered on its own and then refilled. User keys now get a fixed window: the TTL is set when the key is created and never extended. Applied to all five Lua scripts that touch the key so the writers cannot drift apart. Execution-scoped keys keep sliding — they are refreshed on the same schedule as the data they account for, so their counter and bytes stay in step. --- apps/sim/lib/execution/event-buffer.test.ts | 35 +++++++++ apps/sim/lib/execution/event-buffer.ts | 11 ++- .../lib/execution/redis-budget.server.test.ts | 76 +++++++++++++++++++ apps/sim/lib/execution/redis-budget.server.ts | 20 ++++- .../uploads/utils/user-file-base64.server.ts | 4 +- 5 files changed, 137 insertions(+), 9 deletions(-) create mode 100644 apps/sim/lib/execution/redis-budget.server.test.ts 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..af18490f52a 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,9 +109,6 @@ 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 - redis.call('EXPIRE', KEYS[5], budget_ttl_seconds) - end end for i = 9, #ARGV, 2 do redis.call('ZADD', KEYS[1], ARGV[i], ARGV[i + 1]) @@ -148,7 +147,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 From 4c7df926eae4f52f36b7f469ad8fd8dc0152ff67 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sat, 1 Aug 2026 21:03:28 -0700 Subject: [PATCH 3/4] fix(execution): heal a user budget key that somehow has no expiry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The no-op branch of the flush script dropped the user-key EXPIRE outright rather than guarding it like every other site. No current path can create the key without an expiry, but if one ever did, that branch was the one place that would never give it one — and a user counter with no expiry is the unbounded version of the bug this series fixes. Guard it instead, so the branch heals such a key rather than skipping it, and so all five scripts read the same way. --- apps/sim/lib/execution/event-buffer.ts | 3 +++ 1 file changed, 3 insertions(+) diff --git a/apps/sim/lib/execution/event-buffer.ts b/apps/sim/lib/execution/event-buffer.ts index af18490f52a..f2661d21334 100644 --- a/apps/sim/lib/execution/event-buffer.ts +++ b/apps/sim/lib/execution/event-buffer.ts @@ -109,6 +109,9 @@ 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 and redis.call('TTL', KEYS[5]) < 0 then + redis.call('EXPIRE', KEYS[5], budget_ttl_seconds) + end end for i = 9, #ARGV, 2 do redis.call('ZADD', KEYS[1], ARGV[i], ARGV[i + 1]) From 7bb2ca6ea6e4e6d41d66ae96c35f0550df293982 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sat, 1 Aug 2026 21:11:13 -0700 Subject: [PATCH 4/4] fix(stream): end a replay cleanly when terminal metadata has no terminal event MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. The reader required both, so the degraded case threw and turned a replay that was merely incomplete into a broken one. Metadata without a matching event means the buffer degraded, not that the run is still going. Log it and close the stream: the reader has already received every event that was buffered, and the durable record is unaffected either way. --- .../executions/[executionId]/stream/route.test.ts | 6 ++---- .../[id]/executions/[executionId]/stream/route.ts | 13 ++++++++++++- 2 files changed, 14 insertions(+), 5 deletions(-) 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() }