fix(core): batch stream writes via writeMulti#2995
Conversation
…chunk flushablePipe awaited each writer.write() and the WritableStream sink serializes chunks one at a time, so WorkflowServerWritableStream's buffer never held more than one chunk and its writeMulti batching path (gated on chunksToFlush.length > 1) never engaged — every chunk became its own server round trip. WorkflowServerWritableStream now exposes a durable batch write under STREAM_WRITE_BATCH_SYMBOL. flushablePipe feature-detects it and runs a producer/consumer that coalesces chunks arriving while a previous batch is still in flight into a single writeMulti, preserving the existing durability guarantees (pendingOps counts read-but-not-durable chunks; source-done awaits every batch before closing). Sinks without the batch entry point keep the unchanged per-chunk path. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
🦋 Changeset detectedLatest commit: fa004f3 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✅ All tests passed Summary
Details by Category✅ ▲ Vercel Production
✅ 💻 Local Development
✅ 📦 Local Production
✅ 🐘 Local Postgres
✅ 🪟 Windows
✅ 📋 Other
✅ vercel-multi-region
|
📊 Workflow Benchmarkscommit Backend:
Avg 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) 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) 🟢/🔴 mark percentiles within/above target. Targets (p75/p90/p99, ms) — TTFS 200/300/600 · SL 50/60/125 · 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 |
VaguelySerious
changed the title
1 participant
Problem
Stream writes never batch. Every chunk becomes its own server round trip, even under a fast burst —
world.streams.writeMultiis effectively dead code.Two facts combine to defeat the buffer/
writeMultimachinery inWorkflowServerWritableStream:WritableStreamcontract serializes the sink.write(chunk)for the next chunk is not invoked until the previouswrite()'s promise settles.flushablePipeawaits eachwriter.write()before reading the next chunk, and the sink'swrite()resolves only after its flush completes (needed sopendingOps/durability tracking is accurate).Together these mean the sink's buffer never holds more than one chunk when the flush timer fires, so the batching path (
chunksToFlush.length > 1) never runs.Reproduction
Piping 20 chunks that are all available instantly through the current coordination:
Fix
WorkflowServerWritableStreamnow exposes a durable batch write under a newSTREAM_WRITE_BATCH_SYMBOL. It reuses the existingflush()path, so a batch goes out as onewriteMulti(or sequentialwrites when the world lackswriteMulti) and resolves only once the whole batch has reached the server.flushablePipefeature-detects the symbol and runs a producer/consumer: the producer reads into a queue, the consumer drains the whole queue into one batch write at a time. Chunks that arrive while a previous batch is in flight coalesce into the next server write — turning a burst into a handful of round trips instead of one per chunk. Bounded read-ahead (WORKFLOW_STREAM_MAX_INFLIGHT_CHUNKS, default 1000) provides backpressure.pendingOpscounts chunks read-but-not-yet-durably-written (incremented at read time, decremented only after the batch write resolves), so the lock-release completion (pollWritableLock/pollReadableLock) never fires while data is still queued or in flight, and the source-done path awaits every batch before closing.WritableStreams,TransformStreamwritables on the read path) keep the unchanged per-chunk path — so all existing behavior and tests are untouched.After the fix, the same 20-chunk burst emits
1initial single write + onewriteMultiof the remaining 19.Tests
flushable-stream.test.ts: coalescing during an in-flight batch, in-order delivery + close,pendingOpsdurability across lock-release, batch-write failure propagation, and backpressure.writable-stream.test.ts: end-to-end on the realWorkflowServerWritableStream+ mock world viaflushablePipe, asserting onewriteMultiwith the coalesced chunks, plus the sequential-writefallback when the world has nowriteMulti.pnpm testscope (vitest run src, world=local): 69 files, 1489 passed, 0 failures. Typecheck clean; no new lint warnings.Notes for reviewers
Promise.raceon read / sink-closed / consumer-failure, and the backpressure wait) is the part most worth a close read.🤖 Generated with Claude Code