[core] Gate the unconsumed-event check on delivery idleness - #3439
Conversation
The events consumer walks the log synchronously, but the resolutions that walk triggers do not resolve synchronously: a step result hydrates in the host, resolves from a detached continuation behind `awaitEarlierDeliveries`, and only then does VM code run far enough to subscribe the consumer for the next event. The walk therefore routinely sits on an ordered event that nobody has claimed yet while the workflow is mid-flight on its way to claiming it. The deferred unconsumed-event check resolved that with a fixed `DEFERRED_CHECK_DELAY_MS` after the promise queue drained, which is a bet that every delivery lands inside the window. Replaying a batch of N parallel step results loses it: the queue drains with N-1 still on the detached path, and the check raises `ReplayDivergenceError` against a log the same replay goes on to reproduce exactly. Measured on the event-log race repro against world-postgres, on identical event logs: 0 of 114 runs corrupted at a 100ms window, 34 of 42 at 10ms, with the no-parallel-delivery control clean at both. `hasParkedCommittedDelivery` already documents this hazard for the suspension path, where `scheduleWhenIdle` guards it by polling. Export that predicate as `isDeliveryIdle`, thread it into `EventsConsumer`, and poll it before starting the delay timer. Termination is inherited: it counts only deliveries that resolve on their own, so nothing can gate its own retirement, and a genuinely orphaned event reaches the check on the first poll.
🦋 Changeset detectedLatest commit: ec5c121 The changes in this PR will be included in the next version bump. This PR includes changesets to release 16 packages
Not sure what this means? Click here to learn what changesets are. Click here if you're a maintainer who wants to add another changeset to this PR |
🧪 E2E Test Results❌ Some tests failed ❌ Failed E2E Tests▲ Vercel Production (1 failed)sveltekit-node (1 failed):
E2E Test SummarySummary
Details by Category❌ ▲ Vercel Production
✅ 💻 Local Development
✅ 📦 Local Production
✅ 🐘 Local Postgres
✅ 🪟 Windows
✅ vercel-multi-region
|
📊 Workflow Benchmarkscommit Backend:
📈 STSO distribution vs main (inline / queue-hop histograms)1020 steps (inline) Cumulative STSO time: main 140719ms → this run 155571ms (Δ +14852ms, +11%) ℹ️ Metric definitions & methodologyThe collapsed STSO distribution section above buckets every step gap of the sequential-steps run (not a sampled window), split by whether the step ending the gap ran inline — in the same warm process as the step before it, so the gap is pure framework overhead — or after a queue-hop — the first step of a fresh process, which pays queue dispatch, client reinit and event-log replay. Bars overlay the two runs: Best/P75/P90/P99 deltas compare against the most recent benchmark run on Metrics — TTFS: time to first step body (in-deployment start() → first step body, deployment clocks) · STSO: step-to-step overhead (gap between consecutive step bodies) · WO: workflow overhead (whole-run time outside step bodies, in-deployment anchored) · SL: stream latency (in-deployment write → read propagation, readAt - writtenAt) · SO: stream overhead (end-to-end write+consume time beyond the modelled generation window) Scenarios — step: one trivial no-op step, no stream; no hooks, so the run stays in turbo mode (in-process fast path) · stream: one streaming step; no hooks, so the run stays in turbo mode (in-process fast path) · hook + stream: registers a hook before one step, which exits turbo mode (dispatch path) · 1020 steps: 1020 trivial sequential steps; STSO is measured between consecutive steps in the given step ranges, and WO is the whole-run overhead outside step bodies · stream latency: parallel reader/writer steps on a dedicated stream; SL is the in-deployment write->read propagation (readAt - writtenAt) · stream overhead (text): writer streams 300 variable-length text token deltas paced at 100/s for 3s (a haiku-size LLM's token throughput) while a parallel reader drains the whole stream; SO is the end-to-end write+consume time beyond the 3s generation window (overhead/backpressure) · stream overhead (structured): same workload as stream overhead (text), but each delta is an AI-SDK-style structured object ({ type: 'text-delta', id, text }) instead of a raw string, so the SO gap vs the text scenario is the added serialization cost 🔴 marks a percentile over its target (within target is left unmarked). Targets (p75/p90/p99, ms) — TTFS 200/300/600 · SL 50/60/125 · SO 250/500/1000 All metrics are measured from deployment-side timestamps only. Runs are triggered by an in-deployment route that stamps the anchor ( Cold starts are kept in the numbers on purpose — they are part of real bursty-workload latency. The workbench deployment cold-starts the |
VaguelySerious
added
the
event-log-race-repro
Event Log Race ReproNo event-log regressions in the latest repro job. Run History
Latest Scenario Breakdown
|
| * read it, for the two such decisions: {@link scheduleWhenIdle} for the | ||
| * suspension, and the events consumer's unconsumed-event check for divergence. | ||
| */ | ||
| export function isDeliveryIdle(ctx: WorkflowOrchestratorContext): boolean { |
There was a problem hiding this comment.
This file is a no-op except for exporting this helper
| // Wait out any delivery still in flight before starting the timer. | ||
| // The queue draining says the host has no hydration work left; it | ||
| // does not say the VM has finished reacting to what was hydrated. | ||
| this.whenDeliveryIdle(checkVersion, () => { |
There was a problem hiding this comment.
Changes in this file just wrap the existing code block for this.pendingUnconsumedTimeout in this.whenDeliveryIdle(checkVersion ,() => {
VaguelySerious
left a comment
VaguelySerious
left a comment
There was a problem hiding this comment.
AI review: no blocking issues
| * Termination is `hasParkedCommittedDelivery`'s: it counts only deliveries | ||
| * that resolve on their own, so nothing here can gate its own retirement. A | ||
| * genuinely orphaned event has no delivery to wait on and reaches `fn` on the | ||
| * first poll. |
There was a problem hiding this comment.
AI Review: Note
Termination holds, but the check stops being a divergence backstop whenever a delivery is in flight, and this paragraph reads as though it still is.
Both decisions now wake from the same isDeliveryIdle edge, with different post-idle timing: scheduleWhenIdle fires on the first timer tick after idle, this check waits a further DEFERRED_CHECK_DELAY_MS. So the suspension always wins, and a pending sleep() arms one on every replay. onWorkflowError's 'suspended' branch then discards what arrives second (state = { type: 'replay' }, nothing surfaced, the interruption was already rejected with the suspension). For a genuinely diverged log with an armed delivery in flight the outcome is therefore "suspend, then demote a later resume to cold replay", not ReplayDivergenceError.
Measured on one context driving both decisions with one armed registerDeliveryBarrier, delay floored to 10ms:
| before delivery lands | after | |
|---|---|---|
| gate off | [divergence] |
[divergence, suspension] |
| gate on | [] |
[suspension, divergence] |
Not a regression to fix here: pre-PR the suspension already won whenever the delivery landed inside the 100ms window, so this determinizes an outcome that was timing-dependent. But "a genuinely orphaned event has no delivery to wait on and reaches fn on the first poll" is only the no-delivery case. Worth a sentence saying that when a delivery is in flight, the suspension gets there first and the divergence this eventually reports is dropped, so nothing should treat the check as the mechanism that catches a diverged log.
| * VM is about to draw and the window is the only thing standing between a | ||
| * healthy run and `ReplayDivergenceError`. On a backend whose deliveries take | ||
| * longer than the window, that bet loses: the local race repro corrupts 34 of | ||
| * 42 runs at a 10ms window and 0 of 114 at 100ms, on the same event logs. |
There was a problem hiding this comment.
AI Review: Note
These numbers establish the mechanism, not the ship decision. 0 of 114 at the default 100ms and 34 of 42 at 10ms says a too-short window manufactures divergence; it does not show the default window being lost. The claim actually carrying the change is that world-vercel deliveries outrun 100ms, and neither the comment nor the PR body has data behind it.
Either cite the world-vercel evidence, or drop the implication and describe this as hardening the check against a window it is not currently observed to lose. As written, someone tuning WORKFLOW_DEFERRED_CHECK_DELAY_MS later will read this table as proof the default is marginal.
| * Defaults to always-idle so the tests that drive a consumer with no | ||
| * orchestrator context keep the pre-existing timing. | ||
| */ | ||
| isDeliveryIdle?: () => boolean; |
There was a problem hiding this comment.
AI Review: Nit
The optional-with-always-idle default means a future second construction site silently opts out of the fix, and silently, because always-idle is exactly the pre-PR behavior. There is one production site today (workflow.ts:396), so making this required and having the unit tests pass () => true explicitly costs a few lines and removes that failure mode.
| } | ||
| // Held in the same field the fired check uses so subscribe() cancels a | ||
| // poll in progress exactly as it cancels the check itself. | ||
| this.pendingUnconsumedTimeout = setTimeout(poll, 0); |
There was a problem hiding this comment.
AI Review: Nit
pendingUnconsumedTimeout is now written by two state machines, and poll() nulls it before checking its version, so a stale poll can clear the live chain's handle and leave its timer unclearable by subscribe().
I tested this rather than guessing: stacked append() calls while a check is parked on the gate, then a late subscribe(). No double-fire, and no report for the claimed event. The version guard covers it, so this is cosmetic. Noting only because the comment says the poll is held here so subscribe() cancels it "exactly as it cancels the check itself", and the two are not quite equivalent: for the poll the clearTimeout is redundant and the version bump is what does the work.
| it('does not declare divergence while a step delivery is outstanding', async () => { | ||
| // Far shorter than the delivery below, so the run survives only if the | ||
| // check waits for the delivery rather than for the clock. | ||
| vi.stubEnv('WORKFLOW_DEFERRED_CHECK_DELAY_MS', '10'); |
There was a problem hiding this comment.
AI Review: Nit
'10' duplicates the un-exported min: 10 floor in getDeferredCheckDelayMs, and the 250ms / 50ms waits below are bare numbers. Raise the floor and these tests keep passing while quietly no longer testing what they claim: the stub clamps up, the delay stops being shorter than the delivery, and the negative assertion holds for the wrong reason. Exporting the floor and deriving the waits from it keeps the tests honest when the knob moves.
TooTallNate
left a comment
TooTallNate
left a comment
There was a problem hiding this comment.
Reviewed at ec5c121 (based on current main, 0 behind). I came to this from the Slack thread's "hard to reason about" warning, so I tried to break it rather than confirm it — and it holds.
Verified locally:
- Full core suite green: 93 files, 2022 passed / 3 expected fail; the new tests also pass under
WORKFLOW_RETAINED_VM=0. - Red-test proof: swapped main's
events-consumer.ts/private.ts/workflow.tsunder this PR's test files — both in-flight cases (step delivery outstanding,payload hydrating) fail on main's code, the orphan case passes on both. The tests drive the production predicate (registerDeliveryBarrier, a realpendingDeliveriesbump), not a mock, so they pin the mechanism rather than the implementation. - Unit suites can't reproduce the live race even at a 10ms window (main passes them too) — expected, since the corruption needs real World delivery latency. The repro tables are the system-level evidence, and their design is airtight: delay-sensitivity on main (0/114 @ 100ms vs 34/42 @ 10ms) proves window-dependence; gate-on/off at 10ms (0/14 vs 9/14) proves the gate removes it rather than widening it;
hook-sleepas the no-batch control staying clean throughout is the discriminator that says the mechanism is the parallel delivery batch and nothing else.
The hard-to-reason-about parts, reasoned about:
- Termination: inherited from
hasParkedCommittedDelivery, whose self-resolving-only counting was the load-bearing invariant in #3183 and the thing #3406 made exact. Deliveries parked behind unclaimed payloads are excluded from the count, so the check can never gate the idle safety net that retires them — no mutual-gating cycle. A true orphan has nothing in flight and reaches the timer on the first poll with unchanged latency. - Detection isn't lost, only correctly deferred: a pathological run that keeps deliveries flowing postpones the check — but any genuine settling point goes through
scheduleWhenIdle, which consults the same predicate, so divergence is still raised at the next quiescence. That's the right semantics: divergence is a judgment about a quiescent VM, and judging it mid-reaction was the bug. - Cancellation: the poll's
setTimeoutlives inpendingUnconsumedTimeout, sosubscribe()cancels a poll-in-progress exactly as it cancels the armed check, and the version check covers every await gap (including a superseded timer firing after the field was nulled — it fails the version check harmlessly). - The holder wiring: nothing can run the consumer before
subscribe(), which happens after the context exists and the holder is repointed — the "idle until the context exists" comment is accurate, not hopeful. - Engine scope: QuickJS uses its own fixed-point drain loop, not
EventsConsumer, so it doesn't share this race; the single production construction site is the one wired.
The conceptual payoff deserves saying out loud: the runtime had two definitions of quiescence — scheduleWhenIdle for "is this replay over?" and a fixed timer for "did this replay go wrong?". #3183 fixed the first; this makes the second use the same answer, and isDeliveryIdle in private.ts is now the single place "in flight" is defined. That's why the diff is small and the effect is large.
One coordination note: #3389's order-tolerant consumer rewrites scheduleUnconsumedCheck (the park-or-fail decision) in this same region — the two must compose so that parking, like divergence, is only judged at delivery idleness. The Slack thread says the combination is already measured at zero corruptions on the #3389 branch; whichever merges second should carry that composition deliberately rather than as a mechanical conflict resolution.
CI: the three nextjs-webpack dev-lane failures are the long-standing HMR rebuild-count flake, sveltekit-node's webhookWorkflow is the known flaky-lane family, python-workbench is the baseline deploy failure — nothing in this PR's domain.
The clanker found it; the evidence table and the single-predicate refactor are what make it trustworthy. Approving.
Signed-off-by: Peter Wielander <mittgfu@gmail.com>
|
Backport PR opened against |
Signed-off-by: Peter Wielander <mittgfu@gmail.com>
2 participants
Problem
The events consumer walks the log synchronously, but the resolutions that walk triggers do not resolve synchronously. A step result hydrates in the host, resolves from a detached continuation behind
awaitEarlierDeliveries, and only then does VM code run far enough tosubscribe()the consumer for the next event. The walk therefore routinely sits on an ordered event (step_created,wait_created) that nobody has claimed yet, while the workflow is mid-flight on its way to claiming it.The deferred unconsumed-event check resolved that by waiting a fixed
DEFERRED_CHECK_DELAY_MS(100ms) after the promise queue drained. That is a bet that every delivery lands inside the window. Replaying a batch of N parallel step results loses the bet: the queue drains with N-1 of them still on the detached path, and the check raisesReplayDivergenceErroragainst a log the very same replay goes on to reproduce exactly. Enough of those in a row and the run ends inCorruptedEventLogError.hasParkedCommittedDeliveryalready documents this exact hazard for the suspension path (#3183), wherescheduleWhenIdleguards against it by polling. The divergence path had no such guard.Evidence
Same branch, same machine, back to back,
WORKFLOW_DEFERRED_CHECK_DELAY_MS=10in both, one line different (the consumer'sisDeliveryIdleoption wired up vs. left at its always-idle default):And on
main, varying only the delay:So the failures track the size of the window, and the gate removes the dependence on it rather than widening it: at a delay 10x shorter than the default, runs that previously failed now pass.
hook-sleepstaying clean throughout is the discriminator, since it has no parallel delivery batch and therefore nothing for the check to fire in the middle of.The diverging event type at a short window is
step_created, matching what shows up on world-vercel, where deliveries are slower than local Postgres and the 100ms window is not reliably enough either.Change
scheduleWhenIdlewas already using asisDeliveryIdle(ctx)(pendingDeliveries === 0 && !hasParkedCommittedDelivery(ctx)), and havescheduleWhenIdlecall it so there is one definition of "in flight".EventsConsumeras an option, late-bound through a holder inworkflow.tsfor the same reason the promise queue is (the consumer is built before the context).scheduleWhenIdlepolls. The existing delay stays as a residual margin once deliveries are idle.Termination is inherited from
hasParkedCommittedDelivery, which counts only deliveries that resolve on their own, so nothing here can gate its own retirement. A genuinely orphaned event has no delivery to wait on and reaches the check on the first poll.Tests
packages/core/src/unconsumed-check-delivery-idle.test.tsdrives the production predicate rather than a mock: a real armed delivery barrier fromregisterDeliveryBarrier, and separately apendingDeliveriesbump, each must hold the check off well past the (floored, 10ms) delay, and the check must still fire for an event no delivery is waiting on. Both in-flight cases fail onmain; the orphan case passes on both.Plus three unit tests in
events-consumer.test.tscovering the option in isolation, including a consumer registering during the wait and claiming the event.