Add additional tests for event consumer fixes for hook/sleep/step race conditions#1528
Add additional tests for event consumer fixes for hook/sleep/step race conditions#1528
Conversation
During replay of a `for await (const payload of hook) { await step() }`
pattern, the EventsConsumer could advance to a step_created event before
the workflow code registered the step consumer. The existing deferred
unconsumed check chained onto the promise queue once and waited 100ms,
but this missed a second round of async work (hook payload deserialization)
triggered by the first drain's resolve().
Fix: after the initial queue drain, yield to the event loop (setTimeout(0))
so microtask chains propagate (e.g., step resolve → for-await resumes →
createHookPromise → new deserialization), then re-chain onto the latest
queue before starting the 100ms timeout.
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
|
📊 Benchmark Results
workflow with no steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Next.js (Turbopack) workflow with 1 step💻 Local Development
▲ Production (Vercel)
🔍 Observability: Next.js (Turbopack) workflow with 10 sequential steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Next.js (Turbopack) workflow with 25 sequential steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Next.js (Turbopack) workflow with 50 sequential steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Next.js (Turbopack) Promise.all with 10 concurrent steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Next.js (Turbopack) Promise.all with 25 concurrent steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Next.js (Turbopack) Promise.all with 50 concurrent steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Next.js (Turbopack) Promise.race with 10 concurrent steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Next.js (Turbopack) Promise.race with 25 concurrent steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Next.js (Turbopack) Promise.race with 50 concurrent steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Next.js (Turbopack) workflow with 10 sequential data payload steps (10KB)💻 Local Development
▲ Production (Vercel)
🔍 Observability: Next.js (Turbopack) workflow with 25 sequential data payload steps (10KB)💻 Local Development
▲ Production (Vercel)
🔍 Observability: Next.js (Turbopack) workflow with 50 sequential data payload steps (10KB)💻 Local Development
▲ Production (Vercel)
🔍 Observability: Next.js (Turbopack) workflow with 10 concurrent data payload steps (10KB)💻 Local Development
▲ Production (Vercel)
🔍 Observability: Next.js (Turbopack) workflow with 25 concurrent data payload steps (10KB)💻 Local Development
▲ Production (Vercel)
🔍 Observability: Next.js (Turbopack) workflow with 50 concurrent data payload steps (10KB)💻 Local Development
▲ Production (Vercel)
🔍 Observability: Next.js (Turbopack) Stream Benchmarks (includes TTFB metrics)workflow with stream💻 Local Development
▲ Production (Vercel)
🔍 Observability: Next.js (Turbopack) stream pipeline with 5 transform steps (1MB)💻 Local Development
▲ Production (Vercel)
🔍 Observability: Next.js (Turbopack) 10 parallel streams (1MB each)💻 Local Development
▲ Production (Vercel)
🔍 Observability: Next.js (Turbopack) fan-out fan-in 10 streams (1MB each)💻 Local Development
▲ Production (Vercel)
🔍 Observability: Next.js (Turbopack) SummaryFastest Framework by WorldWinner determined by most benchmark wins
Fastest World by FrameworkWinner determined by most benchmark wins
Column Definitions
Worlds:
❌ Some benchmark jobs failed:
Check the workflow run for details. |
🧪 E2E Test Results❌ Some tests failed Summary
❌ Failed Tests▲ Vercel Production (1 failed)vite (1 failed):
Details by Category❌ ▲ Vercel Production
✅ 💻 Local Development
✅ 📦 Local Production
✅ 🐘 Local Postgres
✅ 🪟 Windows
✅ 📋 Other
❌ Some E2E test jobs failed:
Check the workflow run for details. |
There was a problem hiding this comment.
Pull request overview
Fixes a replay-time race in EventsConsumer where step_created could be incorrectly treated as orphaned in for await (const payload of hook) { await step(payload) } patterns, by adding an additional event-loop yield + promise-queue re-check before triggering the deferred “unconsumed event” failure path.
Changes:
- Update
EventsConsumer’s deferred unconsumed-event check to drain the promise queue, yield to the event loop, then drain again before scheduling the final delayed check. - Add a unit test reproducing the CI event-log pattern (alternating
hook_receivedand step lifecycle events) for the hook + sleep + per-payload step scenario. - Add a patch changeset for
@workflow/core.
Reviewed changes
Copilot reviewed 3 out of 3 changed files in this pull request and generated 2 comments.
| File | Description |
|---|---|
| packages/core/src/events-consumer.ts | Adds a second promise-queue drain after a setTimeout(0) yield to avoid false-positive unconsumed events during replay. |
| packages/core/src/hook-sleep-interaction.test.ts | Adds regression coverage for for await hook loops that call a step per payload while a sleep is pending. |
| .changeset/fix-unconsumed-event-hook-step-loop.md | Declares a patch release for the core fix. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| .then( | ||
| // Yield to the event loop after the first queue drain. This allows | ||
| // microtask chains triggered by the preceding resolve() (e.g., a | ||
| // step result delivery that resumes a for-await loop, which then | ||
| // calls createHookPromise and appends a second round of async work | ||
| // to the promise queue) to propagate before we re-check the queue. | ||
| () => new Promise<void>((resolve) => setTimeout(resolve, 0)) | ||
| ) | ||
| .then(() => this.getPromiseQueue()) | ||
| .then(() => { |
There was a problem hiding this comment.
The deferred check promise chain always schedules the extra timer(s) even if a later subscribe() has already invalidated checkVersion. This can leave behind unnecessary timers (including the new setTimeout(0) yield) under high churn, keeping the event loop alive and doing extra work even though the check can never fire. Consider short‑circuiting before scheduling the setTimeout (and/or before the second getPromiseQueue() call) when this.unconsumedCheckVersion !== checkVersion so cancellation is cheap and no timers are created for stale checks.
| .then( | |
| // Yield to the event loop after the first queue drain. This allows | |
| // microtask chains triggered by the preceding resolve() (e.g., a | |
| // step result delivery that resumes a for-await loop, which then | |
| // calls createHookPromise and appends a second round of async work | |
| // to the promise queue) to propagate before we re-check the queue. | |
| () => new Promise<void>((resolve) => setTimeout(resolve, 0)) | |
| ) | |
| .then(() => this.getPromiseQueue()) | |
| .then(() => { | |
| .then(() => { | |
| // If a newer subscribe() has already invalidated this check, bail out | |
| // before scheduling the extra yield timer. | |
| if (this.unconsumedCheckVersion !== checkVersion) { | |
| return; | |
| } | |
| // Yield to the event loop after the first queue drain. This allows | |
| // microtask chains triggered by the preceding resolve() (e.g., a | |
| // step result delivery that resumes a for-await loop, which then | |
| // calls createHookPromise and appends a second round of async work | |
| // to the promise queue) to propagate before we re-check the queue. | |
| return new Promise<void>((resolve) => setTimeout(resolve, 0)); | |
| }) | |
| .then(() => { | |
| // Short-circuit stale checks before performing a second queue drain. | |
| if (this.unconsumedCheckVersion !== checkVersion) { | |
| return; | |
| } | |
| return this.getPromiseQueue(); | |
| }) | |
| .then(() => { | |
| // If a later subscribe() has invalidated this check by the time both | |
| // queue drains (and the yield) have completed, do not schedule the | |
| // final timeout at all. | |
| if (this.unconsumedCheckVersion !== checkVersion) { | |
| return; | |
| } |
| const ctx = setupWorkflowContext([ | ||
| { | ||
| eventId: 'evnt_0', | ||
| runId: 'wrun_test', | ||
| eventType: 'hook_created', | ||
| correlationId: `hook_${CORR_IDS[0]}`, | ||
| eventData: { token: 'test-token', isWebhook: false }, | ||
| createdAt: new Date(), |
There was a problem hiding this comment.
This test is intended to ensure an "unconsumed event" error does not occur, but setupWorkflowContext() configures EventsConsumer with onUnconsumedEvent: () => {}. That means the regression would manifest as a hang until the test times out (and it won’t surface the same error the real runtime throws). To make the regression signal deterministic and match production, consider letting the test pass an onUnconsumedEvent handler that rejects via ctx.onWorkflowError (or throws) so a failure is immediate and clearly attributed to the unconsumed-event path.
|
Double checked this with an agent. I'm adding another e2e test to this PR and then will attempt to ship |
Signed-off-by: Peter Wielander <mittgfu@gmail.com>
TooTallNate
left a comment
TooTallNate
left a comment
There was a problem hiding this comment.
Review
Good intent — adding multiple angles of regression coverage for the fix from #1778. The e2e test and the test-harness fix in hook-sleep-interaction.test.ts are particularly valuable. I have one blocking issue and some observations.
Blocking
The new test in workflow.test.ts imports DEFERRED_CHECK_DELAY_MS from ./events-consumer.js, but that export doesn't exist (neither on main nor on this branch). The "add more tests" commit (4701796) introduced this export, but the "WIP undo" commit (06a3f42) reverted the events-consumer.ts changes and left the test dangling.
At runtime, DEFERRED_CHECK_DELAY_MS is undefined, so DEFERRED_CHECK_DELAY_MS + 50 is NaN, and setTimeout(r, NaN) coerces to setTimeout(r, 0). The test passes, but it's not actually exercising the case it claims to — "hydrate latency exceeds the deferred-check window." It's running the happy path with near-zero hydrate delay.
See inline comment for fix options.
Non-blocking observations
-
hook-sleep-interaction.test.tsharness fix is the most valuable change in this PR. The prioronUnconsumedEvent: () => {}silently swallowed errors, meaning any existing test in that file that triggered this error would have silently passed. Wiring it throughctx.onWorkflowErrormeans existing tests now actually catch false-positive unconsumed events. This is a meaningful test infrastructure improvement. -
e2e test is good — verifies the fix through real queue/world infrastructure, not just unit-level mocks. The 5s + 3s sleeps between operations are necessary for the replay timing but slightly fragile; worth monitoring for flakes.
-
Overlap with #1778's test: the existing test in #1778 already covers the alternating
hook_received+ step-lifecycle pattern with 150ms hydrate delay on call #3. The newhook-sleep-interaction.test.tstest covers similar territory but at the orchestrator-context level rather thanrunWorkflowlevel — a useful lower-level variant. Not duplicative.
| // production this corresponds to slower-than-expected encrypted payload | ||
| // decryption (cold cache, contended CPU, etc.); the fix must not rely on | ||
| // hydrate completing within the timer window. | ||
| const hydrateDelayMs = DEFERRED_CHECK_DELAY_MS + 50; |
There was a problem hiding this comment.
Blocking: DEFERRED_CHECK_DELAY_MS is imported but doesn't exist, so this test isn't actually exercising slow hydrate.
- Line 6 imports
DEFERRED_CHECK_DELAY_MSfrom./events-consumer.js events-consumer.tson this branch (and onmain) does not exportDEFERRED_CHECK_DELAY_MS— the 100ms value is an inline magic number atevents-consumer.ts:141- At runtime:
DEFERRED_CHECK_DELAY_MSisundefined,undefined + 50 = NaN,setTimeout(r, NaN)behaves assetTimeout(r, 0) - Net effect: the test claims to mock hydrate with
> 100msdelay but actually mocks it with0msdelay
So this test runs, passes, but isn't testing what the comment claims. The "WIP undo" commit removed the DEFERRED_CHECK_DELAY_MS export from events-consumer.ts (added in the "add more tests" commit) but left the import here.
Fix options:
-
Re-export
DEFERRED_CHECK_DELAY_MSfromevents-consumer.ts— this is the cleanest and also improves the production code by replacing the magic100atevents-consumer.ts:141with a named constant. Since the fix already shipped in fix false-positive unconsumed event in hook loop replay #1778, this would be a small follow-up enhancement worth including in this test-only PR. -
Hardcode the delay — e.g.
const hydrateDelayMs = 150. Simplest, but couples the test to an implementation constant.
Option 1 is better. Recommend adding back:
// packages/core/src/events-consumer.ts
export const DEFERRED_CHECK_DELAY_MS = 100;
// ...
this.pendingUnconsumedTimeout = setTimeout(() => {
// ...
}, DEFERRED_CHECK_DELAY_MS);
TooTallNate
left a comment
TooTallNate
left a comment
There was a problem hiding this comment.
Blocking issue from prior review is resolved. The 'fix test' commit (5279e839) replaces the phantom DEFERRED_CHECK_DELAY_MS import with local constants:
const DEFFERED_CHECK_DELAY_MS = 100;
const BUFFER_FOR_TEST = 50;
const hydrateDelayMs = DEFFERED_CHECK_DELAY_MS + BUFFER_FOR_TEST; // 150mshydrateDelayMs is now a real 150, which exceeds the actual 100ms deferred-check window hardcoded in events-consumer.ts:141. The test genuinely exercises the slow-hydrate race condition now.
Minor nit (non-blocking): DEFFERED is misspelled twice — should be DEFERRED. Worth fixing while you're in there. If/when the constant is eventually promoted to a proper export in events-consumer.ts (my earlier preferred option), the test can swap these locals out.
Otherwise LGTM — the test harness improvement in hook-sleep-interaction.test.ts remains the most valuable piece here, and the e2e test plus the additional workflow.test.ts case give good coverage of the race from multiple angles.
Fix shipped in #1778, so this PR now only adds tests, see old description below
Summary
EventsConsumerwhere astep_createdevent could be falsely flagged as unconsumed during replay offor await (const payload of hook) { await step(payload); }patternsRoot Cause
During replay, the
EventsConsumeradvances through events viaprocess.nextTick. When it reaches astep_createdevent for the second loop iteration, no step consumer exists yet because:step_completeddeserialization resolvesprocessPayload()→ triggers for-await loop continuationcreateHookPromise()→ chains hook payload deserialization onto the promise queueprocessPayload()called → step consumer registeredThe existing deferred check only chained onto the promise queue once, missing the second round of async work added in step 2. The fix adds a
setTimeout(0)yield between queue drains to let microtask chains propagate.CI Failure
wrun_01KMP440KF1D1GKE7SSQPGXQGV— failed withUnconsumed event in event log: eventType=step_createdTest plan
events-consumer.test.tstests passhook-sleep-interaction.test.tstests pass (including new test)workflow.test.tstests pass🤖 Generated with Claude Code