Skip to content

Commit 017dca5

Browse files
committed
fix(executor): resume payload compaction on a fresh stack every 256 levels
The bottom-up compaction from #9316 walks a tree on one synchronous stack until something spills, so deep trees overflow the stack where the old awaiting walk compacted them. Every 256 nesting levels the walk now resumes from a microtask; later siblings wait for the resumed subtree to finish walking, so objects are first seen in the same order and output and stored values stay identical.
1 parent 053c933 commit 017dca5

2 files changed

Lines changed: 126 additions & 22 deletions

File tree

‎apps/sim/lib/execution/payloads/serializer.test.ts‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -336,6 +336,24 @@ describe('compactExecutionPayload', () => {
336336
expect(stripped.other).toBe(withBase64.other)
337337
})
338338

339+
it('compacts trees nested deeper than one synchronous stack can walk', async () => {
340+
const nest = (depth: number, level: (child: unknown, index: number) => unknown) => {
341+
let value: unknown = 'leaf'
342+
for (let index = 0; index < depth; index++) value = level(value, index)
343+
return value
344+
}
345+
// Both overflow an unbroken synchronous walk even on Vitest's larger worker stack.
346+
const plain = nest(10_000, (child) => ({ child }))
347+
// A Date has a toJSON, so every level serializes its subtree from where the walk stands.
348+
const datePerLevel = nest(8_000, (child, index) => ({ at: new Date(index), child }))
349+
const datePerLevelCompacted = nest(8_000, (child) => ({ at: {}, child }))
350+
351+
expect(JSON.stringify(await compactExecutionPayload(plain))).toBe(JSON.stringify(plain))
352+
expect(JSON.stringify(await compactExecutionPayload(datePerLevel))).toBe(
353+
JSON.stringify(datePerLevelCompacted)
354+
)
355+
})
356+
339357
it('rejects durable compaction when storage context is incomplete', async () => {
340358
await expect(
341359
compactExecutionPayload(

‎apps/sim/lib/execution/payloads/serializer.ts‎

Lines changed: 108 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -213,24 +213,60 @@ function measuredNode(value: unknown): CompactedNode {
213213
return toNode(value, measureJson(value))
214214
}
215215

216+
/**
217+
* Nesting levels walked on one synchronous stack before the walk resumes on a
218+
* fresh one. Each level costs several frames, and a level with a `toJSON`
219+
* serializes its subtree from where it stands, so an unbroken walk overflows
220+
* the stack on trees an awaiting walk handles.
221+
*/
222+
const RESUME_WALK_EVERY_DEPTH = 256
223+
224+
/**
225+
* A subtree whose walk continues on a fresh stack. `walked` settles once the
226+
* walk has visited every object below it, with its result (still pending while
227+
* a spill is in flight). Later siblings start only after it, so objects are
228+
* first seen in the same order as an unbroken walk.
229+
*/
230+
class ResumedWalk<T> {
231+
constructor(readonly walked: Promise<{ result: MaybePromise<T> }>) {}
232+
}
233+
234+
/** The result of walking a subtree: ready, pending on a spill, or still walking. */
235+
type Walk<T> = MaybePromise<T> | ResumedWalk<T>
236+
237+
/** Unwraps a walk into the promise of its result. */
238+
async function settleWalk<T>(walk: Walk<T>): Promise<T> {
239+
return walk instanceof ResumedWalk ? (await walk.walked).result : walk
240+
}
241+
216242
function thenMaybe<T, R>(
217243
value: MaybePromise<T>,
218244
next: (resolved: T) => MaybePromise<R>
219245
): MaybePromise<R> {
220246
return value instanceof Promise ? value.then(next) : next(value)
221247
}
222248

249+
function thenWalk<T, R>(walk: Walk<T>, next: (resolved: T) => MaybePromise<R>): Walk<R> {
250+
return walk instanceof ResumedWalk
251+
? new ResumedWalk(walk.walked.then(({ result }) => ({ result: thenMaybe(result, next) })))
252+
: thenMaybe(walk, next)
253+
}
254+
223255
/**
224256
* Starts every step in order and joins them like `Promise.all`, staying
225257
* synchronous when none is async. A step that throws synchronously still
226-
* observes the steps already pending, so none rejects unhandled.
258+
* observes the steps already pending, so none rejects unhandled. A step whose
259+
* walk resumed holds back the rest until it has walked.
227260
*/
228-
function startAll<T>(count: number, start: (index: number) => MaybePromise<T>): MaybePromise<T[]> {
261+
function startAll<T>(count: number, start: (index: number) => Walk<T>): Walk<T[]> {
229262
const results = new Array<MaybePromise<T>>(count)
230263
let hasPending = false
231264
try {
232265
for (let index = 0; index < count; index++) {
233266
const result = start(index)
267+
if (result instanceof ResumedWalk) {
268+
return new ResumedWalk(continueAll(result, index, count, start, results))
269+
}
234270
results[index] = result
235271
if (result instanceof Promise) hasPending = true
236272
}
@@ -241,31 +277,56 @@ function startAll<T>(count: number, start: (index: number) => MaybePromise<T>):
241277
return hasPending ? Promise.all(results) : (results as T[])
242278
}
243279

280+
async function continueAll<T>(
281+
resumed: ResumedWalk<T>,
282+
resumedIndex: number,
283+
count: number,
284+
start: (index: number) => Walk<T>,
285+
results: MaybePromise<T>[]
286+
): Promise<{ result: MaybePromise<T[]> }> {
287+
let hasPending = false
288+
/** Observes each pending step now, as a later step may throw before the join. */
289+
const accept = (result: MaybePromise<T>, index: number) => {
290+
results[index] = result
291+
if (result instanceof Promise) {
292+
hasPending = true
293+
result.catch(() => {})
294+
}
295+
}
296+
for (let index = 0; index < resumedIndex; index++) accept(results[index], index)
297+
accept((await resumed.walked).result, resumedIndex)
298+
for (let index = resumedIndex + 1; index < count; index++) {
299+
const result = start(index)
300+
accept(result instanceof ResumedWalk ? (await result.walked).result : result, index)
301+
}
302+
return { result: hasPending ? Promise.all(results) : (results as T[]) }
303+
}
304+
244305
/** Runs `start` for each index in order, awaiting only when a step is async. */
245306
function forEachInOrder<T>(
246307
count: number,
247-
start: (index: number) => MaybePromise<T>,
308+
start: (index: number) => Walk<T>,
248309
accept: (result: T, index: number) => void
249310
): MaybePromise<void> {
250311
for (let index = 0; index < count; index++) {
251312
const result = start(index)
252-
if (result instanceof Promise) {
313+
if (result instanceof Promise || result instanceof ResumedWalk) {
253314
return continueInOrder(result, index, count, start, accept)
254315
}
255316
accept(result, index)
256317
}
257318
}
258319

259320
async function continueInOrder<T>(
260-
pending: Promise<T>,
321+
pending: Promise<T> | ResumedWalk<T>,
261322
pendingIndex: number,
262323
count: number,
263-
start: (index: number) => MaybePromise<T>,
324+
start: (index: number) => Walk<T>,
264325
accept: (result: T, index: number) => void
265326
): Promise<void> {
266-
accept(await pending, pendingIndex)
327+
accept(await settleWalk(pending), pendingIndex)
267328
for (let index = pendingIndex + 1; index < count; index++) {
268-
accept(await start(index), index)
329+
accept(await settleWalk(start(index)), index)
269330
}
270331
}
271332

@@ -293,7 +354,7 @@ function compactValue(
293354
options: CompactExecutionPayloadOptions,
294355
state: CompactState,
295356
depth = 0
296-
): MaybePromise<CompactedNode> {
357+
): Walk<CompactedNode> {
297358
const thresholdBytes = options.thresholdBytes ?? LARGE_VALUE_THRESHOLD_BYTES
298359
if (!value || typeof value !== 'object') {
299360
const measured = measureLeaf(value, thresholdBytes)
@@ -332,7 +393,11 @@ function compactValue(
332393
}
333394
state.seen.add(value)
334395

335-
return thenMaybe(compactEntries(value, options, state, depth), ({ compacted, measured }) => {
396+
const children =
397+
depth > 0 && depth % RESUME_WALK_EVERY_DEPTH === 0
398+
? resumeEntries(value, options, state, depth)
399+
: compactEntries(value, options, state, depth)
400+
return thenWalk(children, ({ compacted, measured }) => {
336401
if (!exceedsThreshold(measured.bytes, thresholdBytes)) {
337402
return toNode(compacted, measured)
338403
}
@@ -354,13 +419,31 @@ function compactValue(
354419
})
355420
}
356421

422+
/**
423+
* Walks a container's entries on a fresh stack. Kept out of `compactValue` so
424+
* that function allocates no closure on every call.
425+
*/
426+
function resumeEntries(
427+
value: object,
428+
options: CompactExecutionPayloadOptions,
429+
state: CompactState,
430+
depth: number
431+
): ResumedWalk<CompactedChildren> {
432+
return new ResumedWalk(
433+
Promise.resolve().then(() => {
434+
const result = compactEntries(value, options, state, depth)
435+
return result instanceof ResumedWalk ? result.walked : { result }
436+
})
437+
)
438+
}
439+
357440
function compactEntryValue(
358441
key: string,
359442
entryValue: unknown,
360443
options: CompactExecutionPayloadOptions,
361444
state: CompactState,
362445
depth: number
363-
): MaybePromise<CompactedNode> {
446+
): Walk<CompactedNode> {
364447
return key === 'finalBlockLogs' && Array.isArray(entryValue)
365448
? compactBlockLogs(entryValue as BlockLog[], options).then(measuredNode)
366449
: compactValue(entryValue, options, state, depth + 1)
@@ -371,7 +454,7 @@ function compactEntries(
371454
options: CompactExecutionPayloadOptions,
372455
state: CompactState,
373456
depth: number
374-
): MaybePromise<CompactedChildren> {
457+
): Walk<CompactedChildren> {
375458
if (options.rejectLargeValues) {
376459
return compactEntriesWithEarlyReject(value, options, state, depth)
377460
}
@@ -380,7 +463,7 @@ function compactEntries(
380463
const pending = startAll(value.length, (index) =>
381464
index in value ? compactValue(value[index], options, state, depth + 1) : OMITTED_NODE
382465
)
383-
return thenMaybe(pending, (nodes) => {
466+
return thenWalk(pending, (nodes) => {
384467
const compacted =
385468
state.reuseUnchanged &&
386469
Object.getPrototypeOf(value) === Array.prototype &&
@@ -395,7 +478,7 @@ function compactEntries(
395478
const pending = startAll(entries.length, (index) =>
396479
compactEntryValue(entries[index][0], entries[index][1], options, state, depth)
397480
)
398-
return thenMaybe(pending, (nodes) => {
481+
return thenWalk(pending, (nodes) => {
399482
const compacted =
400483
state.reuseUnchanged &&
401484
Object.getPrototypeOf(value) === Object.prototype &&
@@ -498,7 +581,8 @@ export async function compactExecutionPayload<T>(
498581
value: T,
499582
options: CompactExecutionPayloadOptions = {}
500583
): Promise<T> {
501-
return (await compactValue(value, options, { seen: new WeakSet<object>() })).value as T
584+
return (await settleWalk(compactValue(value, options, { seen: new WeakSet<object>() })))
585+
.value as T
502586
}
503587

504588
export async function compactWorkflowVariableValue<T>(
@@ -516,13 +600,15 @@ export async function compactSubflowResults<T>(
516600
options: CompactExecutionPayloadOptions = {}
517601
): Promise<T[]> {
518602
const entryOptions = { ...options, preserveRoot: false }
519-
const nodes = await startAll(results.length, (index) =>
520-
index in results
521-
? compactValue(results[index], entryOptions, {
522-
seen: new WeakSet<object>(),
523-
reuseUnchanged: true,
524-
})
525-
: OMITTED_NODE
603+
const nodes = await settleWalk(
604+
startAll(results.length, (index) =>
605+
index in results
606+
? compactValue(results[index], entryOptions, {
607+
seen: new WeakSet<object>(),
608+
reuseUnchanged: true,
609+
})
610+
: OMITTED_NODE
611+
)
526612
)
527613
let compactedResults = nodes.map((node) => node.value) as T[]
528614

0 commit comments

Comments
 (0)