[core] Ack'd write-channel stream writes with framed-v2 writer markers#2731
Open
VaguelySerious wants to merge 28 commits into
Open
[core] Ack'd write-channel stream writes with framed-v2 writer markers#2731VaguelySerious wants to merge 28 commits into
VaguelySerious wants to merge 28 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: 30cb495 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
📊 Workflow Benchmarkscommit Backend:
📜 Previous results (1)db399bfThu, 09 Jul 2026 19:16:20 GMT · run logs
Avg deltas compare against the most recent benchmark run on Metrics — TTFS: time to first step body execution · STSO: step-to-step overhead (gap between consecutive step bodies) · WO: workflow overhead (time outside step bodies, client start → last step body exit) · SL: stream latency (first chunk write → visible to the reader) Scenarios — stream: one step that streams chunks back to the client; no hooks, so the run stays in turbo mode · hook + stream: registers a hook before the same streaming step, which exits turbo mode · 1020 steps: 1020 trivial sequential steps; STSO is measured between consecutive steps in the given step ranges 🟢/🔴 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 TTFS/WO compare client vs deployment clocks and SL compares the step runner’s clock vs the client’s (NTP-synced in CI). WO ends at the last step body exit, the closest observable proxy for the final step-completion request. |
Contributor
🧪 E2E Test Results❌ Some tests failed Summary
❌ Failed Tests▲ Vercel Production (3 failed)express (2 failed):
nitro (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. |
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>
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>
![vercel[bot]](https://avatars.githubusercontent.com/in/8329?s=60&v=4){kind=small}
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>
…ice for one failed attempt, double-counting `consecutiveReconnects`/`totalReconnects` and scheduling two reconnect timers.
This commit fixes the issue reported at packages/core/src/serialization/stream-socket-writer.ts:248
## Bug
`StreamSocketWriter.ensureChannel()` opens a channel via `deps.connect(...)` and, on failure, reports the failure through `onChannelClose`. The epoch guard in `onChannelClose` is meant to make it idempotent: it bumps `this.epoch` and ignores any subsequent call whose captured epoch no longer matches. But the `catch` block passed the *live* `this.epoch` instead of the epoch captured for this attempt, which defeats the guard on the connect-failure path.
### Why it double-fires for a failed WS upgrade
The transport is `createStreamer().connectWrite` in `packages/world-vercel/src/streamer.ts`. It registers listeners in this order:
1. A **persistent** `close` listener → `emitClose(...)` → `handlers.onClose(event)` (registered before awaiting the connect promise).
2. Inside `await new Promise(...)`, a `{ once: true }` `close` listener that **rejects** the connect promise.
When the upgrade fails, a `close` event fires before `open`. Listeners run in registration order:
1. The persistent listener fires **synchronously** → `handlers.onClose` → `this.onChannelClose(epoch, event)`. Captured `epoch === this.epoch`, so it proceeds: sets `channel = null`, increments `this.epoch`, increments `consecutiveReconnects`/`totalReconnects`, schedules a reconnect timer.
2. The `once` listener then rejects the promise (settled as a microtask), so `connectWrite` throws and `ensureChannel`'s `catch` runs → `this.onChannelClose(this.epoch, ...)`. Because `this.epoch` was just bumped in step 1, the passed value **again equals the current** `this.epoch`, so it proceeds a **second** time — double-incrementing the counters and scheduling a second timer.
### Impact
A single failed connect counts as two reconnects. The writer gives up after ~3 real consecutive failures instead of `maxConsecutiveReconnects` (5), and the lifetime `maxTotalReconnects` budget is effectively halved. Two overlapping reconnect timers are also scheduled.
The existing unit tests don't catch this because the test harness's `connect` rejects **without** invoking `handlers.onClose`, so it only exercises the single (catch-only) path. The real transport fires both.
## Fix
In `ensureChannel()`, hoist `const epoch = ++this.epoch;` (and the `this.sentInEpoch = 0` reset) above the `try` so the attempt's epoch is captured once, and pass that **captured** `epoch` into the `catch`'s `onChannelClose` call instead of `this.epoch`.
Now for a failed WS upgrade:
* Step 1 (`handlers.onClose`) runs with the captured epoch, matches, and bumps `this.epoch`.
* Step 2 (`catch`) runs with the same captured epoch, which no longer matches the bumped `this.epoch`, so the guard makes it a stale no-op.
Other paths are unchanged: the connect-reject-only path (no `onClose`) still fires once via the catch with a matching epoch; the `ensureReady`-throws path fires once (epoch bumped, nothing else touched `this.epoch`); the successful-then-closed path fires once via `onClose`. Moving the epoch bump before the `ensureReady` await is behavior-preserving — `abort()`/`teardownChannel()` during `ensureReady` sets `fatalError` (and doesn't bump epoch when there's no channel), and the post-connect `if (epoch !== this.epoch || this.fatalError)` check still short-circuits.
Co-authored-by: Vercel <vercel[bot]@users.noreply.github.com>
Co-authored-by: VaguelySerious <mittgfu@gmail.com>
…pty stream, so `X-Stream-Done` can race run creation in turbo optimistic start (run-not-found / lost close marker).
This commit fixes the issue reported at packages/core/src/serialization.ts:1094
## Bug
Commit `3b08d48` replaced `createSegmentSink` with `createSocketSink` (`packages/core/src/serialization.ts`), but the empty-stream ordering gap remains. `close()` was:
```js
close: async () => {
await socketWriter.close();
await world.streams.close(runId, name);
},
```
The run-ready barrier (`ensureRunReady`) is passed to `StreamSocketWriter` as `ensureReady`, and it is only awaited inside `ensureChannel()`:
```js
private async ensureChannel(): Promise<void> {
if (this.channel || this.connecting || this.fatalError) return;
this.connecting = true;
try {
if (this.deps.ensureReady) {
this.ensureReadyPromise ??= this.deps.ensureReady();
await this.ensureReadyPromise;
}
...
```
`ensureChannel()` is only reached via `pump()`, and `pump()` opens a channel only when `this.buffer.length > 0`. For an **empty stream** (writable opened and closed with no writes), `StreamSocketWriter.close()` sees an empty buffer and returns without ever opening a channel:
```js
async close(): Promise<void> {
this.throwIfUnusable();
if (this.buffer.length > 0) {
await new Promise<void>((resolve) => {
this.drainWaiter = resolve;
this.pump();
});
this.throwIfUnusable();
}
this.teardownChannel();
this.closedDone = true;
}
```
So `ensureReady`/`ensureRunReady` is never awaited, and the sink immediately fires `world.streams.close(runId, name)` (the `X-Stream-Done` PUT).
### Trigger / failure mode
Turbo optimistic start (`runReadyBarrier` provided) + an empty stream that is closed with no writes. The `X-Stream-Done` PUT reaches the World before `run_started` is durable → run-not-found error / lost close marker. The `WorkflowServerWritableStream` doc even asserts the empty-stream close "orders after the run-ready barrier" — which the socket sink did not satisfy.
The sibling `createBatchSink.close()` already handles exactly this case:
```js
await flush();
// A close with an empty buffer skips flush()'s write (and its barrier),
// but can itself be the first write to a brand-new stream — gate it too.
await ensureRunReady();
await world.streams.close(runId, name);
```
## Fix
Await `ensureRunReady()` before `world.streams.close(runId, name)` in `createSocketSink.close()`, mirroring the batch sink. Because `ensureRunReady`/`ensureReadyPromise` is memoized after the first await, this is a no-op when a channel already opened, and correctly orders the empty-stream close after run creation.
## Reachability note
The socket sink is selected only when the world supports `connectWrite` and a `writerId` is present. This is code being wired up, so the fix prevents a latent race once the socket sink is enabled in production.
Co-authored-by: Vercel <vercel[bot]@users.noreply.github.com>
Co-authored-by: VaguelySerious <mittgfu@gmail.com>
…esulting on-wire frame length (chunk.length + FRAME_MARKER_SIZE) exceeds MAX_FRAME_SIZE, so the unframer rejects them — breaking the documented "any chunk the framer accepts can always be decoded by the unframer" invariant.
This commit fixes the issue reported at packages/core/src/serialization.ts:591
## Bug
In `packages/core/src/serialization.ts`, `getByteFramingStream(writerId?)` validates user chunks at write time (now around line 591) with a single `chunk.length > MAX_FRAME_SIZE` check that ignores the framed-v2 marker:
```ts
if (chunk.length > MAX_FRAME_SIZE) { controller.error(...); return; }
if (writerId) {
controller.enqueue(buildFramedV2Frame(chunk, { writerId, seq: seq++ }));
return;
}
```
For **framed-v2** (`writerId` present), `buildFramedV2Frame` (in `serialization/frame-marker.ts`) emits a frame whose 4-byte length prefix declares `bodyLength = FRAME_MARKER_SIZE + inner.length`, where `FRAME_MARKER_SIZE = WRITER_ID_SIZE (8) + 8 = 16` bytes:
```ts
const bodyLength = FRAME_MARKER_SIZE + inner.length;
view.setUint32(0, bodyLength, false);
```
The reader `getByteUnframingStream` rejects any frame whose declared length exceeds the cap:
```ts
if (frameLength > MAX_FRAME_SIZE) { controller.error("...exceeds maximum..."); return; }
```
### Concrete trigger
A framed-v2 user chunk with `chunk.length` in `(MAX_FRAME_SIZE - FRAME_MARKER_SIZE, MAX_FRAME_SIZE]` — e.g. exactly `100_000_000` bytes — passes the writer's `chunk.length > MAX_FRAME_SIZE` check (100,000,000 is not `> 100,000,000`), so the framer enqueues a frame with declared length `100_000_016`. The reader then evaluates `frameLength (100_000_016) > MAX_FRAME_SIZE (100_000_000)` as true and **errors the stream** with "exceeds maximum", failing an otherwise valid framed-v2 stream. This violates the `MAX_FRAME_SIZE` docstring guarantee that "any chunk the framer accepts can always be decoded by the unframer."
## Fix
Tighten the write-time bound to mirror the reader's effective limit, subtracting the marker only when a `writerId` is present:
```ts
const maxChunkSize = writerId
? MAX_FRAME_SIZE - FRAME_MARKER_SIZE
: MAX_FRAME_SIZE;
if (chunk.length > maxChunkSize) { controller.error(... maxChunkSize ...); return; }
```
`FRAME_MARKER_SIZE` is already imported into `serialization.ts` from `./serialization/frame-marker.js` (line 55), so no new import is needed. For framed-v1 (no `writerId`) `maxChunkSize` stays `MAX_FRAME_SIZE`, preserving existing behavior. Now any framed-v2 chunk the framer accepts produces a frame length ≤ `MAX_FRAME_SIZE`, and oversized chunks fail loudly at the producer where the error is actionable.
## Verification note
The code was refactored (commit 3b08d48) but the bug is unchanged; only the line moved (~534 → ~582/591). `buildFramedV2Frame` still sets the length prefix to `FRAME_MARKER_SIZE + inner.length`, and the unframer still enforces `frameLength > MAX_FRAME_SIZE`. Patch re-applied at the new location.
Co-authored-by: Vercel <vercel[bot]@users.noreply.github.com>
Co-authored-by: VaguelySerious <mittgfu@gmail.com>
All stream PUT operations (write, writeMulti, close) now target the v3 stream endpoint, which publishes each chunk's availability as its write lands instead of deferring publishes to the end of the request body. The v2 write path is removed. Sets WORKFLOW_SERVER_URL_OVERRIDE to the backend branch preview so e2e exercises the v3 endpoints — temporary, reverted before merge. Tests that mock the server derive their origin from getHttpUrl() so they hold under any override; tests asserting unset-override defaults skip while an inline override is active. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
# Conflicts: # packages/core/src/serialization.ts # packages/world-vercel/src/streamer.ts
Extends the WORKFLOW_* env-override pattern to the socket writer's config: recycle interval, in-flight window (frames/bytes), and reconnect budgets resolve through envNumber so a dedicated e2e deployment can dial them down and exercise rotation, backpressure, and reconnect paths quickly. Explicit constructor config still wins; the recycle floor guards against connection-churn misconfiguration. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…rumentedFetch Leftover from the abandoned single-request streaming write design (the platform buffers streaming request bodies whole, so long-lived writes go over the WS write channel instead). No caller passes a stream body; writes send fully-buffered bodies. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
… writeMulti
Core now has a single world-agnostic write path: the buffered sink
flushes through writeMulti, passing { retransmitSafe: true } when frames
carry framed-v2 per-writer markers (readers can deduplicate resends).
How a batch is delivered becomes the world's concern — world-vercel
routes retransmit-safe batches over the long-lived acknowledged
WebSocket channel (one writer per stream, drained by close() before the
done marker) and everything else over the batched PUT. The
Streamer.connectWrite channel API is removed from @workflow/world;
StreamSocketWriter and its channel types move into world-vercel. Other
worlds are unaffected: the new writeMulti option is optional and
ignored where unsupported.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…eam no longer tears down world-vercel's per-stream StreamSocketWriter, leaking the writer and leaving its WebSocket reconnecting/resending abandoned data.
This commit fixes the issue reported at packages/world-vercel/src/streamer.ts:369
## What's wrong
Commit `2cd6fcd` moved the WebSocket write transport out of core and behind `writeMulti`. Before the refactor, core's `createSocketSink.abort(reason)` called `socketWriter.abort(reason)`, which tore down the `StreamSocketWriter`: it clears the recycle timer, closes the WebSocket, abandons unacked frames, sets `fatalError`, and rejects in-flight waiters.
After the refactor, the `StreamSocketWriter` lives in world-vercel's `createStreamer` in a per-stream `socketWriters` map, created by `writeMulti(..., { retransmitSafe: true })` and removed only in two places:
- `close()` — drains the writer and deletes the entry, then sends `X-Stream-Done`.
- `writeMulti`'s `catch` — deletes the entry when `writer.write()` throws a *fatal* error.
Core's replacement `createBatchSink.abort` (packages/core/src/serialization.ts) became **purely local** — it clears the buffer and rejects flush waiters but calls nothing on `world.streams`. The `Streamer` interface (packages/world/src/interfaces.ts) had **no stream-abort method**, so there was no way for an aborted `WritableStream` to signal world-vercel to tear the writer down.
## Failure mode / trigger
`WorkflowServerWritableStream` is an exported class whose underlying sink implements `abort(reason)` (which delegates to `createBatchSink.abort`). Aborting a **framed-v2** instance — e.g. `writer.abort(reason)` (a standard, exercised `WritableStream` API; see `writable-stream.test.ts` "abort behavior") — takes the abort path. With the retransmit-safe socket writer in the map, that path never reaches the writer, so:
1. The `StreamSocketWriter` entry stays in `socketWriters` forever (unbounded growth in a long-lived process).
2. If it still holds unacked frames, it keeps the WebSocket open and reconnects/resends them — delivering data for a stream that was aborted, and burning reconnect budget.
3. It is never drained or torn down.
This is a genuine regression of the deliberate teardown that existed pre-refactor, on a public/exported contract.
### Reachability caveat (recorded honestly)
In the *current* production wiring, framed-v2 writables are driven by `flushablePipe`, whose error path releases the writer lock without calling the sink's `abort()`. So the abort *contract* is what is broken today; the primary in-repo trigger is a direct `.abort()` on the writable. (I also filed a separate note: `flushablePipe`'s error path itself never tears the socket writer down — a related, pre-existing leak on *source* errors that this change does not address.)
## The fix
- **packages/world/src/interfaces.ts**: add an optional `abort?(runId, name, reason?): Promise<void>` to `Streamer.streams`, documenting that retransmit-safe worlds must stop reconnecting/resending here. Optional so worlds with no per-stream write state are unaffected (feature-detected).
- **packages/world-vercel/src/streamer.ts**: implement `abort` to look up the stream's `StreamSocketWriter`, delete it from the map, and call `writer.abort(reason)` (teardown + abandon unacked frames).
- **packages/core/src/serialization.ts**: `createBatchSink.abort` now returns `world.streams.abort?.(runId, name, reason)` (optional-chained, so it no-ops for worlds without the method), and `StreamSink.abort`'s return type widens to `void | Promise<void>`. The `WorkflowServerWritableStream` sink's `abort` handler now awaits the sink abort so world-level teardown completes before the writable settles.
This restores the pre-refactor guarantee: aborting a stream tears down the world's transport-level write state instead of leaking it.
Co-authored-by: Vercel <vercel[bot]@users.noreply.github.com>
Co-authored-by: VaguelySerious <mittgfu@gmail.com>
- Fix a drain deadlock: close() hung forever when the recycle rotation completed on the same ack that emptied the buffer (pump() tore the channel down and returned before resolving the drain waiter). - Treat a 1009 close (message too big) as fatal with a chunk-size error instead of resending the oversized frame until the reconnect budget dies. - Lower the default in-flight window to 64 frames so it stays under the server's per-connection backlog cap (100). - Inject W3C trace context on the WS upgrade inside a client span, like every other request to the backend (covered in trace-propagation.test.ts). - Register an ack barrier with the platform's waitUntil so an invocation is not suspended while admitted frames still await durability confirmation. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
- Byte streams (step-returned and workflow-argument ReadableStreams) now emit framed-v2 with a per-writer marker when the target run's capability allows, giving them the same read-side dedupe and retransmit-safe write path as getWritable() streams. Threaded as framedStreamMarkers through the reducers and dehydrate entry points, driven by getRunCapabilities. - Remove the WORKFLOW_EXPERIMENTAL_STREAM_MARKERS escape hatch; the capability cutoff (5.0.0-beta.26, at/below the tree version) turns the framed-v2 + retransmit-safe path on across unit and e2e lanes in CI. TODO(release): bump the cutoff to the first beta that ships framed-v2. - Rewrite comments that still described the abandoned single-request streaming design (tail-scan recovery) — markers exist so readers can deduplicate frames a retransmitting write transport resent. - Split the changeset into core+world and world-vercel entries. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
A workflow-VM getWritable() handle was a bare {name}: a step (or external
client) reviving it saw no framing field and wrote framed-v1, while
Run.getReadable derives framed-v2 from the run's SDK version and strips a
16-byte marker that was never written — corrupting every frame (caught by
world-testing's hooks test hanging on an unparseable event stream).
The run's stream framing is now computed host-side at VM setup
(framedStreamMarkersEnabled over this SDK's version — the VM cannot import
the capability table without pulling the serialization module into the
workflow bundle) and exposed as the WORKFLOW_DEFAULT_STREAM_FRAMING global;
getWritable tags its handles with it, so the existing reducer/reviver
framing round-trip applies to VM-minted handles like every other writable.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
# Conflicts: # packages/core/src/workflow/create-hook.ts
The v3 PUT route (write/writeMulti/close) was functionally identical to v2 PUT server-side (the incremental-publish behavior that motivated a separate version was already reverted), leaving it an untested duplicate. This PR's only new route is the WS write channel; batched PUT writes go back to v2, matching main. The live-read GET and the WS upgrade stay on v3. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Conflict in packages/world-vercel/src/streamer.ts: main added client- observed stream span attributes (#2857) on the same write/read call sites this branch's v2/v3 URL restoration touches. Resolved by merging both telemetry.js import lists; the streamSpanAttributes/spanName additions and the getStreamUrl/getStreamReadUrl split are independent and both apply cleanly to the same call sites. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
karthikscale3
requested changes
Jul 10, 2026
karthikscale3
left a comment
karthikscale3
left a comment
Contributor
There was a problem hiding this comment.
Requesting changes before enabling this transport by version capability. Pipelined acknowledged writes should improve sustained write-heavy throughput, but the client currently has unbounded half-open/drain waits, no operational PUT fallback, a finalization race with the server, and incomplete telemetry/error parity. The changed write-promise durability semantics also need to be made explicit or restored. Details are inline.
| // connection's spans to the caller. (Chunks sent later on the socket | ||
| // carry no per-message headers — the upgrade is the one place the trace | ||
| // link can be established.) | ||
| return trace( |
Contributor
There was a problem hiding this comment.
This is not yet telemetry-equivalent to the existing instrumentedFetch write path. The span currently has only the URL and ends after the handshake; it lacks workflow run/stream/operation attributes, standard HTTP/peer/status attributes, close outcome, reconnect reason, and ACK latency. Please retain a dedicated handshake span with those attributes and add bounded metrics/spans for connection outcome, reconnects, queue depth/bytes, and chunk-to-ACK latency. Also add an end-to-end test proving server message spans remain correlated after the upgrade span has ended, not just that traceparent was placed on the request.
| emitClose({ reason: 'connection error' }); | ||
| }); | ||
|
|
||
| await new Promise<void>((resolve, reject) => { |
Contributor
There was a problem hiding this comment.
This connect promise has no deadline and rejects only on close. The error listener calls emitClose, but if a failed/black-holed handshake never completes the close handshake this can remain pending forever with the writer stuck in connecting. Please add an explicit upgrade timeout, reject directly on the first error, and ensure all listeners/timers are cleaned up on every terminal path.
| private startRecycleTimer(): void { | ||
| const setTimer = this.deps.setTimer ?? setTimeout; | ||
| this.recycleTimer = setTimer(() => { | ||
| this.rotateRequested = true; |
Contributor
There was a problem hiding this comment.
The recycle timer is not a liveness timeout: it sets rotateRequested, stops feeding the socket, and then waits indefinitely for already-sent frames to ACK. A half-open connection can therefore strand producer writes and close() forever. Please add an oldest-unacked ACK deadline plus ping/pong or equivalent liveness detection, and store/clear reconnect/ACK timers during abort, failure, and teardown. A drained-idle timeout would also avoid retaining sockets for bursty streams until the full recycle interval.
| // Resolving here means the chunks are accepted into the writer's | ||
| // bounded in-flight window (backpressure); durability of everything | ||
| // written is confirmed when `close` drains the writer. | ||
| if (options?.retransmitSafe) { |
Contributor
There was a problem hiding this comment.
There is no safe rollout fallback here. A server-version mismatch, unsupported api-workflow proxy upgrade, regional deployment skew, or WebSocket beta regression causes repeated reconnects and then fails the stream even though v2 PUT remains available. Because these frames are retransmit-safe, please add auto | websocket | put selection and a circuit-breaker fallback to batched PUT after upgrade/reconnect failure, with a runtime kill switch and fallback-reason telemetry.
| const writer = socketWriters.get(socketWriterKey(resolvedRunId, name)); | ||
| if (writer) { | ||
| socketWriters.delete(socketWriterKey(resolvedRunId, name)); | ||
| await writer.close(); |
Contributor
There was a problem hiding this comment.
StreamSocketWriter.close() only initiates the WebSocket close; it does not wait for the close handshake or for the server's close-triggered final accounting flush. The done PUT can therefore race server finalization even after all chunk ACKs arrived. Please introduce a FIN/FIN_ACK control exchange and wait for the server to drain writes and flush accounting before issuing X-Stream-Done.
| } | ||
| buffer.push(chunk); | ||
| scheduleFlush(); | ||
| // Resolve only when the scheduled flush completes so callers (flushablePipe) |
Contributor
There was a problem hiding this comment.
This comment and the previous contract say the data reached the server, but retransmit-safe writeMulti now resolves on local window admission, before any durability ACK. That changes observable write/error semantics: a write can resolve and later fail asynchronously, while ackBarrier() deliberately never rejects. Either keep the write promise tied to an ACK barrier, or explicitly change/document the interface and guarantee a later awaited operation observes the failure before the invocation can complete or suspend.
TooTallNate
requested changes
Jul 11, 2026
Member
There was a problem hiding this comment.
The architecture is the strongest part of this PR: core stays world-agnostic with a single write path (buffered sink → writeMulti), framed-v2 markers (writerId + seq) grant retransmitSafe: true permission, and world-vercel alone owns the transport choice. The 4x write-throughput claim and 2-minute e2e speedup are compelling. But this is explicitly and implicitly not mergeable yet:
Blocking:
-
The PR body itself says "NEED TO UPDATE THE VERSION CUTOFF BEFORE MERGING" — an acknowledged unresolved item.
-
WORKFLOW_SERVER_URL_OVERRIDEbaked to a branch deployment URL (marked TEMPORARY in the code, with the lint guard correctly flagging it — theNo Test Overridescheck is red). Must be reverted before merge. -
Unit Tests failing (ubuntu) plus E2E Required Check failing (express + nitro prod). Needs a green run to evaluate the actual behavior.
-
Existing CHANGES_REQUESTED from @karthikscale3 remains open.
-
Backend dependency ordering: the WS write endpoint this rides on must be deployed and stable before any SDK version that prefers the WS transport ships — and the version cutoff (item 1) is exactly the mechanism for that, which is why it must be resolved, not just noted.
Design feedback (for when this is re-rolled):
- The
retransmitSafeopt-in onStreamWriteOptionsis the right abstraction — worlds that ignore it keep current behavior, and the permission is derived from the actual framing capability (framed-v2 markers) rather than a version sniff. Good. - Reader-side dedup of the persisted-but-unacked overlap is the linchpin of the whole reconnect model. When CI is green I'd want to see explicit test coverage of: (a) a reconnect resending an already-persisted chunk and the reader deduplicating it via the writer marker, and (b) two writers (old + new invocation after a crash) interleaving on the same stream — the
writerIdin the marker suggests this is designed for, but I couldn't confirm test coverage from the current red state. streams.abortas an optional world interface addition is clean.
Happy to re-review once the cutoff is set, the override reverted, and CI is green — after the backend is live.
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
3 participants
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.
TLDR;
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).writeMulti. When frames carry framed-v2 markers, the flush passes{ retransmitSafe: true }(new optionalStreamWriteOptionsonwriteMulti), granting the world permission to use a transport that resends unconfirmed chunks across reconnects, since readers can deduplicate the overlap. Worlds that ignore the option keep exactly their current behavior. Aborting a writable tears down the world's transport state via the new optionalstreams.abort.{index, chunkIndex}in order;close()drains all pending acks before sending the done marker. Everything else uses the batched PUT.StreamSocketWriter(in world-vercel): evicts its replay buffer per ack (memory bounded by the in-flight window — 64 frames / 4 MB, deliberately under the server's 100-chunk per-connection backlog cap so a burst can never enter a shed→resend→shed loop), recycles connections proactively (~110 s), and survives unclean closes by resending unacked frames on a fresh channel, under consecutive + lifetime reconnect budgets. A 1009 close (chunk larger than the server's message cap) fails fast with a chunk-size error instead of burning the reconnect budget on a deterministic rejection. All tuning knobs are env-overridable (WORKFLOW_STREAM_WRITE_*).getWritable()object streams and byte streams (step-returned / workflow-argumentReadableStreams) emit framed-v2 when the target run's capability allows; read-side dedupe by(writerId, seq)makes resend overlap exactly-once, including with concurrent writers to one stream (forwarded writables mint their own writerId; framing is per-stream, carried in descriptors and through the workflow VM round-trip). Below the capability cutoff everything stays framed-v1/raw + batch writes with no retransmit grant — byte-identical legacy behavior.writeMultiresolves on window admission (flushes pipeline instead of stop-and-wait); durability is confirmed by acks,close()resolves only after a full drain, and each admitted batch registers an ack barrier with the platform'swaitUntilso an invocation is never suspended while frames await confirmation.trace-propagation.test.ts.write/writeMultifallback/close) move from the v2 to the v3 endpoint (identical handler semantics; makes rollout and request-log analysis unambiguous). The v2 write path is removed.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.
Throughput note: within one connection the backend persists chunks serially (reserve → write → publish per chunk), where the PUT batch path parallelized storage writes within a batch. In exchange the writer pipelines continuously instead of stop-and-waiting one round trip per 10 ms batch, so sustained throughput improves for realistic producers; only extreme small-chunk producers trade peak burst rate for the lower request overhead.
Merge checklist (ordered)
/wsupgrade route added there; merging this PR before that deploy breaks all stream writes.WORKFLOW_SERVER_URL_OVERRIDEto''inpackages/world-vercel/src/utils.ts(the "No Test Overrides" check enforces this; it is expected-red until then).framedStreamMarkerscutoff to the first beta that actually ships this (currently the next one,beta.28—beta.26/beta.27are published without marker support; the too-low cutoff is only safe on this branch, where writer and reader are always built together). ATODO(release)marks the line.