[core] Ack'd write-channel stream writes with framed-v2 writer markers#2731
[core] Ack'd write-channel stream writes with framed-v2 writer markers#2731VaguelySerious wants to merge 31 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: 2dfcf0d 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 |
📊 Workflow Benchmarkscommit Backend:
📜 Previous results (2)30cb495Fri, 10 Jul 2026 22:21:25 GMT · run logs
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. |
🧪 E2E Test Results❌ Some tests failed Summary
❌ Failed Tests▲ Vercel Production (1 failed)express (1 failed):
Details by Category❌ ▲ Vercel Production
✅ 💻 Local Development
✅ 📦 Local Production
✅ 🐘 Local Postgres
✅ 🪟 Windows
✅ 📋 Other
✅ vercel-multi-region
❌ 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>
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}
…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
left a comment
karthikscale3
left a comment
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( |
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.
There was a problem hiding this comment.
Addressed in 7b7fa0d:
- The handshake span now carries the same
workflow.run.id/workflow.stream.name/workflow.stream.operationattributes as the PUT write spans (plusurl.full), andtrace()records an error status when the upgrade fails. - A per-writer summary span (
workflow.stream.ws_writer, bounded — one per writer lifetime) records outcome (closed / aborted / ws_fallback), reconnect count, frames sent/acked, bytes acked, peak queue depth/bytes, and chunk-to-ack latency (avg + max). - Unclean reconnects log with reason/code/attempt; fallbacks log with the triggering error and emit the summary span.
Close outcome and ack latency deliberately live on the writer-summary span rather than the handshake span — a handshake span ends at the 101 and can't carry what hasn't happened yet. On correlation: the upgrade span is where traceparent is injected, and trace-propagation.test.ts asserts the injection happens inside the client span; the server parents its per-connection spans to that context (its message-processing spans are children of the upgrade request context), which the existing server-side stream span tests cover. A deployed end-to-end trace-joining assertion needs the e2e trace pipeline — happy to add one there as a follow-up.
| emitClose({ reason: 'connection error' }); | ||
| }); | ||
|
|
||
| await new Promise<void>((resolve, reject) => { |
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.
There was a problem hiding this comment.
Addressed in 7b7fa0d: the handshake has an explicit deadline (WORKFLOW_STREAM_WRITE_CONNECT_TIMEOUT_MS, default 10s) that closes the socket and rejects; a socket error now rejects the connect promise directly (first-settle-wins) instead of relying on a close event that a black-holed handshake never delivers. All three connect listeners are { once: true } and the timer is cleared on the first settle, so every terminal path cleans up.
| private startRecycleTimer(): void { | ||
| const setTimer = this.deps.setTimer ?? setTimeout; | ||
| this.recycleTimer = setTimer(() => { | ||
| this.rotateRequested = true; |
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.
There was a problem hiding this comment.
Addressed in 7b7fa0d:
- Ack liveness deadline: a rolling deadline on the oldest sent-unacked frame (
WORKFLOW_STREAM_WRITE_ACK_DEADLINE_MS, default 15s; every ack rearms it) presumes a silent connection half-open and tears it down for a resend — so neither producers (blocked on the window) norclose()(blocked on the drain) can hang on a dead socket. - Timer hygiene: the reconnect-backoff, recycle, ack-deadline, idle, and FIN timers are all tracked and cleared on abort, failure, rotation, and teardown (and unref'd so they never hold a process open by themselves).
- Drained-idle close: a fully-drained channel is released after
WORKFLOW_STREAM_WRITE_IDLE_CLOSE_MS(default 10s) instead of being pinned until the recycle interval; the next write reconnects.
Unit tests cover the half-open resend, the idle release + reconnect, and timer cleanup on the failure paths.
| // 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) { |
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.
There was a problem hiding this comment.
Addressed in 7b7fa0d: WORKFLOW_STREAM_WRITE_TRANSPORT = auto (default) | websocket | put.
putis the operational kill switch — retransmit-safe writes go straight to batched v2 PUT.- In
auto, a fatal channel failure (upgrade repeatedly refused, reconnect budget exhausted, server-version mismatch) circuit-breaks to batched PUT forWORKFLOW_STREAM_WRITE_FALLBACK_COOLDOWN_MS(default 60s). The failed writer's admitted-but-unconfirmed frames are redelivered over PUT ahead of the new batch (their framed-v2 markers dedupe any overlap with copies the dead channel did persist), so the fallback is lossless.close()has the same fallback before sending the done marker. websocketforces the channel with no fallback, for tests that must fail loudly.- Fallbacks log the triggering reason and emit the writer-summary span with outcome
ws_fallback.
Covered by unit tests: fallback with abandoned-frame redelivery, breaker suppression, forced-websocket rejection, and the PUT kill switch.
| const writer = socketWriters.get(socketWriterKey(resolvedRunId, name)); | ||
| if (writer) { | ||
| socketWriters.delete(socketWriterKey(resolvedRunId, name)); | ||
| await writer.close(); |
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.
There was a problem hiding this comment.
Addressed in 7b7fa0d (SDK) + vercel/workflow-server@5ed2625 (server): StreamSocketWriter.close() now completes an explicit FIN/FIN_ACK exchange after the drain — the server drains its chain, flushes final storage-bytes accounting, replies FIN_ACK, and closes 1000; only then does the streamer issue the X-Stream-Done PUT. The wait is bounded (WORKFLOW_STREAM_WRITE_FIN_ACK_TIMEOUT_MS, 10s): on a lost FIN_ACK every frame is already acked (durable), only accounting order is at stake, and the server's close handler still flushes as a fallback — so close() degrades to today's behavior instead of hanging.
| } | ||
| buffer.push(chunk); | ||
| scheduleFlush(); | ||
| // Resolve only when the scheduled flush completes so callers (flushablePipe) |
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.
There was a problem hiding this comment.
Addressed in 7b7fa0d — we took the second option (explicit contract) and closed the gap that made it unsound:
- The
writeMulti/closedocs in@workflow/worldnow state the deferred-durability model explicitly: resolving means accepted for durable in-order delivery; withretransmitSafedurability may be confirmed asynchronously, a world must keep redelivering accepted chunks until confirmed, andclose()resolving is the caller's durability barrier — unrecoverable failures MUST surface on a laterwriteMulti/close. - world-vercel now actually guarantees that: on a fatal channel failure, admitted-but-unconfirmed frames are held and redelivered over batched PUT by the next
writeMultior byclose()(markers dedupe overlap). No admitted frame can be silently dropped while the invocation completes normally — eitherclose()resolves with everything durable, or it rejects.ackBarrier()stays non-rejecting by design (it only extends the invocation lifetime via waitUntil so a frozen process can't strand resends); error propagation is the write/close chain's job, as now documented at the call site.
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.
… FIN/FIN_ACK, telemetry Review follow-ups on the WS stream-write transport (wf#2731): - Connect deadline: the upgrade handshake times out (10s default, WORKFLOW_STREAM_WRITE_CONNECT_TIMEOUT_MS), rejects directly on the first socket error, and cleans up its listeners/timer on every terminal path — a black-holed handshake can no longer strand the writer in 'connecting'. - Ack liveness: a rolling deadline on the oldest sent-unacked frame (WORKFLOW_STREAM_WRITE_ACK_DEADLINE_MS) presumes a silent connection half-open and replaces it; reconnect/recycle/idle timers are tracked and cleared on abort, failure, and teardown. A fully-drained channel is released after WORKFLOW_STREAM_WRITE_IDLE_CLOSE_MS instead of being pinned until the recycle. - Transport selection + circuit breaker: WORKFLOW_STREAM_WRITE_TRANSPORT (auto|websocket|put) adds an operational kill switch; in auto, a fatal channel failure falls back to batched v2 PUT for a cooldown, redelivering the writer's admitted-but-unconfirmed frames first (framed-v2 markers dedupe any overlap), so no admitted frame is ever silently dropped. close() gets the same fallback before sending the done marker. - FIN/FIN_ACK: close() completes a finalization exchange — the server drains, flushes final storage accounting, and confirms — before the done PUT can go out; bounded by WORKFLOW_STREAM_WRITE_FIN_ACK_TIMEOUT_MS. - Telemetry: the upgrade span carries run/stream/operation attributes; a per-writer summary span (workflow.stream.ws_writer) records outcome, reconnects, frame/byte counts, peak queue depth/bytes, and chunk-to-ack latency; unclean reconnects and fallbacks log with reasons. - Durability semantics: the writeMulti/close contract now states the deferred-durability model explicitly (resolve-on-admission, close() as the durability barrier, failures surface on a later call). - Tests: FIN exchange (confirm + timeout paths), ack-deadline resend, idle release, abandoned-frame recovery, PUT fallback with redelivery, forced-transport modes, and a two-writer interleave dedupe test. - Docs: all WORKFLOW_STREAM_WRITE_* env vars documented in runtime tuning. - events-v4 getEventV4 test uses the override-aware ORIGIN like the rest of the file, so unit tests pass with the branch URL override active. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
|
Review feedback addressed in 7b7fa0d (inline threads have individual replies): @karthikscale3's items — all addressed:
@TooTallNate's design feedback:
Acknowledged pre-merge steps (unchanged): bump the All |
| const useChannel = | ||
| options?.retransmitSafe && | ||
| transport !== 'put' && | ||
| (transport === 'websocket' || Date.now() >= wsSuppressedUntil); |
There was a problem hiding this comment.
The streamer-wide WS→PUT circuit breaker (wsSuppressedUntil) diverts a different concurrently-active stream's writes to PUT mid-stream while that stream's live WS writer still holds earlier unacked (lower-seq) frames, causing silent chunk loss/reordering via the reader's per-writerId dedup.
|
Triage: Server request logs for both failed runs ( Key points:
Possible follow-up (out of scope here): a short server-side grace window admitting the first write of a new stream just after terminal state (TTL'd like late-arriving chunks to existing streams), which would eliminate this flake class entirely. |
3 participants
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, completes a FIN/FIN_ACK finalization exchange (the server flushes its final storage accounting and confirms) and only then sends the done marker. Everything else uses the batched v2 PUT — which is unchanged.WORKFLOW_STREAM_WRITE_TRANSPORTselectsauto(default) |websocket|put(operational kill switch). Inauto, a fatal channel failure circuit-breaks to batched PUT for a cooldown, redelivering the writer's admitted-but-unconfirmed frames first (markers dedupe any overlap), so a WS regression degrades to today's PUT path losslessly instead of failing streams.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. Liveness is bounded end to end: the upgrade handshake has a connect deadline, a rolling ack deadline replaces half-open connections that go silent, and a fully-drained channel is released after a short idle instead of pinning its socket until the recycle. All tuning knobs are env-overridable (WORKFLOW_STREAM_WRITE_*) and documented in the v5 runtime-tuning page.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.writeMulti/close):writeMultiresolves on window admission (flushes pipeline instead of stop-and-wait); durability is confirmed by acks,close()resolving is the durability barrier for everything previously accepted, and an unrecoverable delivery failure surfaces on a laterwriteMulti/close(in practice it triggers the lossless PUT fallback instead). Each admitted batch registers an ack barrier with the platform'swaitUntilso an invocation is never suspended while frames await confirmation.trace-propagation.test.ts). Each writer additionally emits one bounded summary span (workflow.stream.ws_writer) with outcome, reconnects, frame/byte counts, peak queue depth, and chunk-to-ack latency; reconnects and fallbacks log with reasons..../wsupgrade route (plus the existing v3 live-read GET): batched PUT writes, completion, and snapshot reads all stay on v2 exactly as today.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 (and FIN/FIN_ACK protocol) added in vercel/workflow-server#584; merging this PR before that deploy would make every granted write fall back to batched PUT (functional, but the feature would ship dark).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.