fix(node/web): stream bridged responses, drop wire framing and hop-by-hop headers#253
fix(node/web): stream bridged responses, drop wire framing and hop-by-hop headers#253pi0x wants to merge 4 commits into
Conversation
`_request` serves a null-body native Request once srvx has consumed the body, so undici's own "Body is unusable" guard saw a pristine body and `arrayBuffer()` / `bytes()` / `blob()` / `formData()` resolved empty instead of rejecting, silently masking double-read bugs that throw on every other runtime. Only `text()` / `json()` guarded against this. Guard the remaining body methods, and reject rather than throw when materializing `_request` over a locked stream fails synchronously. Consuming `request.body` directly also never flipped `bodyUsed`, so a later read looked like a first read and threw "... disturbed or locked" synchronously out of the handler. Track the fetch spec's "disturbed" bit by wrapping the body stream. The wrapper uses `highWaterMark: 0` so it does not pull a chunk on construction and mark a body used merely because a handler touched `request.body`; `onDisturb` is inert once `_request` holds the body, since undici tees on `clone()` and a pull may then be the clone being read rather than this request's body. Co-Authored-By: Claude Opus 4.8 <[email protected]>
… headers (#248) `toWebResponse()` treated the captured wire bytes and wire headers of the synthetic `ServerResponse` as if they were the application's body and headers, and waited for the response to finish before handing one back. `useChunkedEncodingByDefault = false` only covers the default: a handler setting `transfer-encoding: chunked` explicitly re-enabled chunk framing, and the framing bytes landed *inside* the captured body ("5\r\nhello\r\n5\r\nworld\r\n0\r\n\r\n"). Force `chunkedEncoding = false` once the head is stored, so nothing is ever framed for this synthetic wire. The head parse copied out everything Node generated for that wire, including hop-by-hop headers. Every bridged response without a content-length carried `connection: close`, which `serve()` writes straight through — so a bridged response hung up the connection it was served on (and warns per-request over HTTP/2). Filter hop-by-hop headers, and stop synthesizing a `Date` for a hop that isn't real; one the handler sets explicitly still passes through. Awaiting `waitToFinish()` meant the `Response` only existed after `res.end()`: an endless SSE handler never resolved at all, and every other body was buffered whole in the stream's queue first. Resolve as soon as the head is stored and let the body stream through, with the writes parking on the consumer's backpressure instead of queueing unbounded. A handler failing after the head can no longer become a 500, so it tears the socket down and errors the body stream instead. The head is settled from an own-property patch of `_storeHeader` (the single funnel for explicit and implicit heads) rather than a prototype override, because Express re-parents the response object. Fixes #248 Co-Authored-By: Claude Opus 4.8 <[email protected]>
|
Caution Review failedAn error occurred during the review process. Please try again later. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
commit: |
pi0
changed the title
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (1)
src/adapters/_node/request.ts (1)
363-395: 🩺 Stability & Availability | 🔵 Trivial | 💤 Low valuePropagate out-of-band errors to the wrapper stream.
Currently, if the underlying
streamerrors before the wrapper is pulled, the error won't surface on the wrapper until a consumer explicitly callspull(). Adding astarthook to catch and propagatereader.closedrejections ensures that out-of-band errors (like a premature socket disconnection) are immediately reflected on the wrapper stream, even if it hasn't been read yet.💡 Proposed optional refactor
function trackDisturbed(stream: ReadableStream, onDisturb: () => void): ReadableStream { const reader = stream.getReader(); return new ReadableStream( { + start(controller) { + reader.closed.catch((error) => { + try { + controller.error(error); + } catch { + // Ignore if the controller is already closed/errored + } + }); + }, async pull(controller) { onDisturb(); const { done, value } = await reader.read(); if (done) { controller.close(); } else { controller.enqueue(value); } }, cancel(reason) { onDisturb(); return reader.cancel(reason); }, }, { highWaterMark: 0 }, ); }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/adapters/_node/request.ts` around lines 363 - 395, Update trackDisturbed to observe reader.closed in a start hook and propagate any rejection by calling controller.error with the underlying error. Preserve the existing pull and cancel disturbance tracking while ensuring underlying stream errors surface on the wrapper even before it is read.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/adapters/_node/web/fetch.ts`:
- Around line 59-64: In the handler rejection catch block, tear down the
associated WebRequestSocket and WebServerResponse before returning the 500
Response. Update the error path around logError(error, req, handler) to close or
destroy both bridge objects using their existing cleanup APIs, while preserving
the current logging and 500 response behavior.
In `@src/adapters/_node/web/response.ts`:
- Around line 22-28: Update the header filtering logic in response handling to
parse all tokens named by each Connection header and add them to the exclusion
set alongside HOP_BY_HOP_HEADERS before forwarding headers. Ensure nominated
headers such as X-Hop are removed with Connection, and add a regression test
covering this pattern.
In `@src/adapters/_node/web/socket.ts`:
- Line 46: Update the _destroy() teardown flow to retain the parked `#resumeWrite`
callback and invoke it with err or prematureCloseError() before clearing it, so
a backpressured res.write() is resolved when the body is cancelled.
---
Nitpick comments:
In `@src/adapters/_node/request.ts`:
- Around line 363-395: Update trackDisturbed to observe reader.closed in a start
hook and propagate any rejection by calling controller.error with the underlying
error. Preserve the existing pull and cancel disturbance tracking while ensuring
underlying stream errors surface on the wrapper even before it is read.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 4127f531-44b5-40c5-945e-76f6ac0c170a
📒 Files selected for processing (5)
src/adapters/_node/request.tssrc/adapters/_node/web/fetch.tssrc/adapters/_node/web/response.tssrc/adapters/_node/web/socket.tstest/node-adapters.test.ts
| } catch (error: any) { | ||
| // Client aborts / premature socket closes are routine (the client is already | ||
| // gone), so don't log them as errors. See https://github.com/h3js/srvx/issues/208 | ||
| const aborted = | ||
| req.signal?.aborted || | ||
| error?.name === "AbortError" || | ||
| error?.code === "ERR_STREAM_PREMATURE_CLOSE"; | ||
| if (!aborted) { | ||
| console.error(error, { cause: { req, handler } }); | ||
| } | ||
| logError(error, req, handler); | ||
| return new Response(null, { | ||
| status: 500, | ||
| statusText: "Internal Server Error", | ||
| }); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Tear down the bridge before returning the pre-head 500.
The rejected handler leaves WebRequestSocket and WebServerResponse open, allowing pending request reads or detached late writes to continue against an orphaned body.
Proposed fix
} catch (error: any) {
logError(error, req, handler);
+ socket.destroy(error instanceof Error ? error : undefined);
return new Response(null, {📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| } catch (error: any) { | |
| // Client aborts / premature socket closes are routine (the client is already | |
| // gone), so don't log them as errors. See https://github.com/h3js/srvx/issues/208 | |
| const aborted = | |
| req.signal?.aborted || | |
| error?.name === "AbortError" || | |
| error?.code === "ERR_STREAM_PREMATURE_CLOSE"; | |
| if (!aborted) { | |
| console.error(error, { cause: { req, handler } }); | |
| } | |
| logError(error, req, handler); | |
| return new Response(null, { | |
| status: 500, | |
| statusText: "Internal Server Error", | |
| }); | |
| } catch (error: any) { | |
| logError(error, req, handler); | |
| socket.destroy(error instanceof Error ? error : undefined); | |
| return new Response(null, { | |
| status: 500, | |
| statusText: "Internal Server Error", | |
| }); |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/adapters/_node/web/fetch.ts` around lines 59 - 64, In the handler
rejection catch block, tear down the associated WebRequestSocket and
WebServerResponse before returning the 500 Response. Update the error path
around logError(error, req, handler) to close or destroy both bridge objects
using their existing cleanup APIs, while preserving the current logging and 500
response behavior.
| // Connection-level headers describing the synthetic wire this response is | ||
| // written to, not the application response itself. Leaking them into the web | ||
| // `Response` breaks the next hop: `connection: close` disables keep-alive for | ||
| // every bridged response re-served over HTTP/1 and prints an UnsupportedWarning | ||
| // per request over HTTP/2, and `transfer-encoding` mislabels a body that is | ||
| // never chunk-framed here. https://datatracker.ietf.org/doc/html/rfc9110#section-7.6.1 | ||
| const HOP_BY_HOP_HEADERS = new Set(["connection", "keep-alive", "transfer-encoding", "upgrade"]); |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Also remove headers nominated by Connection.
A handler can emit Connection: x-hop with X-Hop: ...; this code removes Connection but forwards X-Hop, losing the metadata that marks it hop-specific. Parse all Connection tokens before filtering and add them to the exclusion set. Add a regression case for this pattern.
Also applies to: 154-154
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/adapters/_node/web/response.ts` around lines 22 - 28, Update the header
filtering logic in response handling to parse all tokens named by each
Connection header and add them to the exclusion set alongside HOP_BY_HOP_HEADERS
before forwarding headers. Ensure nominated headers such as X-Hop are removed
with Connection, and add a regression test covering this pattern.
| #_writeBody!: (chunk: Uint8Array) => void; | ||
| #resBodyController?: ReadableStreamDefaultController<Uint8Array>; | ||
| #resBodyClosed?: boolean; | ||
| #resumeWrite?: () => void; |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -eu
# Inspect the declared Node toolchain and existing cancellation coverage.
fd -a -t f '^(package\.json|\.nvmrc|\.node-version|\.tool-versions)$' . \
--exec sh -c 'echo "### $1"; sed -n "1,180p" "$1"' sh {}
rg -n -C5 '`#resumeWrite`|cancelling the response body|res\.write\(' \
src/adapters/_node/web/socket.ts test/node-adapters.test.tsRepository: h3js/srvx
Length of output: 17781
🏁 Script executed:
#!/bin/bash
set -eu
sed -n '1,260p' src/adapters/_node/web/socket.ts | nl -ba | sed -n '1,260p'
printf '\n### resumeWrite references\n'
rg -n '`#resumeWrite`|resumeWrite\?' src/adapters/_node/web/socket.ts
printf '\n### teardown and close-related tests\n'
sed -n '1280,1310p' test/node-adapters.test.ts | nl -ba | sed -n '1,80p'Repository: h3js/srvx
Length of output: 502
🏁 Script executed:
#!/bin/bash
set -eu
sed -n '1,280p' src/adapters/_node/web/socket.ts
printf '\n### cancellation test block\n'
sed -n '1280,1310p' test/node-adapters.test.tsRepository: h3js/srvx
Length of output: 9133
🌐 Web query:
Node.js Writable _write callback destroy pending write callback backpressure stream destroy semantics
💡 Result:
In Node.js Writable streams, the _write callback is a critical part of the stream's backpressure and state management. When you implement a custom Writable, your _write(chunk, encoding, callback) method must call the provided callback to signal that the chunk has been successfully handled or that an error occurred [1][2]. Backpressure Mechanism The stream manages backpressure based on the internal buffer size compared to the highWaterMark [3][4]. When you call writable.write(chunk), the stream returns false if the buffer size meets or exceeds the highWaterMark, indicating that the producer should stop sending data until the 'drain' event is emitted [3][5]. The stream manages this process by not invoking the next _write call until the current _write's callback has been triggered [1][4]. Destroy and Pending Writes The interaction between destroy and pending _write callbacks has been refined to ensure safe and predictable behavior. Specifically: 1. Interaction with Destroy: When stream.destroy is called, the stream ceases to process further data [6][7]. However, it is designed to interact safely with pending write operations [7]. 2. Callback Handling: Recent improvements in Node.js have ensured that even if a stream is destroyed during an active _write operation, the callback for that operation is still invoked [7]. This prevents issues where an implementation might be left waiting indefinitely for a callback that would never fire, and avoids exceptions that could occur if the stream attempted to process more writes after being destroyed [7]. 3. Semantics: destroy is considered a disruptive action intended to tear down state and release resources, rather than a graceful shutdown (which is the role of end) [6]. Implementations should not assume that _destroy will wait for all pending _write operations to complete naturally; instead, the stream infrastructure attempts to clean up while ensuring that pending write callbacks are honored to prevent inconsistent state [6][7]. To ensure robust implementation, always call the _write callback with an error if the operation fails, or with null if it succeeds, even in cases involving potential stream destruction [1][2]. Adding an 'error' event listener is also recommended for reliably detecting errors, as the _write callback may not always receive them depending on the stream's configuration [2].
Citations:
- 1: https://www.thenodebook.com/streams/writable-streams
- 2: https://github.com/nodejs/readable-stream/blob/d22eb88b8fb85a2701a007e00081f50fa49eb415/doc/stream.md
- 3: https://nodejs.org/api/stream.html
- 4: https://iximiuz.com/en/posts/nodejs-writable-streams-distilled/
- 5: https://nodejs.org/learn/modules/backpressuring-in-streams
- 6: stream: Writable should not invoke _destroy while pending ops nodejs/node#29044
- 7: nodejs/node@6ce4ef3
Fail the parked write callback during teardown. Clearing #resumeWrite in _destroy() leaves a backpressured res.write() unresolved after body cancellation; store the callback and invoke it with err ?? prematureCloseError() instead.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/adapters/_node/web/socket.ts` at line 46, Update the _destroy() teardown
flow to retain the parked `#resumeWrite` callback and invoke it with err or
prematureCloseError() before clearing it, so a backpressured res.write() is
resolved when the body is cancelled.
# Conflicts: # src/adapters/_node/request.ts # src/adapters/_node/web/fetch.ts # test/node-adapters.test.ts
2 participants
Fixes #248.
All three issues share the root cause from the issue:
toWebResponse()treated the captured wire bytes and wire headers of the syntheticServerResponseas if they were the application's body and headers, and waited for the response to finish before handing one back.1. Explicit
transfer-encoding: chunkedcorrupts the bodyuseChunkedEncodingByDefault = falseonly covers the default — setting the header explicitly re-enabled framing, and the framing bytes landed inside the body. NowchunkedEncodingis forced back tofalseonce the head is stored, so nothing is ever framed for a wire that isn't real. Thetransfer-encodingheader is dropped as hop-by-hop, so the body and its framing header agree.2. Hop-by-hop headers leak
connection,keep-alive,transfer-encodingandupgradeare now filtered out of the head parse. Node's syntheticDateis no longer generated either (sendDate = false) — stamping it is the serving hop's job, and it re-stamps one. ADatethe handler sets explicitly still passes through.I confirmed the consequence you predicted, with one wrinkle worth recording: it only reproduces through the bridge path.
serve({ fetch: (req) => fetchNodeHandler(handler, req) })passes srvx's own request straight through tocallNodeHandler(req.runtime.nodeis set), so it never builds a synthetic response. Forcing the bridge with a plainRequest, two pipelined requests on one socket:Also worth noting: with a
content-lengththe leak isconnection: keep-aliveinstead, which overrode a client'sConnection: closerequest. Same fix covers it.3. Full buffering breaks streaming/SSE
The
Responsenow resolves as soon as the head is stored — onwriteHead()or the firstwrite()— and the body streams as the handler writes it.fetchNodeHandlerno longer awaits the handler before building the response; it races the handler against the head, so a handler that fails before the head is still a 500. Once the head is out a 500 is impossible, so a late failure tears the socket down and errors the body stream rather than hanging.Writes now park on the consumer's backpressure (
pull) instead of queueing unbounded, so a handler piping a large file can't outrun its reader — this is what makes "large payloads buffer fully in memory" actually go away rather than just move. Cancelling the body tears the response down instead of throwing on a closed controller (newly reachable now that consumers get the body early).Express SSE through
toFetchHandler, served over a real socket:Notes for review
_storeHeaderrather than a prototype override: it's the single funnel for both explicitwriteHead()and implicit heads, and an own property survives Express'sObject.setPrototypeOfre-parenting (the same reason the existing methods are bound in the constructor). The existing express fixture proves this — its head can only settle via that patch.waitToFinish()is removed.toWebResponse()was its only caller and the class isn't exported; its socket-error/premature-close semantics moved into the head promise. Happy to keep it if you'd rather.connectionin the headers, SSE never resolving, 64/64 chunks buffered, cancel, late failure), 2 are guards for behavior that must not break (explicitDatepasses through, pre-head failure is still a 500).arrayBuffer()/bytes()/blob()/formData()resolve empty after body is consumed instead of rejecting #247 (1031 passed), lint/format/typecheck clean.🤖 Generated with Claude Code
Summary by CodeRabbit
Bug Fixes
bodyUsedtracking and standard errors for repeated reads.Tests