[core] Ack'd write-channel stream writes with framed-v2 writer markers#2731
Open
VaguelySerious wants to merge 10 commits into
Open
[core] Ack'd write-channel stream writes with framed-v2 writer markers#2731VaguelySerious wants to merge 10 commits into
VaguelySerious wants to merge 10 commits into
Conversation
Groundwork for single-request streaming stream writes. Adds an opt-in framed-v2 frame header carrying a per-writer marker ([writerId][seq]) to both the byte framer/unframer and the object serialize/deserialize streams, plus a shared marker codec. The reader strips the marker and dedupes replays by max-seq-per-writerId, so a frame recovery re-sends after it was already persisted is delivered exactly once. Not yet wired into any writer (no writerId is passed today), so there is no user-facing behavior change. Capability gating + the streaming segment writer + tail-match recovery follow. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
🦋 Changeset detectedLatest commit: 6cdfe67 The changes in this PR will be included in the next version bump. This PR includes changesets to release 20 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 |
Contributor
Contributor
📊 Benchmark Results
workflow with no steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Express | Next.js (Turbopack) | Nitro workflow with 1 step💻 Local Development
▲ Production (Vercel)
🔍 Observability: Nitro | Express | Next.js (Turbopack) workflow with 10 sequential steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Nitro | Next.js (Turbopack) | Express workflow with 25 sequential steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Nitro | Express | Next.js (Turbopack) workflow with 50 sequential steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Nitro | Express | Next.js (Turbopack) Promise.all with 10 concurrent steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Express | Nitro | Next.js (Turbopack) Promise.all with 25 concurrent steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Express | Nitro | Next.js (Turbopack) Promise.all with 50 concurrent steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Express | Nitro | Next.js (Turbopack) Promise.race with 10 concurrent steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Express | Nitro | Next.js (Turbopack) Promise.race with 25 concurrent steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Nitro | Express | Next.js (Turbopack) Promise.race with 50 concurrent steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Nitro | Express | Next.js (Turbopack) workflow with 10 sequential data payload steps (10KB)💻 Local Development
▲ Production (Vercel)
🔍 Observability: Nitro | Express | Next.js (Turbopack) workflow with 25 sequential data payload steps (10KB)💻 Local Development
▲ Production (Vercel)
🔍 Observability: Nitro | Express | Next.js (Turbopack) workflow with 50 sequential data payload steps (10KB)💻 Local Development
▲ Production (Vercel)
🔍 Observability: Express | Nitro | Next.js (Turbopack) workflow with 10 concurrent data payload steps (10KB)💻 Local Development
▲ Production (Vercel)
🔍 Observability: Express | Nitro | Next.js (Turbopack) workflow with 25 concurrent data payload steps (10KB)💻 Local Development
▲ Production (Vercel)
🔍 Observability: Express | Nitro | Next.js (Turbopack) workflow with 50 concurrent data payload steps (10KB)💻 Local Development
▲ Production (Vercel)
🔍 Observability: Nitro | Express | Next.js (Turbopack) Stream Benchmarks (includes TTFB metrics)workflow with stream💻 Local Development
▲ Production (Vercel)
🔍 Observability: Express | Nitro | Next.js (Turbopack) stream pipeline with 5 transform steps (1MB)💻 Local Development
▲ Production (Vercel)
🔍 Observability: Express | Nitro | Next.js (Turbopack) 10 parallel streams (1MB each)💻 Local Development
▲ Production (Vercel)
🔍 Observability: Nitro | Express | Next.js (Turbopack) fan-out fan-in 10 streams (1MB each)💻 Local Development
▲ Production (Vercel)
🔍 Observability: Nitro | Express | Next.js (Turbopack) SummaryFastest Framework by WorldWinner determined by most benchmark wins
Fastest World by FrameworkWinner determined by most benchmark wins
Column Definitions
Worlds:
|
Contributor
🧪 E2E Test Results✅ All tests passed Summary
Details by Category✅ ▲ Vercel Production
✅ 💻 Local Development
✅ 📦 Local Production
✅ 🐘 Local Postgres
✅ 🪟 Windows
✅ 📋 Other
|
FNV-1a (64-bit) over a seed string → 8-byte writerId. Callers pass a value stable across deterministic replays (a seeded STABLE_ULID), so the writerId is replay-stable without consuming the VM's seeded RNG (which would shift the sequence observed by user code). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
![vercel[bot]](https://avatars.githubusercontent.com/in/8329?s=60&v=4){kind=small}
recoverStreamTail reconstructs, after an unclean segment failure, which of a writer's in-flight frames the backend persisted, and replays only the rest: - Scans the persisted tail window [max(priorIndices)+1, tailIndex], matching the writer's OWN frames by the framed-v2 marker (concurrent writers share a stream, so a server index isn't attributable to one writer). - Handles the reserve-ahead race (tailIndex can point at a reserved-but- unpersisted chunk) with read-with-backoff 10/100/1000ms; a tail chunk still missing after the window is a real write failure, surfaced not skipped. - Replays frames with seq > max-persisted-seq on a fresh writeStream, bounded by maxAttempts (re-scans between attempts since a replay may partially persist). StreamSegmentWriter.recover now receives the prior clean segment's indices as the scan anchor. Both fully unit-tested; not yet wired into the writer. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Rewrite WorkflowServerWritableStream from unconditional 10ms timer batching to lazy sink selection: when the world exposes streams.writeStream and the writer has a writerId (framed-v2), frames flow through StreamSegmentWriter (one long-lived streaming request per ~10s segment, clean 200 drops the buffer, recoverStreamTail replays unconfirmed frames on unclean failure). Otherwise the existing per-batch writeMulti path is used, extracted verbatim into createBatchSink. No behavior change yet: no caller passes a writerId, so all writers stay on the batch path until getWritable is wired for framed-v2. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
| }, | ||
| DEFAULT_SEGMENT_CONFIG | ||
| ); | ||
|
|
Contributor
There was a problem hiding this comment.
The socket sink's close() skips the run-ready barrier on an empty stream, so X-Stream-Done can race run creation in turbo optimistic start (run-not-found / lost close marker).
Replace the streaming-PUT write path with an ack'd WebSocket write
channel, and wire framed-v2 emission end to end.
Transport: `Streamer.connectWrite` opens a channel on the world-vercel
v3 `/ws` stream route (undici WebSocket so the upgrade carries the same
auth headers as every HTTP call). Each chunk is one binary message,
persisted + published on arrival and acked back `{index, chunkIndex}`
in order. The streaming-PUT `writeStream` is removed: the platform
buffers streaming request bodies whole, so it could never deliver
incremental visibility (probed empirically; the WS path delivers
~75ms p50 chunk-to-ack on deployed infra).
Writer: `StreamSocketWriter` replaces the segment writer — the local
buffer is evicted per ack instead of per request, bounded by an
in-flight window; connections recycle proactively under the server's
bound; an unclean close resends everything unacked on a fresh channel,
with consecutive + lifetime reconnect budgets. The tail-scan recovery
module is removed: read-side framed-v2 dedupe (writerId+seq) absorbs
the persisted-but-unacked resend overlap, which acks make tiny.
Flip: `getWritable` derives a replay-stable writerId and emits
framed-v2 when `framedStreamMarkersEnabled(own version)` — a per-run
decision every reader can reproduce. `getReadable` derives the same
answer from the run's `executionContext.workflowCoreVersion` via a
lazy thenable (fetched only when the first chunk arrives). Framing is
per-stream: forwarded writables carry `framing` in their descriptor
(and through the workflow VM round-trip), and a forwarded writer mints
its own writerId. `WORKFLOW_EXPERIMENTAL_STREAM_MARKERS=1`
force-enables for tests/e2e ahead of the version cutoff, so e2e can
assert engagement instead of silently exercising the legacy path.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
| this.ensureReadyPromise ??= this.deps.ensureReady(); | ||
| await this.ensureReadyPromise; | ||
| } | ||
| const epoch = ++this.epoch; |
Contributor
There was a problem hiding this comment.
On a WebSocket connect failure, onChannelClose is invoked twice for one failed attempt, double-counting consecutiveReconnects/totalReconnects and scheduling two reconnect timers.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What
Stream writes move from one short PUT per 10 ms flush batch to a long-lived, acknowledged write channel (WebSocket), with a new framed-v2 wire format that stamps every frame with a per-writer marker (
writerId+seq).Streamer.connectWrite(optional world capability): opens a write channel; each chunk is one binary message, persisted and published by the backend on arrival and acked back{index, chunkIndex}in order.StreamSocketWriter: evicts its replay buffer per ack (memory bounded by a small in-flight window, not connection lifetime), recycles connections proactively (~110 s), and survives unclean closes by resending unacked frames on a fresh channel, under consecutive + lifetime reconnect budgets.(writerId, seq)makes the resend overlap exactly-once, including with concurrent writers to one stream (parent → child forwarded writables mint their own writerId; framing is per-stream and carried in descriptors and through the workflow VM round-trip).getWritableemits framed-v2 based onframedStreamMarkersEnabled(<run's SDK version>);getReadablereproduces the same decision from the run'sexecutionContext.workflowCoreVersion(lazily, on first chunk). Everything below the version cutoff stays byte-identical framed-v1 + batch writes, including all non-Vercel worlds (noconnectWrite→ batch path, unchanged).Why
Today every active writer costs ~100 requests/second and a chunk is only durable-confirmed when its batch's PUT resolves. A streaming request body can't fix this on deployed infra (bodies are delivered to functions only at close — verified empirically), while WS messages arrive incrementally: the deployed probe measured p50 75 ms chunk-to-ack, with live readers seeing chunks immediately via per-chunk publish.
Validation
getWritableflip round-trip (markers on, markers off, batch fallback), and ack-protocol edge cases.WORKFLOW_EXPERIMENTAL_STREAM_MARKERS=1force-enables the path for e2e ahead of the version cutoff, so e2e can assert engagement rather than silently exercising the legacy path.TODO(release):framedStreamMarkersminVersion is5.0.0-beta.27— verify it matches the version this actually ships in.🤖 Generated with Claude Code