From 543eb4d2d312ef8e5680c9bc936f02a071e504a0 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Thu, 22 Oct 2026 03:32:02 +0000 Subject: [PATCH 1/2] fetch: reject a close-delimited body that a connection reset cut short A response with no Content-Length and no chunked framing ends when the server closes the connection. The HTTP/1 client took every close for that end, so a TCP reset completed a truncated body as a whole 200 response. On Linux and Android the client now records how the read side ended: ReadEnd::Fin or ReadEnd::Error. The close-delimited arm of is_body_complete_on_close completes the body only on Fin. A body that a reset cut short fails with ConnectionClosed (ECONNRESET). The kernel gives the errno of a reset to the first call that meets it. The client keeps it in these cases: - Handler::on_close classifies the uSockets close code. It ignored the code before. - maybe_pause_receive does not read SO_ERROR under the rule. That read took the pending errno, and the next recv() returned 0 like a FIN. - A plain-TCP request write uses write_check_error. A send that fails with ECONNRESET sets a latch that turns Fin into Error. On TLS a close_notify from the peer ends its stream, whatever the connection does after it. A TLS connection that had a send rejected (us_socket_ssl_send_rejected) keeps the old rule: the code of its close stands for the lost records, not for a failed read. The rule does not apply to a 101-upgraded connection, a request over a unix socket, or a request through a CONNECT tunnel. On other systems fetch does what it did before. --- src/http/HTTPContext.rs | 45 +++++++++++++++-- src/http/InternalState.rs | 22 ++++++-- src/http/ProxyTunnel.rs | 5 +- src/http/lib.rs | 104 ++++++++++++++++++++++++++++++++++---- src/uws_sys/socket.rs | 4 ++ 5 files changed, 161 insertions(+), 19 deletions(-) diff --git a/src/http/HTTPContext.rs b/src/http/HTTPContext.rs index 62cbb44b3e26..783e61508a6b 100644 --- a/src/http/HTTPContext.rs +++ b/src/http/HTTPContext.rs @@ -4,6 +4,7 @@ use core::ptr::NonNull; use crate::Error; use crate::http_thread::InitOpts as HTTPThreadInitOpts; +use crate::internal_state::ReadEnd; use crate::ssl_config::{self, SSLConfig}; use crate::{ self as http, AlpnOffer, HTTPCertError, HTTPClient, InitError, ProxyTunnel, @@ -1465,12 +1466,50 @@ impl Handler { HTTPContext::::terminate_socket(socket); } - pub fn on_close(ptr: *mut c_void, socket: HTTPSocket, _: c_int, _: Option<*mut c_void>) { + /// How the read side of `socket` ended, from the close that uSockets reports with + /// `code`: above `CloseCode::fast_shutdown` that is the error of the failed read (loop.c). + fn read_end(socket: HTTPSocket, code: c_int) -> ReadEnd { + if !http::STRICT_READ_END { + return ReadEnd::Fin; + } + if SSL { + // The peer's close_notify ended its stream, whatever the connection did after + // it. BoringSSL sets the flag for a fatal read error too; uSockets closes that + // session itself, with no errno. + let peer_ended_stream = socket.ssl().is_some_and(|ssl| { + // SAFETY: uSockets frees the `SSL` after the close dispatch that this runs in. + (unsafe { bun_boringssl_sys::SSL_get_shutdown(ssl) } + & bun_boringssl_sys::SSL_RECEIVED_SHUTDOWN) + != 0 + }); + if peer_ended_stream { + return ReadEnd::Fin; + } + // A send that the kernel rejected took the connection's error. The code of the + // close that follows stands for the lost records, not for a failed read, so it + // does not say how the read side ended: the body keeps the old rule. + if socket.ssl_send_rejected() { + return ReadEnd::Fin; + } + } + if http::is_cut(code) { + ReadEnd::Error + } else { + ReadEnd::Fin + } + } + + pub fn on_close( + ptr: *mut c_void, + socket: HTTPSocket, + code: c_int, + _: Option<*mut c_void>, + ) { let tagged = HTTPContext::::get_tagged(ptr); HTTPContext::::mark_socket_as_dead(socket); if let Some(client) = tagged.client_mut() { - return client.on_close::(socket); + return client.on_close::(socket, Self::read_end(socket, code)); } if let Some(session) = tagged.session() { return h2::ClientSession::on_close(session, crate::Error::ConnectionClosed); @@ -1629,7 +1668,7 @@ impl Handler { } else { socket.close(uws::CloseKind::Normal); } - client.on_close::(socket); + client.on_close::(socket, ReadEnd::Fin); return; } if let Some(session) = tagged.session() { diff --git a/src/http/InternalState.rs b/src/http/InternalState.rs index d2c6cb245acf..2fb760b73615 100644 --- a/src/http/InternalState.rs +++ b/src/http/InternalState.rs @@ -98,6 +98,18 @@ pub struct InternalStateFlags { pub(crate) body_compressed: bool, /// Held input or buffered decoder output remains for `HTTPClient::drain_response_body`. pub(crate) decompress_output_pending: bool, + /// A request write failed with ECONNRESET. The failed `send()` took that error from the + /// kernel, so the read side then ends with `recv() == 0` like a FIN. + pub(crate) write_took_error: bool, +} + +/// How the read side of a connection ended. +#[derive(Clone, Copy, PartialEq, Eq)] +pub(crate) enum ReadEnd { + /// At the end of the peer's stream: its FIN, or its close_notify on TLS. + Fin, + /// With an error and no end of stream, such as a reset. + Error, } impl InternalStateFlags { @@ -114,6 +126,7 @@ impl InternalStateFlags { receive_paused: false, body_compressed: false, decompress_output_pending: false, + write_took_error: false, } } } @@ -249,8 +262,9 @@ impl<'a> InternalState<'a> { /// True when a socket close during `in_progress` completes the body rather /// than failing it: chunked decoder already in the trailers state, or a - /// close-delimited response (no Content-Length, no Transfer-Encoding). - pub(crate) fn is_body_complete_on_close(&self) -> bool { + /// close-delimited response (no Content-Length, no Transfer-Encoding) whose + /// connection reached the end of the peer's stream. + pub(crate) fn is_body_complete_on_close(&self, read_end: ReadEnd) -> bool { // Every byte arrived; only the decode is outstanding. if self.flags.decompress_output_pending && self.is_done() { return true; @@ -258,7 +272,9 @@ impl<'a> InternalState<'a> { if self.is_chunked_encoding() { return bun_picohttp::phr_decode_chunked_is_in_trailers(&self.chunked_decoder) != 0; } - self.content_length.is_none() && self.response_stage == HTTPStage::Body + self.content_length.is_none() + && self.response_stage == HTTPStage::Body + && read_end == ReadEnd::Fin } /// Mark the body complete and drive `process_body_buffer` one last time diff --git a/src/http/ProxyTunnel.rs b/src/http/ProxyTunnel.rs index b6ea15bf7be0..ccbd14c59795 100644 --- a/src/http/ProxyTunnel.rs +++ b/src/http/ProxyTunnel.rs @@ -9,7 +9,7 @@ use bun_uws as uws; use crate::http_cert_error::HTTPCertError; use crate::http_context::{HTTPSocket, PeerVerification}; -use crate::internal_state::{HTTPStage, Stage}; +use crate::internal_state::{HTTPStage, ReadEnd, Stage}; use crate::ssl_config::SSLConfig; use crate::ssl_wrapper::{Handlers as SSLWrapperHandlers, InitError, SSLWrapper, WriteDataError}; use crate::{AlpnOffer, HTTPClient}; @@ -513,7 +513,8 @@ fn on_close(ctx: *mut HTTPClient) { && this.state.stage != Stage::Fail && !this.state.flags.is_redirect_pending; let mut fail_err: Option = None; - if in_progress && this.state.is_body_complete_on_close() { + // The inner TLS session ended. How the outer socket ends is for `HTTPClient::on_close`. + if in_progress && this.state.is_body_complete_on_close(ReadEnd::Fin) { match this.state.finalize_body_on_eof() { Ok(()) => { // `this` dead (NLL); reborrow via `client_from_ctx` inside. diff --git a/src/http/lib.rs b/src/http/lib.rs index c59a51b1f5dc..da1512bcee7e 100644 --- a/src/http/lib.rs +++ b/src/http/lib.rs @@ -973,7 +973,7 @@ use bun_uws as uws; use bun_wyhash::Wyhash; use crate::http_context::HTTPSocket as HttpSocket; -use crate::internal_state::{RequestStage, ResponseStage, Stage}; +use crate::internal_state::{ReadEnd, RequestStage, ResponseStage, Stage}; bun_core::declare_scope!(fetch, visible); @@ -1484,15 +1484,51 @@ fn print_response(response: &picohttp::Response<'_>) { Output::flush(); } +/// Whether a close-delimited body completes only at the peer's FIN. Limited to the kernel whose +/// errno order was measured: on Linux a reset that follows the peer's FIN leaves EPIPE, and only +/// a reset with no FIN leaves ECONNRESET. Elsewhere every end of the connection completes such a +/// body. +pub(crate) const STRICT_READ_END: bool = cfg!(any(target_os = "linux", target_os = "android")); + +/// Whether `errno`, a uSockets close code, says that the connection was cut before the peer's +/// stream ended. Up to `FastShutdown` a close code is this side's own close, and EPIPE is the +/// reset that followed the peer's FIN. +pub(crate) fn is_cut(errno: i32) -> bool { + errno > uws::CloseKind::FastShutdown as i32 && errno != bun_sys::E::EPIPE as i32 +} + +/// One write to the socket that carries a request. Returns what `socket.write` returns. +/// +/// A plain-TCP `send()` that fails with ECONNRESET takes the reset out of the kernel, so the +/// read side can no longer report it: `took_error` keeps that for `HTTPClient::on_close`. +/// ECONNRESET only: a send can also fail with a timeout or an abort after the peer's FIN +/// arrived, and the body is whole then. A TLS connection that had a send rejected keeps the +/// old rule (`Handler::read_end`). +fn write_once( + socket: HttpSocket, + data: &[u8], + took_error: &mut bool, +) -> i32 { + if IS_SSL || !STRICT_READ_END { + return socket.write(data); + } + let (written, fatal_errno) = socket.write_check_error(data); + if fatal_errno == bun_sys::E::ECONNRESET as i32 { + *took_error = true; + } + written +} + /// Write data to the socket (Just a error wrapper to easly handle amount written and error handling) fn write_to_socket( socket: HttpSocket, data: &[u8], + took_error: &mut bool, ) -> crate::Result { let mut remaining = data; let mut total_written: usize = 0; while !remaining.is_empty() { - let amount = socket.write(remaining); + let amount = write_once::(socket, remaining, took_error); if amount < 0 { return Err(crate::Error::WriteFailed); } @@ -1511,8 +1547,9 @@ fn write_to_socket_with_buffer_fallback( socket: HttpSocket, buffer: &mut bun_io::StreamBuffer, data: &[u8], + took_error: &mut bool, ) -> crate::Result { - let amount = write_to_socket::(socket, data)?; + let amount = write_to_socket::(socket, data, took_error)?; if amount < data.len() { let _ = buffer.write(&data[amount..]); } @@ -2246,7 +2283,28 @@ impl<'a> HTTPClient<'a> { && self.state.response_stage != ResponseStage::BodyChunk } - pub(crate) fn on_close(&mut self, socket: HttpSocket) { + /// `read_end` as it counts for the body of this request. + fn body_read_end(&self, read_end: ReadEnd) -> ReadEnd { + // An upgraded connection is a byte stream with no body to cut. On a unix socket + // ECONNRESET only says that the peer closed with input unread, whole answer or not. + // Behind a CONNECT tunnel the socket ends with the proxy's teardown, not the origin's. + if self.flags.upgrade_state == HTTPUpgradeState::Upgraded + || !self.unix_socket_path.is_empty() + || self.proxy_tunnel.is_some() + { + return ReadEnd::Fin; + } + if self.state.flags.write_took_error { + return ReadEnd::Error; + } + read_end + } + + pub(crate) fn on_close( + &mut self, + socket: HttpSocket, + read_end: ReadEnd, + ) { bun_core::scoped_log!(fetch, "Closed {}\n", BStr::new(self.url.href)); // the socket is closed, we need to unregister the abort tracker self.unregister_abort_tracker(); @@ -2255,6 +2313,8 @@ impl<'a> HTTPClient<'a> { self.fail(crate::Error::Aborted); return; } + // Before the tunnel goes: a tunnelled body is exempt. + let read_end = self.body_read_end(read_end); self.close_proxy_tunnel(true); let in_progress = self.is_in_progress(); if self.state.flags.is_redirect_pending { @@ -2264,7 +2324,10 @@ impl<'a> HTTPClient<'a> { self.do_redirect::(ctx, socket); return; } - if in_progress && self.state.is_body_complete_on_close() { + // A close-delimited body ends at the peer's FIN (RFC 9112 section 8): a reset leaves + // it incomplete. undici completes it (`onHttpSocketError` in client-h1.js, + // nodejs/undici#784), so Node's fetch resolves with the part that arrived. + if in_progress && self.state.is_body_complete_on_close(read_end) { if let Err(err) = self.state.finalize_body_on_eof() { self.fail(err); return; @@ -3153,7 +3216,8 @@ impl<'a> HTTPClient<'a> { if socket.is_closed() || socket.is_shutdown() { return Err(crate::Error::ConnectionClosed); } - let amount = write_to_socket::(socket, to_send)?; + let amount = + write_to_socket::(socket, to_send, &mut self.state.flags.write_took_error)?; if IS_FIRST_CALL { if amount == 0 { // don't worry about it @@ -3271,7 +3335,11 @@ impl<'a> HTTPClient<'a> { let to_send_len = buffer.slice().len(); if to_send_len > 0 { - let amount = write_to_socket::(socket, buffer.slice())?; + let amount = write_to_socket::( + socket, + buffer.slice(), + &mut self.state.flags.write_took_error, + )?; self.state.request_sent_len += amount; buffer.cursor += amount; if amount < to_send_len { @@ -3290,7 +3358,12 @@ impl<'a> HTTPClient<'a> { // ok we flushed all pending data so we can reset the backpressure if !data.is_empty() { // no backpressure everything was sended so we can just try to send - let sent = write_to_socket_with_buffer_fallback::(socket, buffer, data)?; + let sent = write_to_socket_with_buffer_fallback::( + socket, + buffer, + data, + &mut self.state.flags.write_took_error, + )?; self.state.request_sent_len += sent; // if we didn't send all the data we have backpressure return Ok(sent < data.len()); @@ -3507,9 +3580,14 @@ impl<'a> HTTPClient<'a> { match &mut self.state.original_request_body { HTTPRequestBody::Bytes(_) => { - let to_send = self.request_body(); + // The field, not `request_body()`: the write borrows the latch beside it. + let to_send = self.state.request_body.slice(); if !to_send.is_empty() { - let sent = match write_to_socket::(socket, to_send) { + let sent = match write_to_socket::( + socket, + to_send, + &mut self.state.flags.write_took_error, + ) { Ok(s) => s, Err(err) => { self.close_and_fail::(err, socket); @@ -4247,7 +4325,11 @@ impl<'a> HTTPClient<'a> { || self.proxy_tunnel.is_some() || self.flags.upgrade_state == HTTPUpgradeState::Upgraded || !self.signals.is_receive_paused() - || socket.is_closed_or_has_error() + || socket.is_closed() + || socket.is_shutdown() + // Under the rule, not SO_ERROR: reading it takes a pending reset out of the kernel, + // and the `recv()` after the pause then returns 0 like a FIN. + || (!STRICT_READ_END && socket.get_error() != 0) { return; } diff --git a/src/uws_sys/socket.rs b/src/uws_sys/socket.rs index 2694ff304583..0ac97bb87b14 100644 --- a/src/uws_sys/socket.rs +++ b/src/uws_sys/socket.rs @@ -286,6 +286,10 @@ impl NewSocketHandler { ) } + /// Reads AND clears `SO_ERROR`. Once it has taken a pending reset, the next + /// `recv()` returns 0 like a FIN, so a caller that still reads a body from + /// this socket must not call it: it is for the decision to keep or drop a + /// connection whose exchange is over. #[inline] pub fn is_closed_or_has_error(&self) -> bool { self.is_closed() || self.is_shutdown() || self.get_error() != 0 From 1d173c068bdc195058342588305e19634c2ff6e4 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Thu, 22 Oct 2026 03:32:03 +0000 Subject: [PATCH 2/2] test: a close-delimited fetch body ends at a FIN, not at a reset fetch-gzip.test.ts has the fetch matrix, beside the close-delimited tests of the same function. A body with no Content-Length and no chunked framing fails with ECONNRESET when a reset cuts it short. The strict assertions are for Linux. On other systems the read must settle. The group "still the end of the body" passes with and without the rule. fetch-syscall-fault.test.ts has the deterministic cases. A fault puts the errno of the reset in one send() or recv(). http-reset-errno.test.ts is a source lint. It keeps two lists closed: the writes of the HTTP/1 client to its socket, and the reads of SO_ERROR in src/http. The fetch error section of the docs states the rule and its limits. --- docs/runtime/networking/fetch.mdx | 1 + .../source-lints/http-reset-errno.test.ts | 77 ++ test/js/web/fetch/fetch-gzip.test.ts | 768 +++++++++++++++++- test/js/web/fetch/fetch-syscall-fault.test.ts | 363 ++++++++- 4 files changed, 1206 insertions(+), 3 deletions(-) create mode 100644 test/internal/source-lints/http-reset-errno.test.ts diff --git a/docs/runtime/networking/fetch.mdx b/docs/runtime/networking/fetch.mdx index 280c77533d9d..b538e7f73770 100644 --- a/docs/runtime/networking/fetch.mdx +++ b/docs/runtime/networking/fetch.mdx @@ -378,6 +378,7 @@ Bun's fetch implementation includes several specific error cases: - Using a request body with GET/HEAD methods throws an error (which is expected for the fetch API) - Using the `proxy` and `unix` options together throws an error - Connection failures reject with a `TypeError` whose `code` is the socket error (`ECONNREFUSED`, `ECONNRESET`, `ETIMEDOUT`, `ENOTFOUND`, ...). The message starts with the code. +- On Linux, a response with no `Content-Length` and no chunked encoding completes only when the server ends the connection with a FIN (on TLS, a `close_notify`). If the connection is reset first, the body read rejects with `ECONNRESET`, even when every byte of the body arrived. Node's `fetch` resolves such a body. This is about the connection that Bun itself holds (a proxy in between can hide a reset or add one), and a server avoids it when it sends `Content-Length` or uses chunked encoding. - TLS certificate validation failures when `rejectUnauthorized` is true (or undefined) - S3 operations may throw specific errors related to authentication or permissions diff --git a/test/internal/source-lints/http-reset-errno.test.ts b/test/internal/source-lints/http-reset-errno.test.ts new file mode 100644 index 000000000000..f03f30a5a58c --- /dev/null +++ b/test/internal/source-lints/http-reset-errno.test.ts @@ -0,0 +1,77 @@ +import { Glob } from "bun"; +import { expect, test } from "bun:test"; +import { readFileSync } from "node:fs"; +import path from "node:path"; + +// fetch() completes a response body with no Content-Length and no chunked framing only when the connection reached +// the end of the peer's stream (`is_body_complete_on_close`, src/http/InternalState.rs). It knows a reset from that +// end by the errno of the recv() that failed. The kernel gives the errno of a reset to one caller: the first recv(), +// send() or read of SO_ERROR after the reset. A read of SO_ERROR, or a send outside the helper, takes a reset's errno +// out of the kernel. The next recv() then returns 0, and the body that the reset cut short ends like a FIN. +// +// So these two lists are closed: +// 1. In the HTTP/1 client (src/http/lib.rs) only `write_once` writes to the socket. It keeps the ECONNRESET of a +// failed send for `HTTPClient::on_close`. Send a request with `write_to_socket`. Two writers are outside the +// rule and outside this list: sendfile() of a `Bun.file` upload takes the raw fd, and a CONNECT tunnel writes +// in src/http/ProxyTunnel.rs. +// 2. In src/http, SO_ERROR (`get_error`, `is_closed_or_has_error`) is read only by the functions in `readers`. +// A new reader needs the same proof as they have: no body can still end on that socket after the read. +const why = + "A read of SO_ERROR, or a send outside write_once, takes a reset's errno out of the kernel, " + + "and the close-delimited body that the reset cut short then ends like a FIN. See the comment in this file."; + +const repoRoot = path.resolve(import.meta.dir, "..", "..", ".."); + +// Every call that `pattern` finds in `file`, as ": ". +function calls(file: string, pattern: RegExp) { + // A comment line is not a call. + const code = readFileSync(path.join(repoRoot, file), "utf8").replace(/^[ \t]*\/\/.*$/gm, ""); + return [...code.matchAll(pattern)] + .map(match => { + const enclosing = [...code.slice(0, match.index).matchAll(/\bfn\s+(\w+)/g)].at(-1)?.[1]; + return `${enclosing ?? "(no function)"}: ${match[1]}`; + }) + .sort(); +} + +test("the HTTP/1 client writes to its socket in write_once only", () => { + const writes = calls("src/http/lib.rs", /\b\w*sock\w*\s*\.\s*(\w*write\w*)\s*\(/g); + expect(writes, why).toEqual(["write_once: write", "write_once: write_check_error"]); +}); + +test("src/http reads SO_ERROR only where no body can still end", async () => { + const readers = { + "src/http/HTTPContext.rs": [ + // A pooled socket, before a request takes it. + "find_in: get_error", + // The exchange is over: pool the socket or drop it. + "release_socket: get_error", + ], + "src/http/h2_client/ClientSession.rs": [ + // HTTP/2 frames its bodies: none of them ends when the connection closes. + "fail_all: is_closed_or_has_error", + "maybe_release: is_closed_or_has_error", + ], + "src/http/lib.rs": [ + // The response was a redirect: pool the socket or drop it. + "do_redirect: is_closed_or_has_error", + // Only where the rule is off (`!STRICT_READ_END`). + "maybe_pause_receive: get_error", + // The body is complete: resume the socket and pool it, or drop it. + "send_one_progress_update: is_closed_or_has_error", + "send_one_progress_update: is_closed_or_has_error", + ], + }; + + const found: Record = {}; + let scanned = 0; + for await (const rel of new Glob("**/*.rs").scan({ cwd: path.join(repoRoot, "src/http") })) { + const file = path.join("src/http", rel).replaceAll("\\", "/"); + scanned++; + const reads = calls(file, /\.\s*(get_error|is_closed_or_has_error)\s*\(/g); + if (reads.length > 0) found[file] = reads; + } + // Guard against repoRoot resolving wrong, which would make the lint pass vacuously. + expect(scanned).toBeGreaterThan(20); + expect(found, why).toEqual(readers); +}); diff --git a/test/js/web/fetch/fetch-gzip.test.ts b/test/js/web/fetch/fetch-gzip.test.ts index 8606499d528f..baff3e4ded48 100644 --- a/test/js/web/fetch/fetch-gzip.test.ts +++ b/test/js/web/fetch/fetch-gzip.test.ts @@ -1,9 +1,11 @@ import { Socket } from "bun"; import { beforeAll, describe, expect, it } from "bun:test"; -import { bunEnv, bunExe, gcTick } from "harness"; +import { bunEnv, bunExe, gcTick, isLinux, isWindows, tempDir, tls as tlsCert } from "harness"; import { once } from "node:events"; +import { truncateSync } from "node:fs"; import { createServer } from "node:http"; -import { createServer as createNetServer } from "node:net"; +import { connect, createServer as createNetServer } from "node:net"; +import { createServer as createTlsServer } from "node:tls"; import { brotliCompressSync, deflateRawSync, @@ -14,6 +16,7 @@ import { zstdCompressSync, } from "node:zlib"; import path from "path"; +import { upgradeHeaders } from "./websocket.helpers"; const gzipped = path.join(import.meta.dir, "fixture.html.gz"); const html = path.join(import.meta.dir, "fixture.html"); @@ -678,6 +681,50 @@ describe("corrupt compressed responses", () => { }); } + // A reset is not the FIN. On Linux the body read then fails with ECONNRESET. A stream that is short no + // longer fails with the error of its decoder. A stream that reached an end of its own is cut too: the + // server resets before the second gzip member, or the second zstd frame, leaves. + const cutByReset: [name: string, encoding: string, payload: Buffer][] = [ + ...Object.entries(codecs).map(([encoding, [half]]): [string, string, Buffer] => [ + `${encoding}: truncated stream`, + encoding, + half, + ]), + ["gzip: the first of two members, whole", "gzip", gzipSync(plain)], + ["zstd: the first of two frames, whole", "zstd", zstdCompressSync(plain)], + ]; + for (const [name, encoding, payload] of cutByReset) { + it(`${name}, then a reset: the body read rejects with ECONNRESET`, async () => { + const sockets: import("node:net").Socket[] = []; + const srv = createNetServer(sock => { + sock.on("error", () => {}); + sock.once("data", () => { + sockets.push(sock); + const head = `HTTP/1.1 200 OK\r\nContent-Encoding: ${encoding}\r\nConnection: close\r\n\r\n`; + sock.write(Buffer.concat([Buffer.from(head), payload])); + }); + }); + const port = await listen(srv); + try { + // The client has the head when fetch() resolves, and the payload came with the head. + const res = await fetch(`http://127.0.0.1:${port}/`); + for (const sock of sockets) sock.resetAndDestroy(); + const read = await res.arrayBuffer().then( + body => ({ resolved: body.byteLength }), + (error: Error & { code?: string }) => ({ rejected: error.constructor.name, code: error.code }), + ); + expect({ status: res.status, read }).toEqual({ + status: 200, + // fetch() has this rule on Linux and Android. Elsewhere the read only has to settle. + read: isLinux ? { rejected: "TypeError", code: "ECONNRESET" } : expect.anything(), + }); + } finally { + for (const sock of sockets) sock.destroy(); + srv.close(); + } + }); + } + it("identity: uncompressed close-delimited body is delivered intact", async () => { const body = Buffer.alloc(4096, "x"); const srv = createNetServer(sock => { @@ -728,6 +775,723 @@ describe("corrupt compressed responses", () => { }); }); +// RFC 9112 section 8: a response with no Content-Length and no chunked framing ends when the server closes +// the connection. A reset is not that close, so the body it cuts short fails with ECONNRESET. Node's fetch +// resolves such a body with the part that arrived. +describe("close-delimited body: a reset is not the end of the body", () => { + type NetSocket = import("node:net").Socket; + const HEAD = "HTTP/1.1 200 OK\r\nConnection: close\r\n\r\n"; + const trusted = { tls: { ca: tlsCert.cert } }; + + // A promise as a value to compare: what it resolved with, or the class and the code of its error. + const settled = (promise: Promise) => + promise.then( + value => ({ resolved: value }), + (error: Error & { code?: string }) => ({ rejected: error.constructor.name, code: error.code }), + ); + // The rule is compiled for Linux and Android. Elsewhere every close still ends such a body, so there the + // read of a cut body only has to settle. + const cut = isLinux ? { rejected: "TypeError", code: "ECONNRESET" } : expect.anything(); + // fetch() and the read of the body, for a server that resets on its own. Off Linux that reset can take the + // head with it (Windows drops what the client has yet to read), so there the exchange only has to settle. + const exchange = (response: Promise) => + settled(response.then(async res => ({ status: res.status, text: await settled(res.text()) }))); + const cutExchange = isLinux ? { resolved: { status: 200, text: cut } } : expect.anything(); + // What an exchange delivered: the text of the body, or "rejected" when fetch() or the read rejected. + const delivered = (outcome: any): string => outcome.resolved?.text?.resolved ?? "rejected"; + // A framed answer with no body. + const EMPTY = "HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"; + + // A raw server. `respond` runs once for each connection, when the request head is in: `request` is the + // head and the bytes of the body that came with it. + async function serve( + respond: (socket: NetSocket, request: string) => void, + { tls, unix }: { tls?: import("node:tls").TlsOptions; unix?: string } = {}, + ) { + const sockets = new Set(); + const onConnection = (socket: NetSocket) => { + sockets.add(socket); + socket.on("close", () => sockets.delete(socket)); + socket.on("error", () => {}); + let request = ""; + const onData = (data: Buffer) => { + request += data.toString("latin1"); + if (!request.includes("\r\n\r\n")) return; + socket.off("data", onData); + respond(socket, request); + }; + socket.on("data", onData); + }; + const server = tls + ? createTlsServer({ ...tlsCert, ...tls }, onConnection) + : createNetServer({ allowHalfOpen: true }, onConnection); + await once(unix ? server.listen(unix) : server.listen(0, "127.0.0.1"), "listening"); + const port = unix ? 0 : (server.address() as import("node:net").AddressInfo).port; + return { + port, + url: unix ? "http://localhost/" : `${tls ? "https" : "http"}://127.0.0.1:${port}/`, + // Resets every open connection: the client gets no FIN. + reset() { + for (const socket of sockets) if (!socket.destroyed) socket.resetAndDestroy(); + }, + // close() alone waits for the connections that are still open. + async [Symbol.asyncDispose]() { + for (const socket of sockets) socket.destroy(); + server.close(); + }, + }; + } + + // TLS has no call that resets a connection, or that ends it without a close_notify: a TCP relay in front + // of the TLS server does both. With `bareFin` the close_notify of the origin does not reach the client and + // its FIN does. With "then reset" the relay also leaves the upload unread from the answer on, so its close + // after that FIN resets. With `alertThenReset` the close_notify reaches the client, the upload stays unread + // too, and a reset takes the place of the FIN. All three need a TLS 1.2 origin, where the type of a record + // is readable. With `tunnel` the relay is a CONNECT proxy: it answers the CONNECT of the client first. + async function relayTo( + origin: { port: number }, + { bareFin, alertThenReset, tunnel }: { bareFin?: true | "then reset"; alertThenReset?: true; tunnel?: true } = {}, + ) { + const clients = new Set(); + let tunnels = 0; + const relay = createNetServer(client => { + const upstream = connect(origin.port, "127.0.0.1"); + clients.add(client); + client.on("error", () => {}); + upstream.on("error", () => {}); + client.on("close", () => { + clients.delete(client); + upstream.destroy(); + }); + // The client sends nothing after its CONNECT until the answer to it is in. + let connectHead = tunnel ? "" : undefined; + client.on("data", data => { + if (connectHead === undefined) return void upstream.write(data); + connectHead += data.toString("latin1"); + if (!connectHead.includes("\r\n\r\n")) return; + connectHead = undefined; + tunnels++; + client.write("HTTP/1.1 200 Connection established\r\n\r\n"); + }); + let records = Buffer.alloc(0); + // For `alertThenReset`: the records that the kernel has yet to take, and whether the origin has ended. + let unflushed = 0; + let originEnded = false; + const resetWhenFlushed = () => { + if (originEnded && unflushed === 0 && !client.destroyed) client.resetAndDestroy(); + }; + upstream.on("data", (data: Buffer) => { + if (!bareFin && !alertThenReset) return void client.write(data); + records = Buffer.concat([records, data]); + // A record: its type (1 byte), a version (2), a length (2), then that many bytes. + while (records.length >= 5 && records.length >= 5 + records.readUInt16BE(3)) { + const record = records.subarray(0, 5 + records.readUInt16BE(3)); + records = records.subarray(record.length); + if (record[0] === 21 && bareFin) continue; // an alert: the close_notify + // Application data: the answer. + if (record[0] === 23 && (bareFin === "then reset" || alertThenReset)) client.pause(); + if (!alertThenReset) { + client.write(record); + continue; + } + unflushed++; + client.write(record, () => { + unflushed--; + resetWhenFlushed(); + }); + } + }); + // Every byte of the origin is in by now, and the end() of `client` sends its FIN after them. + upstream.on("end", () => { + originEnded = true; + if (alertThenReset) resetWhenFlushed(); + else if (bareFin === "then reset") client.end(() => client.destroy()); + else if (bareFin) client.end(); + }); + }); + await once(relay.listen(0, "127.0.0.1"), "listening"); + const { port } = relay.address() as import("node:net").AddressInfo; + return { + url: `https://127.0.0.1:${port}/`, + // For the `proxy` option of fetch(), with `tunnel`. + proxy: `http://127.0.0.1:${port}`, + // The CONNECT requests that the relay answered. + get tunnels() { + return tunnels; + }, + // Resets every client: no close_notify and no FIN. + reset() { + for (const client of clients) if (!client.destroyed) client.resetAndDestroy(); + }, + async [Symbol.asyncDispose]() { + for (const client of clients) client.destroy(); + relay.close(); + }, + }; + } + + // A file for a Bun.file upload: 4 MiB, which is more than the socket takes while the server does not read. + // It is a file of holes, so that no data goes to the disk. + function uploadFile() { + const dir = tempDir("fetch-cut-upload", { upload: "" }); + const file = path.join(String(dir), "upload"); + truncateSync(file, 4 * 1024 * 1024); + return { file, [Symbol.dispose]: () => dir[Symbol.dispose]() }; + } + + // A request body that keeps coming, one chunk for each turn of the event loop, until the request is over. + function endlessUpload() { + let open = true; + const chunk = new Uint8Array(64 * 1024).fill(117); + const body = new ReadableStream({ + async pull(controller) { + await new Promise(resolve => setImmediate(resolve)); + if (open) controller.enqueue(chunk); + }, + cancel() { + open = false; + }, + }); + return { body, [Symbol.dispose]: () => void (open = false) }; + } + + // An ambient NO_PROXY that lists 127.0.0.1 makes fetch() ignore its `proxy` option. The HTTP client sees + // an assignment to process.env and not a delete, so this restores with an assignment too. + function withoutNoProxy() { + const saved = [process.env.NO_PROXY, process.env.no_proxy]; + process.env.NO_PROXY = process.env.no_proxy = ""; + return { + [Symbol.dispose]() { + process.env.NO_PROXY = saved[0] ?? ""; + process.env.no_proxy = saved[1] ?? ""; + }, + }; + } + + describe.each([ + ["HTTP/1.1 with Connection: close", HEAD], + ["HTTP/1.0", "HTTP/1.0 200 OK\r\n\r\n"], + ["HTTP/1.1 with no Connection: close and no length", "HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\n\r\n"], + ])("%s", (_, head) => { + // Each test resets the connection when fetch() has resolved: the client has the head and "half" by then. + const serveHalf = () => serve(socket => void socket.write(head + "half")); + + it("fetch() resolves with the head, then text() rejects", async () => { + await using server = await serveHalf(); + const res = await fetch(server.url); + server.reset(); + expect({ status: res.status, text: await settled(res.text()) }).toEqual({ status: 200, text: cut }); + }); + + it("a stream reader gets the chunk that arrived, then rejects", async () => { + await using server = await serveHalf(); + const reader = (await fetch(server.url)).body!.getReader(); + const first = await reader.read(); + server.reset(); + expect({ first: Buffer.from(first.value!).toString(), next: await settled(reader.read()) }).toEqual({ + first: "half", + next: cut, + }); + }); + + it("Bun.write() rejects", async () => { + await using server = await serveHalf(); + using dir = tempDir("fetch-cut-body", {}); + const res = await fetch(server.url); + server.reset(); + expect(await settled(Bun.write(path.join(String(dir), "body"), res))).toEqual(cut); + }); + }); + + it("the head, part of the body and the reset at once: fetch() resolves, then the read rejects", async () => { + await using server = await serve(socket => void socket.write(HEAD + "half", () => socket.resetAndDestroy())); + expect(await exchange(fetch(server.url))).toEqual(cutExchange); + }); + + // The accepted break. A reset is not the end of the body, and fetch() cannot know that every byte of the + // answer arrived. So the read also fails for a server that closes with a reset after a whole reply, with + // nothing of the request left unread. Node's fetch resolves both of these. + it.each([ + ["a whole reply", "the whole body"], + ["a reply with no body", ""], + ])("the accepted break: %s, then a reset with no FIN: the read rejects", async (_, body) => { + await using server = await serve(socket => void socket.write(HEAD + body, () => socket.resetAndDestroy())); + expect(await exchange(fetch(server.url))).toEqual(cutExchange); + }); + + it("a slow reader: 4 MiB, then the reset, and only then the read", async () => { + const answer = Buffer.concat([Buffer.from(HEAD), Buffer.alloc(4 * 1024 * 1024, "x")]); + // The reset comes when the kernel has taken the whole body, or at the latest when the client has the + // head: a kernel that does not buffer 4 MiB for a reader that stopped must not hang the test. + await using server = await serve(socket => void socket.write(answer, () => server.reset())); + const res = await fetch(server.url); + server.reset(); + const length = await settled(res.arrayBuffer().then(body => body.byteLength)); + expect({ status: res.status, length }).toEqual({ status: 200, length: cut }); + }); + + // The order that decides for a slow reader: the reset is pending when fetch() stops reading, which it does + // when 256 KiB of the body wait for a reader. The first part of the body stays below that mark. The part + // that crosses it and the reset then come in one turn of the server, so the client meets that order in + // most runs. fetch-syscall-fault.test.ts has it in every run, for builds with fault injection. + // Linux only: the order exists only under the rule, and the steps depend on when this client stops reading. + it.skipIf(!isLinux)( + "a slow reader: the part that crosses the pause mark and the reset at once: every run rejects", + async () => { + const first = Buffer.concat([Buffer.from(HEAD), Buffer.alloc(256 * 1024 - 1024, "x")]); + const outcomes: unknown[] = []; + for (let run = 0; run < 10; run++) { + const firstWritten = Promise.withResolvers(); + await using server = await serve((socket, request) => { + if (request.startsWith("GET /turn ")) return void socket.end(EMPTY); + // Or the kernel can hold the second part back until the client acknowledges the first one, and the + // reset then drops it. + socket.setNoDelay(true); + socket.write(first, () => firstWritten.resolve(socket)); + }); + // A request that starts now, when it is over: the HTTP thread of the client has taken what the other + // connection had for it before. + const turn = async () => void (await (await fetch(server.url + "turn")).arrayBuffer()); + const res = await fetch(server.url); + const socket = await firstWritten.promise; + await turn(); + socket.write(Buffer.alloc(2048, "y")); + socket.resetAndDestroy(); + await turn(); + outcomes.push({ status: res.status, length: await settled(res.arrayBuffer().then(body => body.byteLength)) }); + } + expect(outcomes).toEqual(Array(10).fill({ status: 200, length: cut })); + }, + ); + + it("a streamed upload in flight: every run rejects", async () => { + await using server = await serve(socket => { + socket.pause(); // the upload stays unread + socket.write(HEAD + "half", () => socket.resetAndDestroy()); + }); + const outcomes: unknown[] = []; + // The call of the client that meets the reset differs from run to run: a recv(), or the send() of a chunk. + for (let run = 0; run < 30; run++) { + using upload = endlessUpload(); + outcomes.push(await exchange(fetch(server.url, { method: "POST", body: upload.body, duplex: "half" }))); + } + expect(outcomes).toEqual(Array(30).fill(cutExchange)); + }); + + // A Bun.file upload over http goes out with sendfile(). A sendfile() that fails rejects fetch() with its + // own errno, so this upload is outside the rule: the error is not always ECONNRESET. One of the two calls + // meets the reset in every run, so no run delivers the cut body. + it("a Bun.file upload in flight: no run delivers the cut body", async () => { + using upload = uploadFile(); + await using server = await serve(socket => { + socket.pause(); // the upload stays unread + socket.write(HEAD + "half", () => socket.resetAndDestroy()); + }); + const outcomes: string[] = []; + for (let run = 0; run < 10; run++) { + const body = Bun.file(upload.file); + outcomes.push(delivered(await exchange(fetch(server.url, { method: "POST", body })))); + } + expect(outcomes).toEqual(Array(10).fill(isLinux ? "rejected" : expect.anything())); + }); + + it("https: fetch() resolves with the head, then text() rejects", async () => { + await using origin = await serve(socket => void socket.write(HEAD + "half"), { tls: {} }); + await using relay = await relayTo(origin); + const res = await fetch(relay.url, trusted); + relay.reset(); + expect({ status: res.status, text: await settled(res.text()) }).toEqual({ status: 200, text: cut }); + }); + + it("https: a reset after the server read the whole upload", async () => { + const upload = Buffer.alloc(1024 * 1024, "u"); + await using origin = await serve( + (socket, request) => { + let read = request.length - request.indexOf("\r\n\r\n") - 4; + const answerWhenWhole = () => void (read === upload.length && socket.write(HEAD + "half")); + socket.on("data", data => { + read += data.length; + answerWhenWhole(); + }); + answerWhenWhole(); + }, + { tls: {} }, + ); + await using relay = await relayTo(origin); + const res = await fetch(relay.url, { ...trusted, method: "POST", body: upload }); + relay.reset(); + expect({ status: res.status, text: await settled(res.text()) }).toEqual({ status: 200, text: cut }); + }); + + // A TLS connection that had a send() rejected keeps the old rule: its close does not say how the read side + // ended. So a run in which a send() of the upload meets the reset first still resolves with the part that + // arrived, as before. Every other run rejects. + it.each([ + ["does not read", (socket: NetSocket) => void socket.pause()], + ["reads", (socket: NetSocket) => void socket.resume()], + ])( + "https, a streamed upload in flight that the server %s: a run rejects, or ends as before", + async (_, takeUpload) => { + await using origin = await serve( + socket => { + takeUpload(socket); + socket.write(HEAD + "half"); + }, + { tls: {} }, + ); + await using relay = await relayTo(origin); + const unexpected: unknown[] = []; + for (let run = 0; run < 10; run++) { + using upload = endlessUpload(); + const res = await fetch(relay.url, { ...trusted, method: "POST", body: upload.body, duplex: "half" }); + relay.reset(); + const text = (await settled(res.text())) as { resolved?: string; code?: string }; + const asTheRuleSays = text.code === "ECONNRESET"; + const asBefore = text.resolved === "half"; + if (res.status !== 200 || (isLinux && !asTheRuleSays && !asBefore)) + unexpected.push({ status: res.status, text }); + } + expect(unexpected).toEqual([]); + }, + ); + + // fetch() sends an idempotent request again when a connection from the keep-alive pool closes before the + // response began. A reset in the body is not that close: the request went out once, and the read rejects. + it.each([ + ["part of the body", "half"], + ["the head and no byte of the body", ""], + ])("a connection from the keep-alive pool, reset after %s: the request is not sent again", async (_, part) => { + // ":" for each request head, in the order in which the server got them. + const seen: string[] = []; + const sockets = new Set(); + let connections = 0; + let cutSocket: NetSocket | undefined; + const server = createNetServer(socket => { + const id = ++connections; + sockets.add(socket); + socket.on("close", () => sockets.delete(socket)); + socket.on("error", () => {}); + let buffered = ""; + socket.on("data", data => { + buffered += data.toString("latin1"); + for (let end; (end = buffered.indexOf("\r\n\r\n")) !== -1;) { + const requestPath = buffered.slice(0, end).split(" ")[1]; + buffered = buffered.slice(end + 4); + seen.push(`${id}:${requestPath}`); + if (requestPath !== "/cut") { + socket.write("HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok"); + } else if (!cutSocket) { + cutSocket = socket; + socket.write("HTTP/1.1 200 OK\r\n\r\n" + part); + } else { + // A second /cut, which must not come. It ends with a FIN, so the read settles and `seen` shows it. + socket.end("HTTP/1.1 200 OK\r\n\r\n" + part); + } + } + }); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + try { + const base = `http://127.0.0.1:${(server.address() as import("node:net").AddressInfo).port}`; + const warm = await (await fetch(base + "/warm")).text(); + const res = await fetch(base + "/cut"); + cutSocket!.resetAndDestroy(); + const text = await settled(res.text()); + const after = await (await fetch(base + "/after")).text(); + expect({ warm, status: res.status, text, after, seen }).toEqual({ + warm: "ok", + status: 200, + text: cut, + after: "ok", + // /cut went out once, on the connection of /warm. The other connection is the one of /after. + seen: ["1:/warm", "1:/cut", "2:/after"], + }); + } finally { + for (const socket of sockets) socket.destroy(); + server.close(); + } + }); + + it("an unread cut body is no unhandled rejection, and a late text() rejects at once", async () => { + const script = /* js */ ` + import { once } from "node:events"; + import { createServer } from "node:net"; + + const sockets = []; + const server = createServer(socket => { + socket.on("error", () => {}); + socket.once("data", () => { + socket.write(${JSON.stringify(HEAD + "half")}); + sockets.push(socket); + }); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + const url = "http://127.0.0.1:" + server.address().port + "/"; + const settled = promise => + promise.then( + text => "resolved " + text, + error => error.constructor.name + " " + error.code, + ); + + const unread = await fetch(url); + const watched = await fetch(url); + // In the order of the requests: when the second read has settled, the client has seen the first reset too. + for (const socket of sockets) socket.resetAndDestroy(); + const watchedRead = await settled(watched.text()); + // An unhandled rejection ends the process in one of these turns. + for (let turn = 0; turn < 10; turn++) await new Promise(resolve => setImmediate(resolve)); + const late = unread.text(); + const lateIs = Bun.peek.status(late); + console.log(JSON.stringify({ status: unread.status, watched: watchedRead, lateIs, late: await settled(late) })); + server.close(); + `; + await using proc = Bun.spawn({ cmd: [bunExe(), "-e", script], env: bunEnv, stdout: "pipe", stderr: "pipe" }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + let result: unknown = stdout; + try { + result = JSON.parse(stdout); + } catch {} + expect({ result, stderr, exitCode }).toEqual({ + result: isLinux + ? { status: 200, watched: "TypeError ECONNRESET", lateIs: "rejected", late: "TypeError ECONNRESET" } + : expect.objectContaining({ status: 200 }), + stderr: "", + exitCode: 0, + }); + }); + + // `return new Response(upstream.body)`: the proxy must not turn the cut into a complete chunked message. + it("a Bun.serve proxy of a cut body reports it, and does not terminate its own body as complete", async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + `import net from "node:net"; + // The upstream resets only once "chunk-a" has provably reached the + // client of the proxy. + const chunkReachedClient = Promise.withResolvers(); + const upstream = net.createServer(socket => { + socket.on("error", () => {}); + socket.once("data", () => { + socket.write("HTTP/1.1 200 OK\\r\\nConnection: close\\r\\n\\r\\nchunk-a"); + chunkReachedClient.promise.then(() => socket.resetAndDestroy()); + }); + }); + await new Promise(resolve => upstream.listen(0, "127.0.0.1", resolve)); + const server = Bun.serve({ + port: 0, + development: false, + async fetch() { + const res = await fetch("http://127.0.0.1:" + upstream.address().port + "/"); + return new Response(res.body); + }, + }); + const wire = await new Promise(resolve => { + const chunks = []; + const sock = net.connect(server.port, "127.0.0.1", () => { + sock.write("GET / HTTP/1.1\\r\\nHost: x\\r\\nConnection: close\\r\\n\\r\\n"); + }); + sock.on("data", d => { + chunks.push(d); + if (Buffer.concat(chunks).includes("chunk-a")) chunkReachedClient.resolve(); + }); + sock.on("error", () => {}); + sock.on("close", () => resolve(Buffer.concat(chunks).toString("latin1"))); + }); + console.log(JSON.stringify({ + statusLine: wire.split("\\r\\n")[0], + cleanChunkedTerminator: wire.endsWith("0\\r\\n\\r\\n"), + body: wire.split("\\r\\n\\r\\n").slice(1).join("\\r\\n\\r\\n"), + })); + server.stop(true); + upstream.close();`, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ result: stdout.trim() ? JSON.parse(stdout) : stdout, exitCode }).toEqual({ + result: isLinux + ? { statusLine: "HTTP/1.1 200 OK", cleanChunkedTerminator: false, body: "7\r\nchunk-a\r\n" } + : // Either end of the body is right here: the request only has to finish. + expect.objectContaining({ statusLine: "HTTP/1.1 200 OK" }), + exitCode: 0, + }); + // The failure must still reach the operator. + if (isLinux) expect(stderr).toContain("ECONNRESET"); + }); + + // These pass with and without the rule: they pin the ends of a connection that still complete the body. + describe("still the end of the body", () => { + const WHOLE = "the whole body"; + const whole = { resolved: WHOLE }; + // The answer, a FIN (on TLS after a close_notify), and a close with the upload unread: that close resets. + const endThenReset = (socket: NetSocket) => { + socket.pause(); + socket.end(HEAD + WHOLE, () => socket.destroy()); + }; + // Uploads that such a server does not read. What the client meets first differs from run to run: the + // end of the stream, the reset, or a send() that fails with EPIPE. + async function uploads(runs: number, bytes: number, url: string, init: BunFetchRequestInit = {}) { + const body = Buffer.alloc(bytes, "u"); + const outcomes: unknown[] = []; + for (let run = 0; run < runs; run++) { + outcomes.push(await settled(fetch(url, { ...init, method: "POST", body }).then(res => res.text()))); + } + return outcomes; + } + + it("a FIN", async () => { + await using server = await serve(socket => void socket.end(HEAD + WHOLE)); + expect(await settled(fetch(server.url).then(res => res.text()))).toEqual(whole); + }); + + // The next three and the unix socket one are not for Windows: it drops what the client has yet to read + // when the reset arrives, and that can be the answer (fetch.tls.test.ts skips "reads a response that the + // server sent before it reset the upload" for this). + it.skipIf(isWindows)("a FIN, then a reset", async () => { + await using server = await serve(endThenReset); + expect(await uploads(10, 4 * 1024 * 1024, server.url)).toEqual(Array(10).fill(whole)); + }); + + // A Bun.file upload is outside the rule (see "a Bun.file upload in flight" above): a sendfile() that + // meets the reset rejects fetch(), as it did before the rule. No run resolves with less than the whole + // body. + it.skipIf(isWindows)("a FIN, then a reset, with a Bun.file upload in flight: never a short body", async () => { + using upload = uploadFile(); + await using server = await serve(endThenReset); + const outcomes: string[] = []; + for (let run = 0; run < 10; run++) { + const body = Bun.file(upload.file); + outcomes.push(delivered(await exchange(fetch(server.url, { method: "POST", body })))); + } + expect(outcomes.filter(body => body !== WHOLE && body !== "rejected")).toEqual([]); + }); + + it.skipIf(isWindows)("https: a close_notify, then a reset", async () => { + await using server = await serve(endThenReset, { tls: {} }); + expect(await uploads(6, 4 * 1024 * 1024, server.url, trusted)).toEqual(Array(6).fill(whole)); + }); + + // The close_notify of the origin ended its stream. Here the reset that follows it comes with no FIN, so + // a send() of the upload can fail with ECONNRESET before the client reads the close_notify. The body is + // whole all the same. + it.skipIf(isWindows)("https: a close_notify, then a reset with no FIN", async () => { + await using origin = await serve(socket => void socket.end(HEAD + WHOLE), { tls: { maxVersion: "TLSv1.2" } }); + await using relay = await relayTo(origin, { alertThenReset: true }); + expect(await uploads(6, 8 * 1024 * 1024, relay.url, trusted)).toEqual(Array(6).fill(whole)); + }); + + it.skipIf(isWindows)("https: a FIN with no close_notify, then a reset", async () => { + await using origin = await serve(socket => void socket.end(HEAD + WHOLE), { tls: { maxVersion: "TLSv1.2" } }); + await using relay = await relayTo(origin, { bareFin: "then reset" }); + expect(await uploads(4, 8 * 1024 * 1024, relay.url, trusted)).toEqual(Array(4).fill(whole)); + }); + + it("https: a FIN with no close_notify", async () => { + await using origin = await serve(socket => void socket.end(HEAD + WHOLE), { tls: { maxVersion: "TLSv1.2" } }); + await using relay = await relayTo(origin, { bareFin: true }); + expect(await settled(fetch(relay.url, trusted).then(res => res.text()))).toEqual(whole); + }); + + // A raw node:net server on a socket path, like the ones in fetch.unix.test.ts. On a unix socket a reset + // only says that the peer closed with input unread: it comes for a whole answer too. + it.skipIf(isWindows)("over a unix socket: the answer, an end, then a close with the upload unread", async () => { + using dir = tempDir("fetch-cut-unix", {}); + const unix = path.join(String(dir), "server.sock"); + await using server = await serve(endThenReset, { unix }); + expect(await uploads(10, 8 * 1024 * 1024, server.url, { unix })).toEqual(Array(10).fill(whole)); + }); + + // With no end() the client gets a reset after the answer in every run: the upload is unread. + it.skipIf(isWindows)("over a unix socket: the answer, then a close with the upload unread and no end", async () => { + using dir = tempDir("fetch-cut-unix", {}); + const unix = path.join(String(dir), "server.sock"); + await using server = await serve( + socket => { + socket.pause(); + socket.write(HEAD + WHOLE, () => socket.destroy()); + }, + { unix }, + ); + expect(await uploads(10, 8 * 1024 * 1024, server.url, { unix })).toEqual(Array(10).fill(whole)); + }); + + // Behind a CONNECT proxy the connection of the client ends with the teardown of the proxy, whatever the + // origin did. The rule does not apply to a body that comes through such a tunnel: a reset ends it. + it("through a CONNECT proxy: the whole answer, then the reset of the proxy", async () => { + // The origin leaves its connection open after the answer, so the reset of the proxy is the end. + await using origin = await serve(socket => void socket.write(HEAD + WHOLE), { tls: {} }); + await using proxy = await relayTo(origin, { tunnel: true }); + using _ = withoutNoProxy(); + const res = await fetch(origin.url, { ...trusted, proxy: proxy.proxy }); + const reader = res.body!.getReader(); + const first = await reader.read(); + proxy.reset(); + expect({ + status: res.status, + tunnels: proxy.tunnels, + first: Buffer.from(first.value!).toString(), + next: await settled(reader.read()), + }).toEqual({ status: 200, tunnels: 1, first: WHOLE, next: { resolved: { done: true, value: undefined } } }); + }); + + it("a reset of a 101-upgraded stream: done is true", async () => { + await using server = await serve( + socket => + void socket.write( + "HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n\r\nfirst", + ), + ); + const ended = Promise.withResolvers(); + const res = await fetch(server.url, { + method: "GET", + headers: upgradeHeaders(), + // The upload side of the stream stays open until the read side has ended. + async *body() { + await ended.promise; + }, + }); + const reader = res.body!.getReader(); + const first = await reader.read(); + server.reset(); + const next = await settled(reader.read()); + ended.resolve(); + expect({ status: res.status, first: Buffer.from(first.value!).toString(), next }).toEqual({ + status: 101, + first: "first", + next: { resolved: { done: true, value: undefined } }, + }); + }); + + it("chunked: a reset after the last chunk, before the CRLF that ends the trailers", async () => { + await using server = await serve( + socket => void socket.write("HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nHello\r\n0\r\n"), + ); + const res = await fetch(server.url); + server.reset(); + expect(await settled(res.text())).toEqual({ resolved: "Hello" }); + }); + + it("gzip with a Content-Length: a reset after the last byte, before a reader decodes the body", async () => { + // One libdeflate call inflates a gzip body of this size, so the client holds it whole and undecoded + // until a reader attaches. + const plain = Buffer.alloc(1024 * 1024, "The quick brown fox jumps over the lazy dog. "); + const wire = gzipSync(plain); + const head = `HTTP/1.1 200 OK\r\nContent-Encoding: gzip\r\nContent-Length: ${wire.length}\r\n\r\n`; + await using server = await serve(socket => void socket.write(Buffer.concat([Buffer.from(head), wire]))); + const res = await fetch(server.url); + server.reset(); + // A few turns of the event loop let the client meet the reset before a reader attaches. The body is + // whole in either order. + for (let turn = 0; turn < 10; turn++) await new Promise(resolve => setImmediate(resolve)); + const body = Buffer.from(await res.arrayBuffer()); + expect({ length: body.length, same: body.equals(plain) }).toEqual({ length: plain.length, same: true }); + }); + }); +}); + // RFC 1952 §2.2: a gzip file is a sequence of members. Concatenated members // (cat a.gz b.gz, bgzf, pigz, pre-compressed segment stitching) must all be // decoded. Previously fetch() silently returned only the first member. diff --git a/test/js/web/fetch/fetch-syscall-fault.test.ts b/test/js/web/fetch/fetch-syscall-fault.test.ts index ab19c6cce627..f6101482be82 100644 --- a/test/js/web/fetch/fetch-syscall-fault.test.ts +++ b/test/js/web/fetch/fetch-syscall-fault.test.ts @@ -1,6 +1,9 @@ import { socketFaultInjection as fault } from "bun:internal-for-testing"; import { afterEach, describe, expect, test } from "bun:test"; -import { bunEnv, bunExe, tls as certs, isWindows } from "harness"; +import { bunEnv, bunExe, tls as certs, isLinux, isWindows } from "harness"; +import { once } from "node:events"; +import net from "node:net"; +import tls from "node:tls"; const skip = !fault.available() || isWindows; @@ -57,6 +60,232 @@ async function runClientFetch( return { ...JSON.parse(stdout.trim() || "{}"), stderr, exitCode, signal: proc.signalCode }; } +// The client for the close-delimited tests. It uploads a stream, reads the first piece of the answer, and +// only then arms `rule` and writes a chunk of the upload: nothing else is in flight, so the rule meets the +// send() of that chunk, or the recv() of what the server answers to it. +// spill: https. A send() that returns 0 first leaves the record of the chunk spilled in the TLS layer, +// so `rule` meets the send() that drains the spill. +// askToEnd: a request for /end follows the chunk. See `serveInTwoPieces`. +// buffer: the upload is a Buffer of this many bytes, not a stream. `rule` is armed before fetch(): with +// `after: 1` it passes the send() of the head and the start of the body, and meets the send() of +// the rest of the body, which follows at once. +async function runUploadingFetch( + url: string, + rule: import("bun:internal-for-testing").SocketFaultRule, + opts: { ca?: string; spill?: boolean; askToEnd?: boolean; buffer?: number } = {}, +) { + const fixture = /* js */ ` + const { socketFaultInjection: fault } = require("bun:internal-for-testing"); + const url = ${JSON.stringify(url)}; + const rule = ${JSON.stringify(rule)}; + const spill = ${JSON.stringify(!!opts.spill)}; + const buffer = ${JSON.stringify(opts.buffer ?? 0)}; + const tls = ${opts.ca ? "{ ca: process.env.CA }" : "undefined"}; + const firstRead = Promise.withResolvers(); + let pulls = 0; + const stream = new ReadableStream( + { + async pull(controller) { + if (++pulls === 1) { + await firstRead.promise; + fault.set(spill ? { syscall: "send", action: "zero", repeat: -1 } : rule); + controller.enqueue(new Uint8Array(spill ? 64 * 1024 : 1024)); + } else if (spill && pulls === 2) { + // The sink asks for more: TLS took the chunk and holds its record. The next send() drains it. + fault.set(rule); + } + }, + }, + { highWaterMark: 0 }, + ); + let result, status; + try { + if (buffer) fault.set(rule); + const upload = buffer ? Buffer.alloc(buffer, "u") : stream; + const res = await fetch(url, { method: "POST", body: upload, duplex: "half", tls }); + status = res.status; + const reader = res.body.getReader(); + let length = (await reader.read()).value.length; + firstRead.resolve(); + ${ + opts.askToEnd + ? `// One turn of the event loop: the pump has handed the chunk to the HTTP thread by then, and that + // thread writes what it was handed before it starts a request. + await new Promise(resolve => setImmediate(resolve)); + // A failure of this request must not look like the failure of the body read. + await fetch(url + "end", { tls }) + .then(res => res.arrayBuffer()) + .catch(e => Promise.reject(new Error("the request for /end failed: " + e)));` + : "" + } + for (;;) { + const { value, done } = await reader.read(); + if (done) break; + length += value.length; + } + result = { ok: true, status, length }; + } catch (e) { + result = { ok: false, status, code: e?.code, name: e?.name, message: String(e?.message ?? e) }; + } finally { + fault.clear(); + } + console.log(JSON.stringify(result)); + `; + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", fixture], + env: { ...bunEnv, BUN_DEBUG_QUIET_LOGS: "1", ...(opts.ca ? { CA: opts.ca } : {}) }, + stderr: "pipe", + stdout: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + return { ...JSON.parse(stdout.trim() || "{}"), stderr, exitCode, signal: proc.signalCode }; +} + +// A raw server for a close-delimited answer in two pieces. "first" goes out when the request head is in. +// "rest" and the FIN follow when bytes of the upload arrive, or when a request for /end does: a chunk whose +// send() failed never reaches the server. +async function serveInTwoPieces(secure = false) { + const sockets = new Set(); + let answer: net.Socket | undefined; + let ended = false; + const end = () => { + if (!ended) answer?.end("rest"); + ended = true; + }; + const onConnection = (socket: net.Socket) => { + sockets.add(socket); + socket.on("close", () => sockets.delete(socket)); + socket.on("error", () => {}); + let head = ""; + socket.on("data", data => { + if (socket === answer) return end(); + head += data.toString("latin1"); + if (!head.includes("\r\n\r\n")) return; + if (head.startsWith("GET /end ")) { + end(); + return void socket.end("HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"); + } + answer = socket; + socket.write("HTTP/1.1 200 OK\r\nConnection: close\r\n\r\nfirst"); + }); + }; + const server = secure + ? tls.createServer({ key: certs.key, cert: certs.cert }, onConnection) + : net.createServer(onConnection); + await once(server.listen(0, "127.0.0.1"), "listening"); + return { + url: `${secure ? "https" : "http"}://127.0.0.1:${(server.address() as net.AddressInfo).port}/`, + // close() alone waits for the connections that are still open. + async [Symbol.asyncDispose]() { + for (const socket of sockets) socket.destroy(); + server.close(); + }, + }; +} + +// A raw server for a close-delimited answer in two parts. The first part stays 4096 bytes below the 256 KiB +// of unread body at which fetch() stops reading, and the second part crosses that mark with its last bytes. +// Each request for / starts a new answer. +// / the head and the first part +// /first answered when the kernel has taken the whole first part +// /cut the second part on the connection of / and its reset, one call after the other, then the answer +// to /cut +async function serveAcrossTheMark() { + const MARK = 256 * 1024; + const OK = "HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"; + const sockets = new Set(); + let firstPartWritten = Promise.withResolvers(); + let answer: net.Socket | undefined; + const server = net.createServer(socket => { + sockets.add(socket); + socket.on("close", () => sockets.delete(socket)); + socket.on("error", () => {}); + let head = ""; + const onData = (data: Buffer) => { + head += data.toString("latin1"); + if (!head.includes("\r\n\r\n")) return; + socket.off("data", onData); + if (head.startsWith("GET /first ")) { + firstPartWritten.promise.then(() => socket.end(OK)); + } else if (head.startsWith("GET /cut ")) { + answer!.write(Buffer.alloc(4096 + 64, "y")); + answer!.resetAndDestroy(); + socket.end(OK); + } else { + answer = socket; + const written = (firstPartWritten = Promise.withResolvers()); + // Or the kernel can hold the second part back until the client acknowledges the first one, and + // the reset then drops it. + socket.setNoDelay(true); + const first = Buffer.concat([ + Buffer.from("HTTP/1.1 200 OK\r\nConnection: close\r\n\r\n"), + Buffer.alloc(MARK - 4096, "x"), + ]); + socket.write(first, () => written.resolve()); + } + }; + socket.on("data", onData); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + return { + url: `http://127.0.0.1:${(server.address() as net.AddressInfo).port}/`, + // close() alone waits for the connections that are still open. + async [Symbol.asyncDispose]() { + for (const socket of sockets) socket.destroy(); + server.close(); + }, + }; +} + +// The client for `serveAcrossTheMark`. In each round nothing reads the body until the server has reset the +// connection. The server resets one call after it wrote the second part. When it stalls between the two (a +// collection of garbage is enough), the client reaches the mark first, and that round shows nothing: hence +// three rounds. +async function runSlowReader(url: string) { + const fixture = /* js */ ` + const { socketFaultInjection: fault } = require("bun:internal-for-testing"); + const url = ${JSON.stringify(url)}; + // A request that is not the one under test. Its failure must not look like the failure of the body read. + const ask = path => + fetch(url + path) + .then(res => res.arrayBuffer()) + .catch(e => Promise.reject(new Error("the request for /" + path + " failed: " + e))); + const rounds = []; + let failed; + try { + for (let round = 0; round < 3; round++) { + const res = await fetch(url); + // The client has the first part when this request is over: the server answers it after that part. + await ask("first"); + // One byte for each recv() from here on. The client needs 4096 of them to reach the mark, and the + // reset follows the second part at once. + fault.set({ syscall: "recv", action: "short", bytes: 1, repeat: -1 }); + await ask("cut"); + fault.clear(); + rounds.push( + await res.arrayBuffer().then( + body => ({ ok: true, status: res.status, length: body.byteLength }), + e => ({ ok: false, status: res.status, code: e?.code, name: e?.name }), + ), + ); + } + } catch (e) { + failed = String(e?.message ?? e); + } finally { + fault.clear(); + } + console.log(JSON.stringify({ rounds, failed })); + `; + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", fixture], + env: { ...bunEnv, BUN_DEBUG_QUIET_LOGS: "1" }, + stderr: "pipe", + stdout: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + return { ...JSON.parse(stdout.trim() || "{}"), stderr, exitCode, signal: proc.signalCode }; +} + describe.skipIf(skip)("fetch() under injected syscall faults (http)", () => { test("recv → ECONNRESET on response rejects with a connection error", async () => { using server = Bun.serve({ @@ -252,6 +481,138 @@ describe.skipIf(skip)("fetch() under injected syscall faults (https)", () => { }); }); +// A body with no Content-Length and no chunked framing ends at the FIN of the server. These are the +// deterministic twins of the resets in fetch-gzip.test.ts: a reset reaches the client as the errno of one +// call, and the fault puts that errno in a call of our choice. +describe.skipIf(skip)("fetch() of a close-delimited body under injected syscall faults", () => { + // fetch() has this rule on Linux and Android. Elsewhere the body ends as before: the read has to settle. + const cut = isLinux ? { ok: false, name: "TypeError", code: "ECONNRESET" } : {}; + const settled = { status: 200, signal: null, exitCode: 0 }; + + // A send() that fails with ECONNRESET takes the reset out of the kernel: the recv() after it returns 0, + // like a FIN. The body is not complete for that. + test("send → ECONNRESET, then the FIN of the server: the body read rejects", async () => { + await using server = await serveInTwoPieces(); + const rule = { syscall: "send", action: "errno", errno: "ECONNRESET", repeat: 1 } as const; + expect(await runUploadingFetch(server.url, rule, { askToEnd: true })).toMatchObject({ ...cut, ...settled }); + }); + + // On Linux EPIPE says that the FIN of the server came before its reset. This passes without the rule too. + test("send → EPIPE, then the FIN of the server: the whole body resolves", async () => { + await using server = await serveInTwoPieces(); + const rule = { syscall: "send", action: "errno", errno: "EPIPE", repeat: 1 } as const; + expect(await runUploadingFetch(server.url, rule, { askToEnd: true })).toMatchObject({ + ok: true, + length: "firstrest".length, + ...settled, + }); + }); + + // Only ECONNRESET is the reset. A send() also fails for good with a timeout, and that can come after the + // FIN of the server: the body is whole then. + test("send → ETIMEDOUT, then the FIN of the server: the whole body resolves", async () => { + await using server = await serveInTwoPieces(); + const rule = { syscall: "send", action: "errno", errno: "ETIMEDOUT", repeat: 1 } as const; + expect(await runUploadingFetch(server.url, rule, { askToEnd: true })).toMatchObject({ + ok: true, + length: "firstrest".length, + ...settled, + }); + }); + + // uSockets retries a send() whose errno is not in its list of "the peer is gone" (socket.c), so fetch() + // does not see that errno. A cut whose errno such a send() took still ends like a FIN. + test("outside the rule: send → ECONNREFUSED, then the FIN of the server: the body resolves", async () => { + await using server = await serveInTwoPieces(); + const rule = { syscall: "send", action: "errno", errno: "ECONNREFUSED", repeat: 1 } as const; + expect(await runUploadingFetch(server.url, rule, { askToEnd: true })).toMatchObject({ + ok: true, + length: "firstrest".length, + ...settled, + }); + }); + + // uSockets rejects a send() that reports more bytes than it was given, and then closes the connection + // itself (socket.c, loop.c). That close is not the FIN of the server either. + test("send → more bytes than it was given: the body read rejects", async () => { + await using server = await serveInTwoPieces(); + const rule = { syscall: "send", action: "excess", repeat: 1 } as const; + expect(await runUploadingFetch(server.url, rule)).toMatchObject({ ...cut, ...settled }); + }); + + // An upload that is a Buffer goes out in its own calls of the client: the head with the start of the body, + // then the rest of the body. 1 MiB is more than the first of the two takes. + test("Buffer upload: send → ECONNRESET, then the FIN of the server: the body read rejects", async () => { + await using server = await serveInTwoPieces(); + const rule = { syscall: "send", action: "errno", errno: "ECONNRESET", after: 1, repeat: 1 } as const; + expect(await runUploadingFetch(server.url, rule, { askToEnd: true, buffer: 1024 * 1024 })).toMatchObject({ + ...cut, + ...settled, + }); + }); + + test("Buffer upload: send → EPIPE, then the FIN of the server: the whole body resolves", async () => { + await using server = await serveInTwoPieces(); + const rule = { syscall: "send", action: "errno", errno: "EPIPE", after: 1, repeat: 1 } as const; + expect(await runUploadingFetch(server.url, rule, { askToEnd: true, buffer: 1024 * 1024 })).toMatchObject({ + ok: true, + length: "firstrest".length, + ...settled, + }); + }); + + test("recv → ECONNRESET after the first piece: the body read rejects", async () => { + await using server = await serveInTwoPieces(); + const rule = { syscall: "recv", action: "errno", errno: "ECONNRESET", repeat: -1 } as const; + expect(await runUploadingFetch(server.url, rule)).toMatchObject({ ...cut, ...settled }); + }); + + // A TLS connection that had a send() rejected keeps the old rule. uSockets closes it with ECONNRESET + // whatever the errno of that send() was, so the close does not say how the read side ended. These pass + // without the rule too. The ECONNRESET rows are outside the rule: that send() met a reset with no FIN. + // + // uSockets sends a spilled record by itself, from the writable event. + test.each(["ECONNRESET", "EPIPE"] as const)( + "https: the send of a spilled record → %s: the body resolves", + async errno => { + await using server = await serveInTwoPieces(true); + const rule = { syscall: "send", action: "errno", errno, repeat: 1 } as const; + expect(await runUploadingFetch(server.url, rule, { ca: certs.cert, spill: true })).toMatchObject({ + ok: true, + length: "first".length, + ...settled, + }); + }, + ); + + // The send() of a chunk that TLS writes at once. + test.each(["ECONNRESET", "EPIPE", "ETIMEDOUT"] as const)( + "https: the send of a chunk → %s: the body resolves", + async errno => { + await using server = await serveInTwoPieces(true); + const rule = { syscall: "send", action: "errno", errno, repeat: 1 } as const; + expect(await runUploadingFetch(server.url, rule, { ca: certs.cert })).toMatchObject({ + ok: true, + length: "first".length, + ...settled, + }); + }, + ); + + // fetch() stops reading when 256 KiB of the body wait for a reader. It must not look for an error on the + // socket at that moment: a read of SO_ERROR takes a reset that is pending, and the recv() after it then + // returns 0, like a FIN. The client gets one byte for each recv() while the server writes the part of the + // body that crosses that mark and resets, so the reset is pending when the client reaches the mark. + test("a reset that is pending when fetch() stops reading: the body read rejects", async () => { + await using server = await serveAcrossTheMark(); + expect(await runSlowReader(server.url)).toMatchObject({ + rounds: Array(3).fill({ ...cut, status: 200 }), + signal: null, + exitCode: 0, + }); + }); +}); + describe.skipIf(skip)("fetch() seeded short-I/O fuzz", () => { const seed = Number(process.env.BUN_SOCKET_FUZZ_SEED ?? 0x4e7c) >>> 0 || 1; function makePrng(s: number) {