Skip to content

Commit f169aff

Browse files
committed
fix(execution): offload buffered event values at a budget-aware threshold
The replay buffer keeps 1000 events per execution inside a 64 MiB budget, so a full ring only fits if events stay modest. Values were offloaded to object storage at 8 MiB, far above that, so a run emitting large block outputs exhausted its budget within a few dozen events and stayed pinned there for the rest of its life. Give buffered events their own offload threshold, sized against the observed distribution of per-event payload sizes so ordinary runs stay fully inline and only the outlier tail is offloaded. The shared LARGE_VALUE_THRESHOLD_BYTES stays at 8 MiB: executor snapshots, provider attachment limits and inline UserFile base64 all derive from it and are bounded per payload rather than accumulated across a thousand of them. Offloading costs fidelity — a ref renders as a truncated preview, and for arrays and objects as a shape summary rather than contents — which is why the threshold tracks the measured distribution instead of the stricter bound the budget arithmetic alone would suggest.
1 parent 3de63c9 commit f169aff

3 files changed

Lines changed: 54 additions & 1 deletion

File tree

apps/sim/lib/execution/event-buffer.test.ts

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -368,6 +368,28 @@ describe('execution event buffer', () => {
368368
expect(persistedEntries).toEqual([])
369369
})
370370

371+
/**
372+
* Buffered events accumulate against a 64 MiB per-execution budget across a
373+
* 1000-event ring, so a value that a one-off payload would happily inline
374+
* must still be offloaded here.
375+
*/
376+
it('offloads values far below the shared large-value threshold', async () => {
377+
mockRedis.incrby.mockResolvedValue(100)
378+
// Comfortably under LARGE_VALUE_THRESHOLD_BYTES (8 MiB), over the event one.
379+
const payload = 'x'.repeat(512 * 1024)
380+
381+
const writer = createExecutionEventWriter('exec-1', {
382+
workspaceId: 'ws-1',
383+
workflowId: 'wf-1',
384+
})
385+
await writer.write(makeEvent(payload))
386+
await writer.flush()
387+
388+
expect(persistedEntries).toHaveLength(1)
389+
const persistedBytes = Buffer.byteLength(JSON.stringify(persistedEntries[0]), 'utf8')
390+
expect(persistedBytes).toBeLessThan(256 * 1024)
391+
})
392+
371393
it('preserves requested UserFile base64 when buffering terminal events', async () => {
372394
mockRedis.incrby.mockResolvedValue(100)
373395
const base64 = Buffer.from('hello').toString('base64')

apps/sim/lib/execution/event-buffer.ts

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,10 @@ import { toError } from '@sim/utils/errors'
33
import { randomInt } from '@sim/utils/random'
44
import { env } from '@/lib/core/config/env'
55
import { getRedisClient } from '@/lib/core/config/redis'
6-
import { LARGE_VALUE_THRESHOLD_BYTES } from '@/lib/execution/payloads/large-value-ref'
6+
import {
7+
EXECUTION_EVENT_VALUE_THRESHOLD_BYTES,
8+
LARGE_VALUE_THRESHOLD_BYTES,
9+
} from '@/lib/execution/payloads/large-value-ref'
710
import { compactExecutionPayload } from '@/lib/execution/payloads/serializer'
811
import type { LargeValueStoreContext } from '@/lib/execution/payloads/store'
912
import {
@@ -301,6 +304,9 @@ async function compactEventForBuffer(
301304
executionId: context.executionId ?? event.executionId,
302305
requireDurable: context.requireDurablePayloads,
303306
preserveRoot: true,
307+
// Buffered events accumulate against a per-execution byte budget, so they
308+
// offload far earlier than a one-off payload needs to.
309+
thresholdBytes: EXECUTION_EVENT_VALUE_THRESHOLD_BYTES,
304310
}
305311

306312
let compactedData = await compactExecutionPayload(event.data, {

apps/sim/lib/execution/payloads/large-value-ref.ts

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,31 @@
11
export const LARGE_VALUE_REF_MARKER = '__simLargeValueRef'
22

33
export const LARGE_VALUE_THRESHOLD_BYTES = 8 * 1024 * 1024
4+
5+
/**
6+
* Offload threshold for values inlined into buffered execution events.
7+
*
8+
* The replay buffer keeps `EVENT_LIMIT` (1000) events per execution inside a
9+
* `MAX_EXECUTION_REDIS_BYTES` (64 MiB) budget. Values were offloaded at 8 MiB,
10+
* so a run emitting large block outputs exhausted that budget within a few
11+
* dozen events and stayed pinned at its ceiling for the rest of its life.
12+
*
13+
* Sized against the observed distribution of per-event payload sizes rather
14+
* than the budget arithmetic alone: it sits well above what a typical run
15+
* emits and well below the outlier tail, so ordinary runs stay fully inline
16+
* and only the shape that exhausts the budget is offloaded.
17+
*
18+
* Offloading is not free: a ref renders in the terminal as a truncated string
19+
* preview, or for arrays and objects as a shape summary rather than their
20+
* contents. That cost is why this tracks the measured distribution instead of
21+
* the stricter bound the budget arithmetic alone would suggest.
22+
*
23+
* Deliberately separate from {@link LARGE_VALUE_THRESHOLD_BYTES}, which stays
24+
* at 8 MiB for executor snapshots and inline UserFile base64. Those are bounded
25+
* per payload rather than accumulated across a thousand of them, and the base64
26+
* limit is derived from it arithmetically.
27+
*/
28+
export const EXECUTION_EVENT_VALUE_THRESHOLD_BYTES = 256 * 1024
429
export const LARGE_VALUE_REF_VERSION = 1
530

631
export const LARGE_VALUE_KINDS = ['array', 'object', 'string', 'json'] as const

0 commit comments

Comments
 (0)