perf(core): path-independent stream write batching (group commit in the server writable)#3078
Conversation
…eam (group commit) Batching previously lived in flushablePipe's coalescing loop, so it only engaged on paths that used flushablePipe (getWritable). A raw ReadableStream crossing a workflow/step boundary is piped with native pipeTo(), which does not pull chunk N+1 until write(chunk N) resolves — and write() resolved only after the flush timer AND the server round trip, so the buffer never held more than one chunk and every token became its own server request. The sink now group-commits: - write() resolves when the chunk enters a bounded client buffer; the bound counts buffered AND in-request chunks (WORKFLOW_STREAM_MAX_INFLIGHT_CHUNKS, preserving its documented meaning) plus a byte bound (WORKFLOW_STREAM_MAX_BUFFERED_BYTES, new, default 8 MiB, documented in runtime-tuning). A full buffer applies backpressure until a group lands durably. - The flush interval is a real group-commit window; chunks arriving while a request is in flight accumulate and form the next writeMulti group. One request in flight at a time preserves chunk order. - Per-request wire limits (1,000 chunks / 1 MiB) split groups exactly as the coalescing pipe did; an oversized single chunk goes alone. - Durability moved to an explicit barrier (STREAM_DRAIN_SYMBOL): close() drains before closing; flushablePipe adopts the barrier so lock-release completion (step completion) still means 'everything written is durable'; abort() DRAINS the accepted prefix (never closing) so a producer error after acked writes cannot lose data — native pipeTo aborts the sink on source failure; and a failed pipe drains before settling so a step failure is not persisted ahead of the emitted prefix. A dispatch failure retains the group, poisons the sink, and surfaces at the next write/close/drain. flushablePipe is now a plain per-chunk pump responsible only for lock-release completion and durability tracking; its coalescing machinery and STREAM_WRITE_BATCH_SYMBOL are removed. Covered: native-pipeTo batching (the regression), awaited per-chunk loops coalescing into one writeMulti, in-flight accumulation, wire-cap splits (count/byte/oversized), in-flight-inclusive backpressure for both bounds, sequential fallback without writeMulti, source-error prefix delivery through abort, failed-pipe drain-before-reject, early-ack sticky errors, turbo run-ready barrier gating (incl. dwell telemetry), drain-barrier adoption/rejection, and group-level flush spans. 1,572 core unit tests pass; e2e tier requires a deployment and was not run here. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
🦋 Changeset detectedLatest commit: a57ba7c 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 |
📊 Workflow Benchmarkscommit Backend:
📜 Previous results (4)5f57e09Fri, 24 Jul 2026 05:05:30 GMT · run logs
b5718b0Fri, 24 Jul 2026 01:05:26 GMT · run logs
7e5ad94Fri, 24 Jul 2026 00:29:08 GMT · run logs
9792b58Fri, 24 Jul 2026 00:09:07 GMT · run logs
ℹ️ Metric definitions & methodologyBest/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 · STSO (1-20) 20/30/60 · STSO (101-120) 30/45/90 · STSO (1001-1020) 40/60/120 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 |
🧪 E2E Test Results❌ Some tests failed ❌ Failed E2E Tests📦 Local Production (1 failed)nextjs-turbopack-stable (1 failed):
E2E Test SummarySummary
Details by Category✅ ▲ Vercel Production
✅ 💻 Local Development
❌ 📦 Local Production
✅ 🐘 Local Postgres
✅ 🪟 Windows
✅ 📋 Other
✅ vercel-multi-region
|
![vercel[bot]](https://avatars.githubusercontent.com/in/8329?s=60&v=4){kind=small}
Review (bot): a write landing between the dispatch loop's empty-buffer exit and the reaction clearing the in-flight marker armed no timer (scheduleGroupCommit saw the marker set) and was never dispatched on an open stream — only a later write/close/drain would pick it up. The settle reaction now re-dispatches when the buffer is non-empty, treating the chunk as an in-request arrival; drain waiters settle with the new chain. Regression test aims a write at the settle gap and asserts both chunks flush without a close. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
VaguelySerious
left a comment
VaguelySerious
left a comment
There was a problem hiding this comment.
AI review: no blocking issues
| }, | ||
| abort(reason) { | ||
| // Clean up timer to prevent leaks | ||
| async abort(reason) { |
There was a problem hiding this comment.
AI Review: Note
abort() changes from "discard the buffer" to await drain() — it now delivers the accepted prefix over the network and only skips the server-side close. This is the correct choice given the early-ack contract (those chunks' write() already resolved to the producer), but it's a networked cancel/abort path, and world-local's synchronous cancel() masks exactly this class of behavior — it only exercises on the world-vercel prod E2E tier, which the PR notes was not run. Worth validating both abort triggers there before merge: source-error-during-pipeTo (deliver prefix) and explicit cancellation/teardown (where drain() may writeMulti to an already-terminal run and swallow the resulting conflict).
There was a problem hiding this comment.
On validation: all 13 E2E Vercel Prod lanes pass on this PR, so the networked write/close/cancel paths are exercised against the real world-vercel transport. On the two specific triggers: (1) source-error-during-pipeTo prefix delivery is pinned by a unit regression at the sink layer (mock world, asserts delivery + no close); (2) teardown-drain into a terminal run rejects into abort's existing catch — swallowed by design (best-effort delivery on a stream being torn down), now noted in the abort comment (e4c8a82). Happy to add a dedicated e2e for the source-error trigger as a follow-up if you want it pinned at that tier too.
| // skipped. A dispatch failure during this drain is already sticky | ||
| // and surfaces through the sink's error paths. | ||
| try { | ||
| await drain(); |
There was a problem hiding this comment.
AI Review: Note
Unlike the step-executor's inline flush (bounded at 500ms with a waitUntil fallback), this drain() inside abort() has no timeout. When abort is driven by pipeTo on a source error, a stalled world.streams.write/writeMulti would stall the abort with no upper bound. Low likelihood, but a bound (or a comment noting the reliance on the transport's own timeout) would make the teardown path's worst case explicit.
There was a problem hiding this comment.
Documented in e4c8a82: the drain is deliberately un-timeboxed — a bound that gives up would drop acked chunks, the exact loss this path exists to prevent — and the comment now states the worst case explicitly: the World transport's finite timeout/retry budget terminates a stalled write with a rejection, which lands in the catch and completes the abort.
…t handling Review note: the abort-path drain is deliberately un-timeboxed (a bound would drop acked chunks); its worst case is owned by the World transport's finite timeout/retry budget, and a teardown-driven drain into an already-terminal run rejects into the existing catch. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
The native-pipeTo batching test flaked on a slow CI runner: a fixed 25ms wait raced the 10ms commit window plus scheduler jitter. All 'dispatch has happened' assertions now poll the expectation (bounded); intentional negatives keep their fixed windows. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
|
No backport to This rework builds directly on To override, re-run the Backport to stable workflow manually via |
2 participants
Problem
Batching was implemented in
flushablePipe's coalescing loop, so it depended on which piping path ran.getWritable()batched; a rawReadableStreamcrossing a workflow/step boundary uses nativepipeTo(), which doesn't pull chunk N+1 untilwrite(chunk N)resolves — andwrite()resolved only after the flush timer and the server round trip. The buffer never held more than one chunk: one request per token.Fix
Batching moves into
WorkflowServerWritableStream, identical on every path:write()resolves when the chunk enters a bounded client buffer (WORKFLOW_STREAM_MAX_INFLIGHT_CHUNKScounts buffered and in-request chunks; new byte boundWORKFLOW_STREAM_MAX_BUFFERED_BYTES, default 8 MiB, documented).writeMultigroup. One request in flight preserves order; 1,000-chunk / 1 MiB per-request limits preserved.close()drains first;flushablePipe(now a plain pump: lock-release + durability tracking only) adopts the barrier so step completion still implies durability;abort()drains the accepted prefix without closing (a producer error after acked writes can't lose data); a failed pipe drains before settling. Dispatch failures are sticky and surface at the next write/close/drain.Trade-off
An open
getWritable()stream's first chunk now waits up to the group-commit interval (10ms default) instead of dispatching eagerly — watch TTFC /buffer_dwell_msafter release.Validation
1,572 core unit tests pass, including new regressions for: native-
pipeTobatching, awaited per-chunk loops coalescing into onewriteMulti, wire-cap splits, in-flight-inclusive backpressure on both bounds, source-error prefix delivery through abort, failed-pipe drain-before-reject, and turbo run-ready barrier gating. E2E tier requires a deployment and was not run here.Addressed some of the streaming perf concerns for use cases mentioned in #1930