From bf07709ace62ffecebce22ff0baa40c8f63585f3 Mon Sep 17 00:00:00 2001 From: Louis Ponet Date: Mon, 24 Aug 2026 17:12:44 +0100 Subject: [PATCH] feat(network): add poll-driven HTTP server and client over TCP Introduce flux_network::http::HttpNetwork, a single-event-loop HTTP/1.x layer on top of TcpNetwork. One instance can listen for inbound requests and maintain persistent outbound endpoints; parsed requests and responses are delivered borrowed through poll_with, and responses may be deferred across polls (one pending request per connection, pipelining buffered). Server side supports Content-Length request bodies, automatic Expect: 100-continue, HEAD semantics, keep-alive/close negotiation, and protocol errors (400/413/431/501) handled before user code. Client side supports Content-Length, chunked, and EOF-delimited response bodies with interim-response skipping. Head/body/header-count limits and an idle timeout for accepted connections are configurable. To carry HTTP on the wire, TcpNetwork gains a per-group Framing choice: the existing length-prefixed framing (default) or new raw passthrough which caps read chunks at max_frame_size and skips latency telemetry (no send timestamps). Also adds disconnect_when_drained so a server can flush a final response before closing, with sends rejected while draining. The ByteQueue remainder path is generalized to optional headers to serve both framings. httparse (workspace dep) does head parsing; body handling stays in flux. --- Cargo.lock | 7 + Cargo.toml | 1 + crates/flux-network/Cargo.toml | 1 + crates/flux-network/src/http.rs | 939 +++++++++++++++++++ crates/flux-network/src/lib.rs | 2 + crates/flux-network/src/tcp/mod.rs | 2 +- crates/flux-network/src/tcp/network.rs | 262 ++++-- crates/flux-network/tests/http.rs | 872 +++++++++++++++++ crates/flux-network/tests/tcp_network_raw.rs | 307 ++++++ 9 files changed, 2327 insertions(+), 66 deletions(-) create mode 100644 crates/flux-network/src/http.rs create mode 100644 crates/flux-network/tests/http.rs create mode 100644 crates/flux-network/tests/tcp_network_raw.rs diff --git a/Cargo.lock b/Cargo.lock index 5b1b6b3..de3ce06 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -649,6 +649,7 @@ dependencies = [ "flux-communication", "flux-timing", "flux-utils", + "httparse", "libc", "mio", "serde", @@ -956,6 +957,12 @@ version = "0.5.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c" +[[package]] +name = "httparse" +version = "1.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" + [[package]] name = "humantime" version = "2.3.0" diff --git a/Cargo.toml b/Cargo.toml index 9cc69f9..3d78b16 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -56,6 +56,7 @@ crossterm = "0.28.1" ctrlc = "3" directories = "5.0.1" governor = "0.6.3" +httparse = "1" humantime = "2.1" indexmap = { features = ["serde"], version = "2.11.1" } libc = "0.2.177" diff --git a/crates/flux-network/Cargo.toml b/crates/flux-network/Cargo.toml index dcdb859..e218f62 100644 --- a/crates/flux-network/Cargo.toml +++ b/crates/flux-network/Cargo.toml @@ -13,6 +13,7 @@ flux.workspace = true flux-communication.workspace = true flux-timing.workspace = true flux-utils.workspace = true +httparse.workspace = true libc.workspace = true mio.workspace = true serde.workspace = true diff --git a/crates/flux-network/src/http.rs b/crates/flux-network/src/http.rs new file mode 100644 index 0000000..bbef668 --- /dev/null +++ b/crates/flux-network/src/http.rs @@ -0,0 +1,939 @@ +//! Poll-driven HTTP over [`crate::tcp::TcpNetwork`]. +//! +//! [`HttpNetwork`] can listen for requests and maintain outbound endpoints in +//! one event loop. Events borrow parsed data only for the callback duration. +//! +//! ```no_run +//! use std::net::SocketAddr; +//! use flux_network::http::{HttpEvent, HttpNetwork}; +//! let mut http = HttpNetwork::default(); +//! http.listen("127.0.0.1:8080".parse::().unwrap())?; +//! let peer = http.connect("127.0.0.1:8081".parse::().unwrap()); +//! loop { +//! let mut response = None; +//! let mut request = false; +//! http.poll_with(|event| match event { +//! HttpEvent::Request { token, .. } => response = Some(token), +//! HttpEvent::Connected { token } if token == peer => request = true, +//! _ => {} +//! }); +//! if let Some(token) = response { http.respond(token, 200, &[], b"hello"); } +//! if request { http.request(peer, "GET", "/", &[], &[]); } +//! } +//! # Ok::<(), std::io::Error>(()) +//! ``` +//! +//! # Limitations +//! HTTP/1.1 and HTTP/1.0 responses are supported. Request bodies require +//! `Content-Length`; chunked requests are rejected with `501`. Response bodies +//! may use `Content-Length`, chunked transfer coding, or EOF delimiting. +//! `Expect: 100-continue` is handled automatically. +//! +//! TLS, HTTP/2, compression, trailer exposure, upgrades, and `WebSockets` are +//! not supported. Valid response trailers are parsed and discarded. There is no +//! half-close support. After an error response, the connection closes without a +//! lingering-close delay. Pipelined requests are served strictly one at a time +//! per connection. + +use std::{ + io::{self, Write as _}, + net::SocketAddr, +}; + +use flux_timing::{Duration, Instant}; +use mio::Token; + +use crate::tcp::{Framing, TcpEvent, TcpGroup, TcpGroupConfig, TcpNetwork}; + +pub enum HttpEvent<'a> { + Accepted { token: Token, peer_addr: SocketAddr }, + Connected { token: Token }, + Response { token: Token, response: HttpResponse<'a> }, + Request { token: Token, request: HttpRequest<'a> }, + Disconnected { token: Token }, +} +#[derive(Clone, Copy)] +enum State { + Idle, + Pending, + Draining, +} +enum Role { + Accepted { state: State, close: bool, continued: bool, head_request: bool }, + Outbound { addr: SocketAddr, method: Option }, +} +struct Conn { + token: Token, + buf: Vec, + dirty: bool, + over_limit: bool, + last_activity: Instant, + role: Role, +} +#[derive(Clone, Copy)] +enum Lifecycle { + Connected(Token, Option), + Disconnected(Token), +} +pub struct HttpNetwork { + network: TcpNetwork, + group: Option, + name: &'static str, + max_head_bytes: usize, + max_body_bytes: usize, + max_headers: usize, + idle_timeout: Option, + socket_buf_size: Option, + conns: Vec, + lifecycle: Vec, +} +impl Default for HttpNetwork { + fn default() -> Self { + Self { + network: TcpNetwork::default(), + group: None, + name: "http", + max_head_bytes: 16 * 1024, + max_body_bytes: 1024 * 1024, + max_headers: 64, + idle_timeout: Some(Duration::from_secs(30)), + socket_buf_size: None, + conns: Vec::new(), + lifecycle: Vec::new(), + } + } +} +impl HttpNetwork { + /// Sets the TCP group name. + pub fn with_name(mut self, name: &'static str) -> Self { + assert!(self.group.is_none(), "configure before listen or connect"); + self.name = name; + self + } + /// Sets the maximum message head size before rejecting it. + pub fn with_max_head_bytes(mut self, max_head_bytes: usize) -> Self { + assert!(self.group.is_none(), "configure before listen or connect"); + self.max_head_bytes = max_head_bytes; + self + } + /// Sets the maximum message body size before rejecting it. + pub fn with_max_body_bytes(mut self, max_body_bytes: usize) -> Self { + assert!(self.group.is_none(), "configure before listen or connect"); + self.max_body_bytes = max_body_bytes; + self + } + /// Sets the maximum number of request headers accepted. + pub fn with_max_headers(mut self, max_headers: usize) -> Self { + assert!(self.group.is_none(), "configure before listen or connect"); + self.max_headers = max_headers; + self + } + /// Sets the TCP socket buffer size. + pub fn with_socket_buf_size(mut self, socket_buf_size: usize) -> Self { + assert!(self.group.is_none(), "configure before listen or connect"); + self.socket_buf_size = Some(socket_buf_size); + self + } + /// Sets the idle timeout for accepted connections; outbound endpoints + /// remain persistent. + pub fn with_idle_timeout(mut self, idle_timeout: Duration) -> Self { + assert!(self.group.is_none(), "configure before listen or connect"); + self.idle_timeout = Some(idle_timeout); + self + } + /// Disables the idle connection sweep. + pub fn without_idle_timeout(mut self) -> Self { + assert!(self.group.is_none(), "configure before listen or connect"); + self.idle_timeout = None; + self + } + fn group(&mut self) -> TcpGroup { + let Self { network, group, name, max_head_bytes, max_body_bytes, socket_buf_size, .. } = + self; + *group.get_or_insert_with(|| { + network.add_group(TcpGroupConfig { + name, + framing: Framing::Raw, + socket_buf_size: *socket_buf_size, + max_frame_size: usize::MAX, + max_backlog_bytes: Some(max_head_bytes.saturating_add(*max_body_bytes)), + backlog_warn_bytes: None, + ..Default::default() + }) + }) + } + pub fn listen(&mut self, addr: SocketAddr) -> io::Result<()> { + let group = self.group(); + self.network.listen(group, addr) + } + /// Immediately disconnects an accepted client. + pub fn disconnect(&mut self, token: Token) -> bool { + self.group.is_some() && + self.conns + .iter() + .any(|conn| conn.token == token && matches!(conn.role, Role::Accepted { .. })) && + self.network.disconnect(token) + } + /// Polls the network and delivers connection, request, and response events. + /// + /// A request event remains pending until [`Self::respond`] is called. The + /// handler may defer that call until a later poll; requests behind it stay + /// buffered until the response is sent. + pub fn poll_with(&mut self, mut handler: F) + where + F: for<'a> FnMut(HttpEvent<'a>), + { + let Some(group) = self.group else { return }; + let limit = self.buffer_limit(); + let conns = &mut self.conns; + let lifecycle = &mut self.lifecycle; + self.network.poll_with(|event| match event { + TcpEvent::Accepted { group: event_group, token, peer_addr } if event_group == group => { + conns.push(Conn { + token, + buf: Vec::new(), + dirty: false, + over_limit: false, + last_activity: Instant::now(), + role: Role::Accepted { + state: State::Idle, + close: false, + continued: false, + head_request: false, + }, + }); + lifecycle.push(Lifecycle::Connected(token, Some(peer_addr))); + } + TcpEvent::Connected { group: event_group, token, .. } if event_group == group => { + lifecycle.push(Lifecycle::Connected(token, None)); + } + TcpEvent::Message { group: event_group, token, payload, .. } + if event_group == group => + { + if let Some(conn) = conns.iter_mut().find(|conn| conn.token == token) && + !is_draining(&conn.role) && + !conn.over_limit + { + let available = limit.saturating_sub(conn.buf.len()); + if payload.len() > available { + conn.buf.extend_from_slice(&payload[..available]); + conn.over_limit = true; + } else { + conn.buf.extend_from_slice(payload); + } + conn.dirty = true; + conn.last_activity = Instant::now(); + } + } + TcpEvent::Disconnected { group: event_group, token, .. } if event_group == group => { + lifecycle.push(Lifecycle::Disconnected(token)); + } + _ => {} + }); + for event in std::mem::take(&mut self.lifecycle) { + self.emit_lifecycle(event, &mut handler); + } + self.parse_dirty(&mut handler); + for conn in &mut self.conns { + if conn.over_limit && !is_draining(&conn.role) { + self.network.disconnect(conn.token); + conn.over_limit = false; + } + } + if let Some(timeout) = self.idle_timeout { + let expired: Vec<_> = self + .conns + .iter() + .filter(|conn| { + matches!(conn.role, Role::Accepted { .. }) && + conn.last_activity.elapsed() >= timeout + }) + .map(|conn| conn.token) + .collect(); + for token in expired { + self.network.disconnect(token); + } + } + } + fn emit_lifecycle(&mut self, event: Lifecycle, handler: &mut F) + where + F: for<'a> FnMut(HttpEvent<'a>), + { + match event { + Lifecycle::Connected(token, Some(peer_addr)) => { + handler(HttpEvent::Accepted { token, peer_addr }); + } + Lifecycle::Connected(token, None) => handler(HttpEvent::Connected { token }), + Lifecycle::Disconnected(token) => { + if let Some(i) = self.conns.iter().position(|conn| conn.token == token) { + if self.conns[i].dirty { + self.parse_connection(i, handler); + } + if matches!(self.conns[i].role, Role::Outbound { .. }) { + self.parse_eof_outbound(i, handler); + self.conns[i].buf.clear(); + self.conns[i].dirty = false; + set_outbound_method(&mut self.conns[i].role, None); + } else { + self.conns.remove(i); + } + } + handler(HttpEvent::Disconnected { token }); + } + } + } + fn parse_dirty(&mut self, handler: &mut F) + where + F: for<'a> FnMut(HttpEvent<'a>), + { + for i in 0..self.conns.len() { + if self.conns[i].dirty { + self.parse_connection(i, handler); + } + } + } + fn parse_connection(&mut self, i: usize, handler: &mut F) + where + F: for<'a> FnMut(HttpEvent<'a>), + { + if matches!(self.conns[i].role, Role::Accepted { .. }) { + self.parse_and_emit(i, handler); + } else { + self.parse_outbound(i, handler); + } + } + pub fn connect(&mut self, addr: SocketAddr) -> Token { + let group = self.group(); + let token = self.network.connect(group, addr); + self.conns.push(Conn { + token, + buf: Vec::new(), + dirty: false, + over_limit: false, + last_activity: Instant::now(), + role: Role::Outbound { addr, method: None }, + }); + token + } + /// Permanently removes an outbound endpoint and stops it reconnecting. + pub fn remove(&mut self, token: Token) -> bool { + if self.group.is_none() || + !self + .conns + .iter() + .any(|conn| conn.token == token && matches!(conn.role, Role::Outbound { .. })) || + !self.network.remove(token) + { + return false + } + self.conns.retain(|conn| conn.token != token); + true + } + /// Queues one request on an outbound endpoint. + pub fn request( + &mut self, + token: Token, + method: &str, + path: &str, + headers: &[(&str, &str)], + body: &[u8], + ) -> bool { + if !valid_token(method) || + path.is_empty() || + path.contains(['\r', '\n', ' ']) || + headers.iter().any(|(n, v)| { + !valid_token(n) || + v.contains(['\r', '\n']) || + n.eq_ignore_ascii_case("content-length") || + n.eq_ignore_ascii_case("transfer-encoding") + }) + { + return false + } + let Some(c) = + self.conns.iter_mut().find(|c| c.token == token && outbound_method(&c.role).is_none()) + else { + return false + }; + let Role::Outbound { addr, .. } = &c.role else { return false }; + let host = addr.to_string(); + let sent = self.network.send_with(token, |out| { + write!(out, "{method} {path} HTTP/1.1\r\n").unwrap(); + let mut has_host = false; + for (n, v) in headers { + has_host |= n.eq_ignore_ascii_case("host"); + out.extend_from_slice(n.as_bytes()); + out.extend_from_slice(b": "); + out.extend_from_slice(v.as_bytes()); + out.extend_from_slice(b"\r\n"); + } + if !has_host { + write!(out, "Host: {host}\r\n").unwrap(); + } + write!(out, "Content-Length: {}\r\n\r\n", body.len()).unwrap(); + out.extend_from_slice(body); + }); + if sent { + set_outbound_method(&mut c.role, Some(method.to_owned())); + } + sent + } + fn fail_outbound(&mut self, i: usize) { + let token = self.conns[i].token; + self.conns[i].buf.clear(); + set_outbound_method(&mut self.conns[i].role, None); + self.network.disconnect(token); + } + fn buffer_limit(&self) -> usize { + self.max_head_bytes.saturating_add(self.max_body_bytes) + } + fn parse_and_emit(&mut self, i: usize, handler: &mut F) + where + F: for<'a> FnMut(HttpEvent<'a>), + { + if !matches!(accepted_state_mut(&mut self.conns[i].role), State::Idle) { + return + } + let over_limit = self.conns[i].over_limit; + let buf = &self.conns[i].buf; + let mut hs = vec![httparse::EMPTY_HEADER; self.max_headers]; + let mut req = httparse::Request::new(&mut hs); + let Ok(state) = req.parse(buf) else { + self.error(i, 400); + return + }; + let httparse::Status::Complete(head) = state else { + // A partial parse means every buffered byte is still head bytes. + if !crlf_only(buf) { + self.error(i, 400); + } else if over_limit || buf.len() > self.max_head_bytes { + self.error(i, 431); + } + return + }; + if !crlf_only(&buf[..head]) { + self.error(i, 400); + return + } + if head > self.max_head_bytes { + self.error(i, 431); + return + } + let Some(len) = request_content_length(req.headers) else { + self.error(i, 400); + return + }; + if req.headers.iter().any(|h| h.name.eq_ignore_ascii_case("transfer-encoding")) { + self.error(i, 501); + return + } + if len > self.max_body_bytes { + self.error(i, 413); + return + } + let Some(end) = head.checked_add(len) else { + self.error(i, 413); + return + }; + if buf.len() < end { + if over_limit { + self.error(i, 413); + return + } + if has_token(req.headers, "expect", b"100-continue") && + !accepted_continued_mut(&self.conns[i].role) + { + let token = self.conns[i].token; + if self + .network + .send_with(token, |out| write!(out, "HTTP/1.1 100 Continue\r\n\r\n").unwrap()) + { + set_accepted_continued(&mut self.conns[i].role, true); + } + } + return + } + let close = req.version == Some(0) && !has_token(req.headers, "connection", b"keep-alive") || + has_token(req.headers, "connection", b"close"); + let token = self.conns[i].token; + let head_request = req.method == Some("HEAD"); + let request = HttpRequest { + method: req.method.unwrap_or(""), + path: req.path.unwrap_or(""), + version: req.version.unwrap_or(1), + headers: req.headers, + body: &buf[head..end], + }; + handler(HttpEvent::Request { token, request }); + self.conns[i].buf.drain(..end); + self.conns[i].dirty = !self.conns[i].buf.is_empty(); + set_accepted_state(&mut self.conns[i].role, State::Pending); + set_accepted_close(&mut self.conns[i].role, close); + set_accepted_continued(&mut self.conns[i].role, false); + set_accepted_head_request(&mut self.conns[i].role, head_request); + } + fn error(&mut self, i: usize, status: u16) { + set_accepted_state(&mut self.conns[i].role, State::Pending); + set_accepted_close(&mut self.conns[i].role, true); + let token = self.conns[i].token; + let _ = self.respond(token, status, &[], &[]); + } + /// Sends the response for a pending request and returns whether it was + /// queued. + /// + /// Call this from a request handler or later from the poll loop. Each call + /// completes exactly one request for `token`. + pub fn respond( + &mut self, + token: Token, + status: u16, + headers: &[(&str, &str)], + body: &[u8], + ) -> bool { + let Some(i) = self + .conns + .iter() + .position(|c| c.token == token && matches!(accepted_state(&c.role), State::Pending)) + else { + return false + }; + if !(200..=599).contains(&status) || + headers.iter().any(|(n, v)| { + !valid_token(n) || + v.contains(['\r', '\n']) || + n.eq_ignore_ascii_case("content-length") || + n.eq_ignore_ascii_case("transfer-encoding") + }) + { + return false + } + let caller_close = headers.iter().any(|(n, v)| { + n.eq_ignore_ascii_case("connection") && has_value_token(v.as_bytes(), b"close") + }); + let close = accepted_close(&self.conns[i].role) || caller_close; + let suppress_body = + accepted_head_request(&self.conns[i].role) || matches!(status, 100..=199 | 204 | 304); + let include_length = !matches!(status, 100..=199 | 204); + let ok = self.network.send_with(token, |out| { + write!(out, "HTTP/1.1 {status} {}\r\n", reason_phrase(status)).unwrap(); + // Caller Connection headers only feed the close decision; exactly + // one canonical Connection header is always written below. + for (n, v) in headers { + if n.eq_ignore_ascii_case("connection") { + continue + } + out.extend_from_slice(n.as_bytes()); + out.extend_from_slice(b": "); + out.extend_from_slice(v.as_bytes()); + out.extend_from_slice(b"\r\n"); + } + if include_length { + write!(out, "Content-Length: {}\r\n", body.len()).unwrap(); + } + out.extend_from_slice(if close { + b"Connection: close\r\n" + } else { + b"Connection: keep-alive\r\n" + }); + out.extend_from_slice(b"\r\n"); + if !suppress_body { + out.extend_from_slice(body); + } + }); + if ok { + self.conns[i].dirty = !close && !self.conns[i].buf.is_empty(); + set_accepted_state( + &mut self.conns[i].role, + if close { State::Draining } else { State::Idle }, + ); + if close { + self.network.disconnect_when_drained(token); + } + } + ok + } + fn parse_outbound(&mut self, i: usize, handler: &mut F) + where + F: for<'a> FnMut(HttpEvent<'a>), + { + self.conns[i].dirty = false; + while outbound_method(&self.conns[i].role).is_some() { + let b = &self.conns[i].buf; + let mut hs = vec![httparse::EMPTY_HEADER; self.max_headers]; + let mut response = httparse::Response::new(&mut hs); + let parsed = response.parse(b); + let Ok(state) = parsed else { + self.fail_outbound(i); + return + }; + let httparse::Status::Complete(head) = state else { + // A partial parse means every buffered byte is still head bytes. + if !crlf_only(b) || b.len() > self.max_head_bytes { + self.fail_outbound(i); + } + return + }; + if !crlf_only(&b[..head]) || head > self.max_head_bytes { + self.fail_outbound(i); + return + } + let status = response.code.unwrap_or(0); + let no_body = outbound_method(&self.conns[i].role).map(String::as_str) == Some("HEAD") || + matches!(status, 100..=199 | 204 | 304); + let chunked = transfer_chunked(response.headers); + let content_length = response_content_length(response.headers); + if status == 101 || + matches!(content_length, ContentLength::Invalid) || + chunked.is_none() || + (chunked == Some(true) && !matches!(content_length, ContentLength::Absent)) || + (!no_body && + matches!(content_length, ContentLength::Present(length) if length > self.max_body_bytes)) + { + self.fail_outbound(i); + return + } + let (consumed, decoded) = if no_body { + (head, None) + } else if chunked == Some(true) { + match Self::decode_chunked(&b[head..], self.max_body_bytes, self.max_headers) { + Ok(Some((consumed, decoded))) => { + let Some(consumed) = head.checked_add(consumed) else { + self.fail_outbound(i); + return + }; + (consumed, Some(decoded)) + } + Ok(None) => return, + Err(()) => { + self.fail_outbound(i); + return + } + } + } else if let ContentLength::Present(length) = content_length { + let Some(consumed) = head.checked_add(length) else { + self.fail_outbound(i); + return + }; + if b.len() < consumed { + return + } + (consumed, None) + } else { + return + }; + if status < 200 { + self.conns[i].buf.drain(..consumed); + self.conns[i].dirty = !self.conns[i].buf.is_empty(); + continue + } + let token = self.conns[i].token; + let close = response.version == Some(0) && + !has_token(response.headers, "connection", b"keep-alive") || + has_token(response.headers, "connection", b"close"); + let response_event = HttpResponse { + version: response.version.unwrap_or(1), + status, + reason: response.reason.unwrap_or(""), + headers: response.headers, + body: if no_body { + &[] + } else if let Some(decoded) = decoded.as_deref() { + decoded + } else { + &b[head..consumed] + }, + }; + handler(HttpEvent::Response { token, response: response_event }); + self.conns[i].buf.drain(..consumed); + self.conns[i].dirty = !self.conns[i].buf.is_empty(); + set_outbound_method(&mut self.conns[i].role, None); + if close { + self.network.disconnect(token); + return + } + } + } + fn parse_eof_outbound(&self, i: usize, handler: &mut F) + where + F: for<'a> FnMut(HttpEvent<'a>), + { + if outbound_method(&self.conns[i].role).is_none() { + return + } + let b = &self.conns[i].buf; + let mut headers = vec![httparse::EMPTY_HEADER; self.max_headers]; + let mut response = httparse::Response::new(&mut headers); + let Ok(httparse::Status::Complete(head)) = response.parse(b) else { return }; + if !crlf_only(&b[..head]) || head > self.max_head_bytes { + return + } + let status = response.code.unwrap_or(0); + let no_body = outbound_method(&self.conns[i].role).map(String::as_str) == Some("HEAD") || + matches!(status, 100..=199 | 204 | 304); + if no_body || + transfer_chunked(response.headers) != Some(false) || + !matches!(response_content_length(response.headers), ContentLength::Absent) || + b.len() - head > self.max_body_bytes + { + return + } + let token = self.conns[i].token; + handler(HttpEvent::Response { + token, + response: HttpResponse { + version: response.version.unwrap_or(1), + status, + reason: response.reason.unwrap_or(""), + headers: response.headers, + body: &b[head..], + }, + }); + } + fn decode_chunked( + bytes: &[u8], + max_body_bytes: usize, + max_headers: usize, + ) -> Result)>, ()> { + let Some((end, body_len)) = Self::chunked_end(bytes, max_body_bytes, max_headers)? else { + return Ok(None) + }; + let mut body = Vec::with_capacity(body_len); + let mut at = 0; + while at < end { + let httparse::Status::Complete((consumed, size)) = + httparse::parse_chunk_size(&bytes[at..]).map_err(|_| ())? + else { + return Err(()) + }; + at = at.checked_add(consumed).ok_or(())?; + let size = usize::try_from(size).map_err(|_| ())?; + if size == 0 { + return Ok(Some((end, body))) + } + body.extend_from_slice(&bytes[at..at + size]); + at = at.checked_add(size + 2).ok_or(())?; + } + Err(()) + } + fn chunked_end( + bytes: &[u8], + max_body_bytes: usize, + max_headers: usize, + ) -> Result, ()> { + let mut at = 0; + let mut body_len = 0; + loop { + let httparse::Status::Complete((consumed, size)) = + httparse::parse_chunk_size(&bytes[at..]).map_err(|_| ())? + else { + return Ok(None) + }; + let size = usize::try_from(size).map_err(|_| ())?; + if size > max_body_bytes.saturating_sub(body_len) { + return Err(()) + } + at = at.checked_add(consumed).ok_or(())?; + if size == 0 { + let mut headers = vec![httparse::EMPTY_HEADER; max_headers]; + let httparse::Status::Complete((consumed, _)) = + httparse::parse_headers(&bytes[at..], &mut headers).map_err(|_| ())? + else { + return Ok(None) + }; + let end = at.checked_add(consumed).ok_or(())?; + if !crlf_only(&bytes[at..end]) { + return Err(()) + } + return Ok(Some((end, body_len))) + } + let Some(chunk_end) = at.checked_add(size).and_then(|at| at.checked_add(2)) else { + return Err(()) + }; + if bytes.len() < chunk_end || &bytes[at + size..chunk_end] != b"\r\n" { + return Ok(None) + } + body_len += size; + at = chunk_end; + } + } +} + +fn crlf_only(bytes: &[u8]) -> bool { + bytes.iter().enumerate().all(|(i, b)| *b != b'\n' || i > 0 && bytes[i - 1] == b'\r') +} +fn request_content_length(headers: &[httparse::Header<'_>]) -> Option { + let mut length = None; + for header in headers.iter().filter(|h| h.name.eq_ignore_ascii_case("content-length")) { + let value = std::str::from_utf8(header.value).ok()?; + if value.is_empty() || !value.bytes().all(|b| b.is_ascii_digit()) { + return None + } + let parsed = value.parse().ok()?; + if length.replace(parsed).is_some_and(|previous| previous != parsed) { + return None + } + } + Some(length.unwrap_or(0)) +} +fn has_token(headers: &[httparse::Header<'_>], name: &str, value: &[u8]) -> bool { + headers + .iter() + .filter(|h| h.name.eq_ignore_ascii_case(name)) + .any(|h| has_value_token(h.value, value)) +} +fn has_value_token(value: &[u8], wanted: &[u8]) -> bool { + value.split(|b| *b == b',').any(|part| part.trim_ascii().eq_ignore_ascii_case(wanted)) +} +fn valid_token(value: &str) -> bool { + !value.is_empty() && + value.bytes().all(|b| b.is_ascii_alphanumeric() || b"!#$%&'*+-.^_`|~".contains(&b)) +} +fn accepted_state(role: &Role) -> State { + match role { + Role::Accepted { state, .. } => *state, + Role::Outbound { .. } => State::Draining, + } +} +fn accepted_state_mut(role: &mut Role) -> &mut State { + match role { + Role::Accepted { state, .. } => state, + Role::Outbound { .. } => panic!("accepted role"), + } +} +fn is_draining(role: &Role) -> bool { + matches!(role, Role::Accepted { state: State::Draining, .. }) +} +fn set_accepted_state(role: &mut Role, state: State) { + *accepted_state_mut(role) = state; +} +fn accepted_close(role: &Role) -> bool { + matches!(role, Role::Accepted { close: true, .. }) +} +fn set_accepted_close(role: &mut Role, close: bool) { + if let Role::Accepted { close: current, .. } = role { + *current = close; + } +} +fn accepted_continued_mut(role: &Role) -> bool { + matches!(role, Role::Accepted { continued: true, .. }) +} +fn set_accepted_continued(role: &mut Role, continued: bool) { + if let Role::Accepted { continued: current, .. } = role { + *current = continued; + } +} +fn accepted_head_request(role: &Role) -> bool { + matches!(role, Role::Accepted { head_request: true, .. }) +} +fn set_accepted_head_request(role: &mut Role, head_request: bool) { + if let Role::Accepted { head_request: current, .. } = role { + *current = head_request; + } +} +fn outbound_method(role: &Role) -> Option<&String> { + match role { + Role::Outbound { method, .. } => method.as_ref(), + Role::Accepted { .. } => None, + } +} +fn set_outbound_method(role: &mut Role, method: Option) { + if let Role::Outbound { method: current, .. } = role { + *current = method; + } +} + +enum ContentLength { + Absent, + Present(usize), + Invalid, +} +fn response_content_length(headers: &[httparse::Header<'_>]) -> ContentLength { + let mut length = None; + for header in headers.iter().filter(|h| h.name.eq_ignore_ascii_case("content-length")) { + let Ok(value) = std::str::from_utf8(header.value) else { return ContentLength::Invalid }; + if value.is_empty() || !value.bytes().all(|b| b.is_ascii_digit()) { + return ContentLength::Invalid + } + let Ok(parsed) = value.parse() else { return ContentLength::Invalid }; + if length.replace(parsed).is_some_and(|previous| previous != parsed) { + return ContentLength::Invalid + } + } + length.map_or(ContentLength::Absent, ContentLength::Present) +} +fn transfer_chunked(headers: &[httparse::Header<'_>]) -> Option { + let mut found = false; + for value in headers + .iter() + .filter(|h| h.name.eq_ignore_ascii_case("transfer-encoding")) + .flat_map(|h| h.value.split(|b| *b == b',')) + { + if value.trim_ascii().eq_ignore_ascii_case(b"chunked") && !found { + found = true; + } else { + return None + } + } + Some(found) +} + +/// HTTP reason phrase for common status codes. +pub fn reason_phrase(status: u16) -> &'static str { + match status { + 100 => "Continue", + 200 => "OK", + 201 => "Created", + 204 => "No Content", + 400 => "Bad Request", + 401 => "Unauthorized", + 403 => "Forbidden", + 404 => "Not Found", + 405 => "Method Not Allowed", + 408 => "Request Timeout", + 411 => "Length Required", + 413 => "Payload Too Large", + 431 => "Request Header Fields Too Large", + 500 => "Internal Server Error", + 501 => "Not Implemented", + 503 => "Service Unavailable", + _ => "Unknown", + } +} + +pub struct HttpRequest<'a> { + pub method: &'a str, + pub path: &'a str, + pub version: u8, + pub headers: &'a [httparse::Header<'a>], + pub body: &'a [u8], +} +impl<'a> HttpRequest<'a> { + pub fn header(&self, name: &str) -> Option<&'a [u8]> { + self.headers.iter().find(|h| h.name.eq_ignore_ascii_case(name)).map(|h| h.value) + } +} +pub struct HttpResponse<'a> { + pub version: u8, + pub status: u16, + pub reason: &'a str, + pub headers: &'a [httparse::Header<'a>], + pub body: &'a [u8], +} +impl<'a> HttpResponse<'a> { + pub fn header(&self, name: &str) -> Option<&'a [u8]> { + self.headers.iter().find(|h| h.name.eq_ignore_ascii_case(name)).map(|h| h.value) + } +} + +#[cfg(test)] +mod tests { + use super::HttpRequest; + + #[test] + fn header_lookup_is_case_insensitive() { + let headers = [httparse::Header { name: "Content-Type", value: b"text/plain" }]; + let request = + HttpRequest { method: "GET", path: "/", version: 1, headers: &headers, body: &[] }; + assert_eq!(request.header("content-type"), Some(&b"text/plain"[..])); + } +} diff --git a/crates/flux-network/src/lib.rs b/crates/flux-network/src/lib.rs index 0a7a7f9..159dbb3 100644 --- a/crates/flux-network/src/lib.rs +++ b/crates/flux-network/src/lib.rs @@ -1,2 +1,4 @@ +pub mod http; + pub mod tcp; pub use mio::Token; diff --git a/crates/flux-network/src/tcp/mod.rs b/crates/flux-network/src/tcp/mod.rs index 1badf9b..ef3bbc0 100644 --- a/crates/flux-network/src/tcp/mod.rs +++ b/crates/flux-network/src/tcp/mod.rs @@ -3,6 +3,6 @@ mod network; mod stream; pub use connector::{PollEvent, SendBehavior, TcpConnector}; -pub use network::{TcpEvent, TcpGroup, TcpGroupConfig, TcpNetwork}; +pub use network::{Framing, TcpEvent, TcpGroup, TcpGroupConfig, TcpNetwork}; pub(crate) use stream::set_socket_buf_size; pub use stream::{ConnState, TcpStream, TcpTelemetry}; diff --git a/crates/flux-network/src/tcp/network.rs b/crates/flux-network/src/tcp/network.rs index fc80aa1..7552aa7 100644 --- a/crates/flux-network/src/tcp/network.rs +++ b/crates/flux-network/src/tcp/network.rs @@ -31,6 +31,17 @@ const BACKLOG_WARNING_INTERVAL_SECS: u64 = 10; #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] pub struct TcpGroup(usize); +/// Selects how a TCP group encodes messages on the wire. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub enum Framing { + /// Messages carry Flux's length and send-timestamp header. + #[default] + LengthPrefixed, + /// Bytes pass through untouched. Received chunks do not preserve message + /// boundaries, and their event timestamp is the local receive time. + Raw, +} + /// Configuration shared by every listener and connection in a [`TcpGroup`]. #[derive(Clone)] pub struct TcpGroupConfig { @@ -55,8 +66,11 @@ pub struct TcpGroupConfig { /// Disconnect the peer before its queued bytes would exceed this limit. /// `None` allows the queue to grow without a hard limit. pub max_backlog_bytes: Option, - /// Largest accepted or emitted frame payload. + /// Largest accepted or emitted frame payload. For [`Framing::Raw`], this + /// caps a single send and bounds each received read chunk. pub max_frame_size: usize, + /// Wire encoding used by this group. + pub framing: Framing, /// Per-connection latency and allocation telemetry. pub telemetry: TcpTelemetry, } @@ -74,6 +88,7 @@ impl Default for TcpGroupConfig { backlog_warn_bytes: Some(DEFAULT_BACKLOG_WARN_BYTES), max_backlog_bytes: None, max_frame_size: DEFAULT_MAX_FRAME_SIZE, + framing: Framing::LengthPrefixed, telemetry: TcpTelemetry::Disabled, } } @@ -93,7 +108,9 @@ pub enum TcpEvent<'a> { Accepted { group: TcpGroup, token: Token, peer_addr: SocketAddr }, /// A persistent outbound endpoint established a connection. Connected { group: TcpGroup, token: Token, peer_addr: SocketAddr }, - /// A complete framed message was received. + /// A complete length-prefixed message or a raw read chunk was received. + /// For raw groups, chunks do not preserve message boundaries and `send_ts` + /// is the local receive time. Message { group: TcpGroup, token: Token, payload: &'a [u8], send_ts: Nanos }, /// An established connection was closed. Disconnected { group: TcpGroup, token: Token, peer_addr: SocketAddr }, @@ -129,12 +146,13 @@ struct Connection { peer_addr: SocketAddr, kind: ConnectionKind, state: ConnectionState, + close_when_drained: bool, timers: Option, } #[derive(Clone, Copy)] struct NetworkTimers { - latency: Timer, + latency: Option, alloc: Timer, } @@ -144,11 +162,13 @@ impl NetworkTimers { group_name: &str, token: Token, peer_addr: SocketAddr, + framing: Framing, ) -> Option { let TcpTelemetry::Enabled { app_name } = telemetry else { return None }; let label = format!("{group_name}-{}-{peer_addr}", token.0); Some(Self { - latency: Timer::new(app_name, format!("tcp_latency_{label}")), + latency: (framing == Framing::LengthPrefixed) + .then(|| Timer::new(app_name, format!("tcp_latency_{label}"))), alloc: Timer::new(app_name, format!("tcp_alloc_{label}")), }) } @@ -213,13 +233,15 @@ impl NetworkState { assert!(group.0 < self.groups.len(), "unknown TCP group"); let token = self.next_token(); let config = self.config(group); - let timers = NetworkTimers::new(config.telemetry, config.name, token, peer_addr); + let timers = + NetworkTimers::new(config.telemetry, config.name, token, peer_addr, config.framing); self.connections.push(Connection { token, group, peer_addr, kind: ConnectionKind::Outbound, state: ConnectionState::Disconnected, + close_when_drained: false, timers, }); self.start_connect(self.connections.len() - 1); @@ -346,10 +368,18 @@ impl NetworkState { let mut stream = FramedStream::new(socket, token, peer_addr, config.max_frame_size); if let Some(message) = config.on_connect_msg.as_deref() { - let mut header = [0; FRAME_HEADER_SIZE]; - write_frame_header(&mut header, message.len(), Nanos::now()); - if stream.write_frame(self.poll.registry(), &header, message, config, &mut timers) == - StreamState::Disconnected + let header = (config.framing == Framing::LengthPrefixed).then(|| { + let mut header = [0; FRAME_HEADER_SIZE]; + write_frame_header(&mut header, message.len(), Nanos::now()); + header + }); + if stream.write_frame( + self.poll.registry(), + header.as_ref(), + message, + config, + &mut timers, + ) == StreamState::Disconnected { stream.close(self.poll.registry()); self.connections[index].timers = timers; @@ -408,15 +438,23 @@ impl NetworkState { continue; } - let mut timers = - NetworkTimers::new(config.telemetry, config.name, token, peer_addr); + let mut timers = NetworkTimers::new( + config.telemetry, + config.name, + token, + peer_addr, + config.framing, + ); let mut stream = FramedStream::new(socket, token, peer_addr, config.max_frame_size); if let Some(message) = config.on_connect_msg.as_deref() { - let mut header = [0; FRAME_HEADER_SIZE]; - write_frame_header(&mut header, message.len(), Nanos::now()); + let header = (config.framing == Framing::LengthPrefixed).then(|| { + let mut header = [0; FRAME_HEADER_SIZE]; + write_frame_header(&mut header, message.len(), Nanos::now()); + header + }); if stream.write_frame( self.poll.registry(), - &header, + header.as_ref(), message, config, &mut timers, @@ -435,6 +473,7 @@ impl NetworkState { peer_addr, kind: ConnectionKind::Accepted, state: ConnectionState::Connected(stream), + close_when_drained: false, timers, }); info!(group = group_name, %peer_addr, "tcp connection accepted"); @@ -469,20 +508,25 @@ impl NetworkState { let group = self.connections[index].group; let peer_addr = self.connections[index].peer_addr; let config = &self.groups[group.0].config; - let connection = &mut self.connections[index]; - let ConnectionState::Connected(stream) = &mut connection.state else { unreachable!() }; - let state = stream.poll_with( - self.poll.registry(), - event, - config, - &mut connection.timers, - &mut |payload, send_ts| { - handler(TcpEvent::Message { group, token, payload, send_ts }); - }, - ); + let (state, queue_empty) = { + let connection = &mut self.connections[index]; + let ConnectionState::Connected(stream) = &mut connection.state else { unreachable!() }; + let state = stream.poll_with( + self.poll.registry(), + event, + config, + &mut connection.timers, + &mut |payload, send_ts| { + handler(TcpEvent::Message { group, token, payload, send_ts }); + }, + ); + (state, stream.send_queue.is_empty()) + }; if state == StreamState::Disconnected { handler(TcpEvent::Disconnected { group, token, peer_addr }); self.disconnect_index(index, false); + } else if self.connections[index].close_when_drained && queue_empty { + self.disconnect_index(index, true); } } @@ -510,6 +554,7 @@ impl NetworkState { peer_addr: self.connections[index].peer_addr, }; let kind = self.connections[index].kind; + self.connections[index].close_when_drained = false; let was_connected = self.close_connection_socket(index); if kind == ConnectionKind::Accepted { self.connections.swap_remove(index); @@ -543,7 +588,8 @@ impl NetworkState { return false; } if self.send_payload.len() > config.max_frame_size || - u32::try_from(self.send_payload.len()).is_err() + (config.framing == Framing::LengthPrefixed && + u32::try_from(self.send_payload.len()).is_err()) { error!( group = config.name, @@ -554,7 +600,9 @@ impl NetworkState { self.send_payload.clear(); return false; } - write_frame_header(&mut self.send_header, self.send_payload.len(), Nanos::now()); + if config.framing == Framing::LengthPrefixed { + write_frame_header(&mut self.send_header, self.send_payload.len(), Nanos::now()); + } true } @@ -563,7 +611,9 @@ impl NetworkState { F: FnOnce(&mut Vec), { let Some(index) = self.connections.iter().position(|connection| { - connection.token == token && matches!(connection.state, ConnectionState::Connected(_)) + connection.token == token && + !connection.close_when_drained && + matches!(connection.state, ConnectionState::Connected(_)) }) else { return false; }; @@ -574,9 +624,10 @@ impl NetworkState { let config = &self.groups[group.0].config; let connection = &mut self.connections[index]; let ConnectionState::Connected(stream) = &mut connection.state else { unreachable!() }; + let header = (config.framing == Framing::LengthPrefixed).then_some(&self.send_header); let state = stream.write_frame( self.poll.registry(), - &self.send_header, + header, &self.send_payload, config, &mut connection.timers, @@ -596,7 +647,9 @@ impl NetworkState { return 0; } if !self.connections.iter().any(|connection| { - connection.group == group && matches!(connection.state, ConnectionState::Connected(_)) + connection.group == group && + !connection.close_when_drained && + matches!(connection.state, ConnectionState::Connected(_)) }) { return 0; } @@ -609,6 +662,7 @@ impl NetworkState { while index != 0 { index -= 1; if self.connections[index].group != group || + self.connections[index].close_when_drained || !matches!(self.connections[index].state, ConnectionState::Connected(_)) { continue; @@ -620,9 +674,11 @@ impl NetworkState { let ConnectionState::Connected(stream) = &mut connection.state else { unreachable!() }; + let header = + (config.framing == Framing::LengthPrefixed).then_some(&self.send_header); stream.write_frame( self.poll.registry(), - &self.send_header, + header, &self.send_payload, config, &mut connection.timers, @@ -647,6 +703,21 @@ impl NetworkState { true } + fn disconnect_when_drained(&mut self, token: Token) -> bool { + let Some(index) = self.connections.iter().position(|connection| connection.token == token) + else { + return false; + }; + let ConnectionState::Connected(stream) = &self.connections[index].state else { + return false; + }; + if stream.send_queue.is_empty() { + return self.disconnect(token); + } + self.connections[index].close_when_drained = true; + true + } + fn remove(&mut self, token: Token) -> bool { let Some(index) = self.connections.iter().position(|connection| connection.token == token) else { @@ -681,10 +752,12 @@ impl TcpNetwork { #[must_use = "the group handle identifies listeners and outbound endpoints"] pub fn add_group(&mut self, config: TcpGroupConfig) -> TcpGroup { assert!(config.max_frame_size > 0, "max_frame_size must be nonzero"); - assert!( - u32::try_from(config.max_frame_size).is_ok(), - "max_frame_size exceeds the TCP wire length field" - ); + if config.framing == Framing::LengthPrefixed { + assert!( + u32::try_from(config.max_frame_size).is_ok(), + "max_frame_size exceeds the TCP wire length field" + ); + } if let Some(message) = &config.on_connect_msg { assert!(!message.is_empty(), "on_connect_msg must be nonempty"); assert!( @@ -734,8 +807,10 @@ impl TcpNetwork { self.state.drain_pending_disconnects(&mut handler); } - /// Serializes and sends one frame to a connected token. The closure is not - /// called when the token is unknown or currently disconnected. + /// Serializes and sends one payload to a connected token. Length-prefixed + /// groups add a frame header; raw groups send the payload unchanged. The + /// closure is not called when the token is unknown or currently + /// disconnected. pub fn send_with(&mut self, token: Token, serialise: F) -> bool where F: FnOnce(&mut Vec), @@ -743,8 +818,9 @@ impl TcpNetwork { self.state.send_with(token, serialise) } - /// Serializes one frame and sends it to every connected member of `group`. - /// Returns the number of recipients attempted. + /// Serializes one payload and sends it to every connected member of + /// `group`. Length-prefixed groups add a frame header; raw groups send + /// the payload unchanged. Returns the number of recipients attempted. pub fn broadcast_with(&mut self, group: TcpGroup, serialise: F) -> usize where F: FnOnce(&mut Vec), @@ -759,6 +835,14 @@ impl TcpNetwork { self.state.disconnect(token) } + /// Closes a connected socket after its queued bytes have been written. + /// Returns `false` for unknown or disconnected tokens; sends to a draining + /// token are rejected. A peer that never drains is bounded only by + /// `TCP_USER_TIMEOUT`. + pub fn disconnect_when_drained(&mut self, token: Token) -> bool { + self.state.disconnect_when_drained(token) + } + /// Permanently removes a connection or outbound endpoint. Returns whether /// the token was found. pub fn remove(&mut self, token: Token) -> bool { @@ -836,7 +920,23 @@ impl ByteQueue { flux_utils::safe_assert!(written < frame_len); return false; } - let additional = frame_len - written; + if written < FRAME_HEADER_SIZE { + self.append_remainder(&header[written..], payload) + } else { + self.append_remainder(&[], &payload[written - FRAME_HEADER_SIZE..]) + } + } + + fn append_raw_remainder(&mut self, payload: &[u8], written: usize) -> bool { + if written >= payload.len() { + flux_utils::safe_assert!(written < payload.len()); + return false; + } + self.append_remainder(&[], &payload[written..]) + } + + fn append_remainder(&mut self, prefix: &[u8], payload: &[u8]) -> bool { + let additional = prefix.len() + payload.len(); let old_capacity = self.bytes.capacity(); if self.head != 0 && self.bytes.capacity() - self.bytes.len() < additional { @@ -846,12 +946,8 @@ impl ByteQueue { self.head = 0; } self.bytes.reserve(additional); - if written < FRAME_HEADER_SIZE { - self.bytes.extend_from_slice(&header[written..]); - self.bytes.extend_from_slice(payload); - } else { - self.bytes.extend_from_slice(&payload[written - FRAME_HEADER_SIZE..]); - } + self.bytes.extend_from_slice(prefix); + self.bytes.extend_from_slice(payload); self.queued_since.get_or_insert_with(Instant::now); self.bytes.capacity() != old_capacity } @@ -921,16 +1017,32 @@ impl FramedStream { F: for<'a> FnMut(&'a [u8], Nanos), { if event.is_readable() { - loop { - match self.read_frame(config.max_frame_size) { - ReadOutcome::Message { payload, send_ts } => { - if let Some(timers) = timers { - timers.latency.emit_latency_from_nanos(send_ts, Nanos::now()); + if config.framing == Framing::Raw { + loop { + match self.socket.read(&mut self.rx_buffer) { + Ok(0) => return StreamState::Disconnected, + Ok(read) => on_message(&self.rx_buffer[..read], Nanos::now()), + Err(err) if err.kind() == io::ErrorKind::WouldBlock => break, + Err(err) => { + debug!(?err, %self.peer_addr, "tcp raw read failed"); + return StreamState::Disconnected; + } + } + } + } else { + loop { + match self.read_frame(config.max_frame_size) { + ReadOutcome::Message { payload, send_ts } => { + if let Some(timers) = timers { + if let Some(latency) = &mut timers.latency { + latency.emit_latency_from_nanos(send_ts, Nanos::now()); + } + } + on_message(payload, send_ts); } - on_message(payload, send_ts); + ReadOutcome::WouldBlock => break, + ReadOutcome::Disconnected => return StreamState::Disconnected, } - ReadOutcome::WouldBlock => break, - ReadOutcome::Disconnected => return StreamState::Disconnected, } } } @@ -1021,7 +1133,7 @@ impl FramedStream { fn write_frame( &mut self, registry: &Registry, - header: &[u8; FRAME_HEADER_SIZE], + header: Option<&[u8; FRAME_HEADER_SIZE]>, payload: &[u8], config: &TcpGroupConfig, timers: &mut Option, @@ -1035,10 +1147,15 @@ impl FramedStream { } } - match self.socket.write_vectored(&[IoSlice::new(header.as_slice()), IoSlice::new(payload)]) - { + let result = if let Some(header) = header { + self.socket.write_vectored(&[IoSlice::new(header.as_slice()), IoSlice::new(payload)]) + } else { + self.socket.write(payload) + }; + let total = header.map_or(payload.len(), |_| FRAME_HEADER_SIZE + payload.len()); + match result { Ok(0) => StreamState::Disconnected, - Ok(written) if written == FRAME_HEADER_SIZE + payload.len() => StreamState::Alive, + Ok(written) if written == total => StreamState::Alive, Ok(written) => { self.enqueue_remainder(registry, header, payload, written, config, timers) } @@ -1055,18 +1172,18 @@ impl FramedStream { fn enqueue_remainder( &mut self, registry: &Registry, - header: &[u8; FRAME_HEADER_SIZE], + header: Option<&[u8; FRAME_HEADER_SIZE]>, payload: &[u8], written: usize, config: &TcpGroupConfig, timers: &mut Option, ) -> StreamState { - let frame_len = FRAME_HEADER_SIZE + payload.len(); - if written >= frame_len { - flux_utils::safe_assert!(written < frame_len); + let total = header.map_or(payload.len(), |_| FRAME_HEADER_SIZE + payload.len()); + if written >= total { + flux_utils::safe_assert!(written < total); return StreamState::Disconnected; } - let additional = frame_len - written; + let additional = total - written; if let Some(max) = config.max_backlog_bytes && self.send_queue.would_exceed(additional, max) { @@ -1083,7 +1200,11 @@ impl FramedStream { } let started = Nanos::now(); - let allocated = self.send_queue.append_frame_remainder(header, payload, written); + let allocated = if let Some(header) = header { + self.send_queue.append_frame_remainder(header, payload, written) + } else { + self.send_queue.append_raw_remainder(payload, written) + }; if allocated && let Some(timers) = timers { timers.alloc.emit_latency_from_nanos(started, Nanos::now()); } @@ -1166,6 +1287,17 @@ mod tests { } } + #[test] + fn byte_queue_preserves_raw_unwritten_suffix() { + let payload = [2; 16]; + + for written in [0, 3, 7] { + let mut queue = ByteQueue::default(); + queue.append_raw_remainder(&payload, written); + assert_eq!(queue.remaining(), &payload[written..]); + } + } + #[test] fn byte_queue_compacts_consumed_prefix_before_growing() { let first_header = [1; FRAME_HEADER_SIZE]; @@ -1231,7 +1363,7 @@ mod tests { assert_eq!( stream.write_frame( Poll::new().unwrap().registry(), - &header, + Some(&header), &payload, &config, &mut None diff --git a/crates/flux-network/tests/http.rs b/crates/flux-network/tests/http.rs new file mode 100644 index 0000000..06039a9 --- /dev/null +++ b/crates/flux-network/tests/http.rs @@ -0,0 +1,872 @@ +use std::{ + io::{self, Read, Write}, + net::{Ipv4Addr, SocketAddr}, + thread, + time::{Duration, Instant}, +}; + +use flux_network::http::{HttpEvent, HttpNetwork}; + +const TIMEOUT: Duration = Duration::from_secs(10); + +fn unused_addr() -> SocketAddr { + let listener = std::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).unwrap(); + let addr = listener.local_addr().unwrap(); + drop(listener); + addr +} + +fn read_available(stream: &mut std::net::TcpStream, out: &mut Vec) -> bool { + let mut buf = [0; 8192]; + match stream.read(&mut buf) { + Ok(0) => true, + Ok(n) => { + out.extend_from_slice(&buf[..n]); + false + } + Err(e) if e.kind() == io::ErrorKind::WouldBlock => false, + // A server that hard-closes after responding can RST bytes still in + // flight from the client; everything sent before the close has + // already been received, so treat it as EOF. + Err(e) if e.kind() == io::ErrorKind::ConnectionReset => true, + Err(e) => panic!("read failed: {e}"), + } +} + +fn response_len(bytes: &[u8]) -> Option { + let head = bytes.windows(4).position(|b| b == b"\r\n\r\n")? + 4; + let text = std::str::from_utf8(&bytes[..head]).unwrap(); + let length = text + .lines() + .find_map(|line| line.strip_prefix("Content-Length: ")) + .unwrap() + .parse::() + .unwrap(); + (bytes.len() >= head + length).then_some(head + length) +} + +fn server() -> (HttpNetwork, SocketAddr) { + let addr = unused_addr(); + let mut server = HttpNetwork::default(); + server.listen(addr).unwrap(); + (server, addr) +} + +#[test] +fn get_keepalive_two_requests() { + let (mut server, addr) = server(); + let mut client = std::net::TcpStream::connect(addr).unwrap(); + client.set_nonblocking(true).unwrap(); + let deadline = Instant::now() + TIMEOUT; + let mut first = Vec::new(); + client.write_all(b"GET /one HTTP/1.1\r\nHost: x\r\n\r\n").unwrap(); + while Instant::now() < deadline && response_len(&first).is_none() { + let mut replies = Vec::new(); + server.poll_with(|e| { + if let HttpEvent::Request { token, request } = e { + replies.push(( + token, + if request.path == "/one" { b"one".to_vec() } else { b"two".to_vec() }, + )); + } + }); + for (token, body) in replies { + assert!(server.respond(token, 200, &[], &body)); + } + read_available(&mut client, &mut first); + thread::sleep(Duration::from_millis(1)); + } + assert!(response_len(&first).is_some()); + client.write_all(b"GET /two HTTP/1.1\r\nHost: x\r\n\r\n").unwrap(); + let mut second = Vec::new(); + while Instant::now() < deadline && response_len(&second).is_none() { + let mut replies = Vec::new(); + server.poll_with(|e| { + if let HttpEvent::Request { token, request } = e { + replies.push(( + token, + if request.path == "/one" { b"one".to_vec() } else { b"two".to_vec() }, + )); + } + }); + for (token, body) in replies { + assert!(server.respond(token, 200, &[], &body)); + } + read_available(&mut client, &mut second); + thread::sleep(Duration::from_millis(1)); + } + assert!(first.ends_with(b"one")); + assert!(second.ends_with(b"two")); +} + +#[test] +fn post_echo_body() { + let (mut server, addr) = server(); + let body = vec![42; 4096]; + let mut client = std::net::TcpStream::connect(addr).unwrap(); + client.set_nonblocking(true).unwrap(); + client + .write_all( + format!("POST / HTTP/1.1\r\nHost: x\r\nContent-Length: {}\r\n\r\n", body.len()) + .as_bytes(), + ) + .unwrap(); + client.write_all(&body).unwrap(); + let mut received = Vec::new(); + let deadline = Instant::now() + TIMEOUT; + while Instant::now() < deadline && response_len(&received).is_none() { + let mut replies = Vec::new(); + server.poll_with(|e| { + if let HttpEvent::Request { token, request } = e { + replies.push((token, request.body.to_vec())); + } + }); + for (token, body) in replies { + server.respond(token, 200, &[], &body); + } + read_available(&mut client, &mut received); + thread::sleep(Duration::from_millis(1)); + } + assert!(received.ends_with(&body)); +} + +#[test] +fn post_binary_body_lone_lf() { + let (mut server, addr) = server(); + let body: Vec = (0..=255u8).cycle().take(4096).collect(); + assert!(body.windows(2).any(|w| w[0] != b'\r' && w[1] == b'\n')); + let mut client = std::net::TcpStream::connect(addr).unwrap(); + client.set_nonblocking(true).unwrap(); + client + .write_all( + format!("POST / HTTP/1.1\r\nHost: x\r\nContent-Length: {}\r\n\r\n", body.len()) + .as_bytes(), + ) + .unwrap(); + client.write_all(&body).unwrap(); + let mut received = Vec::new(); + let deadline = Instant::now() + TIMEOUT; + while Instant::now() < deadline && response_len(&received).is_none() { + let mut replies = Vec::new(); + server.poll_with(|e| { + if let HttpEvent::Request { token, request } = e { + replies.push((token, request.body.to_vec())); + } + }); + for (token, body) in replies { + assert!(server.respond(token, 200, &[], &body)); + } + read_available(&mut client, &mut received); + thread::sleep(Duration::from_millis(1)); + } + assert!(received.ends_with(&body)); + + // The connection must stay usable for a text request after a binary body. + client.write_all(b"GET /after HTTP/1.1\r\nHost: x\r\n\r\n").unwrap(); + let mut second = Vec::new(); + while Instant::now() < deadline && response_len(&second).is_none() { + let mut replies = Vec::new(); + server.poll_with(|e| { + if let HttpEvent::Request { token, request } = e { + assert_eq!(request.path, "/after"); + replies.push(token); + } + }); + for token in replies { + assert!(server.respond(token, 200, &[], b"after")); + } + read_available(&mut client, &mut second); + thread::sleep(Duration::from_millis(1)); + } + assert!(second.ends_with(b"after")); +} + +#[test] +fn connection_close_large_body() { + let addr = unused_addr(); + let mut server = HttpNetwork::default().with_socket_buf_size(1024); + server.listen(addr).unwrap(); + let body = vec![7; 256 * 1024]; + let mut client = std::net::TcpStream::connect(addr).unwrap(); + client.set_nonblocking(true).unwrap(); + client.write_all(b"GET / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n").unwrap(); + let mut received = Vec::new(); + let deadline = Instant::now() + TIMEOUT; + while Instant::now() < deadline && !read_available(&mut client, &mut received) { + let mut tokens = Vec::new(); + server.poll_with(|e| { + if let HttpEvent::Request { token, .. } = e { + tokens.push(token); + } + }); + for token in tokens { + server.respond(token, 200, &[], &body); + } + thread::sleep(Duration::from_millis(1)); + } + assert!(received.ends_with(&body)); +} + +#[test] +fn limits_and_errors() { + for (request, status) in [ + (b"GET / HTTP/1.1\r\nX: 1234567890123456789012345678901234567890123456789012345678901234\r\n\r\n".as_slice(), 431), + (b"POST / HTTP/1.1\r\nContent-Length: 9\r\n\r\n".as_slice(), 413), + (b"nope\r\n\r\n".as_slice(), 400), + (b"POST / HTTP/1.1\r\nTransfer-Encoding: chunked\r\n\r\n".as_slice(), 501), + ] { + let addr = unused_addr(); + let mut server = HttpNetwork::default().with_max_head_bytes(64).with_max_body_bytes(8); + server.listen(addr).unwrap(); + let mut client = std::net::TcpStream::connect(addr).unwrap(); + client.set_nonblocking(true).unwrap(); + client.write_all(request).unwrap(); + let mut received = Vec::new(); + let deadline = Instant::now() + TIMEOUT; + while Instant::now() < deadline && !read_available(&mut client, &mut received) { + server.poll_with(|_| {}); + thread::sleep(Duration::from_millis(1)); + } + assert!(std::str::from_utf8(&received).unwrap().starts_with(&format!("HTTP/1.1 {status}"))); + } +} + +#[test] +fn caller_connection_close_header_sent() { + let (mut server, addr) = server(); + let mut client = std::net::TcpStream::connect(addr).unwrap(); + client.set_nonblocking(true).unwrap(); + client.write_all(b"GET / HTTP/1.1\r\nHost: x\r\n\r\n").unwrap(); + let mut received = Vec::new(); + let deadline = Instant::now() + TIMEOUT; + while Instant::now() < deadline && !read_available(&mut client, &mut received) { + let mut tokens = Vec::new(); + server.poll_with(|e| { + if let HttpEvent::Request { token, .. } = e { + tokens.push(token); + } + }); + for token in tokens { + // The caller-supplied Connection header must still result in + // exactly one canonical Connection: close on the wire. + assert!(server.respond(token, 200, &[("Connection", "close")], b"ok")); + } + thread::sleep(Duration::from_millis(1)); + } + let text = String::from_utf8(received).unwrap(); + assert_eq!(text.matches("Connection:").count(), 1, "{text}"); + assert!(text.contains("Connection: close\r\n"), "{text}"); + assert!(text.ends_with("ok")); +} + +#[test] +fn pipelined_requests() { + let (mut server, addr) = server(); + let mut client = std::net::TcpStream::connect(addr).unwrap(); + client.set_nonblocking(true).unwrap(); + client + .write_all(b"GET /one HTTP/1.1\r\nHost: x\r\n\r\nGET /two HTTP/1.1\r\nHost: x\r\n\r\n") + .unwrap(); + let mut paths = Vec::new(); + let mut received = Vec::new(); + let deadline = Instant::now() + TIMEOUT; + while Instant::now() < deadline && paths.len() < 2 { + let mut replies = Vec::new(); + server.poll_with(|e| { + if let HttpEvent::Request { token, request } = e { + paths.push(request.path.to_owned()); + replies.push(token); + } + }); + for token in replies { + server.respond(token, 200, &[], paths.last().unwrap().as_bytes()); + } + read_available(&mut client, &mut received); + thread::sleep(Duration::from_millis(1)); + } + assert_eq!(paths, ["/one", "/two"]); + assert!(received.ends_with(b"/two")); +} + +#[test] +fn client_server_roundtrip() { + let (mut server, addr) = server(); + let mut client = HttpNetwork::default(); + let token = client.connect(addr); + let mut sent = false; + let mut bodies = Vec::new(); + let deadline = Instant::now() + TIMEOUT; + while Instant::now() < deadline && bodies.len() < 2 { + let mut replies = Vec::new(); + server.poll_with(|e| { + if let HttpEvent::Request { token, request } = e { + replies.push((token, request.body.to_vec())); + } + }); + for (token, body) in replies { + server.respond(token, 200, &[("X-Reply", "yes")], &body); + } + let mut send_second = false; + let mut connected = false; + client.poll_with(|e| match e { + HttpEvent::Connected { .. } => connected = true, + HttpEvent::Response { response, .. } => { + bodies.push(response.body.to_vec()); + send_second = true; + } + _ => {} + }); + if connected && !sent { + assert!(client.request(token, "POST", "/", &[("X-Test", "yes")], b"hello")); + assert!(!client.request(token, "GET", "/", &[], b"")); + sent = true; + } + if send_second && bodies.len() == 1 { + assert!(client.request(token, "GET", "/", &[], b"")); + } + thread::sleep(Duration::from_millis(1)); + } + assert_eq!(bodies, [b"hello".to_vec(), Vec::new()]); +} + +#[test] +fn client_chunked_response() { + let listener = std::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).unwrap(); + let addr = listener.local_addr().unwrap(); + let peer = thread::spawn(move || { + let (mut s, _) = listener.accept().unwrap(); + let mut b = [0; 1024]; + let _ = s.read(&mut b); + s.write_all(b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n3 \r\nhey\r\n2;ext=x\r\n!!\r\n0\r\nX: y\r\n\r\n").unwrap(); + }); + let mut client = HttpNetwork::default(); + let token = client.connect(addr); + let mut sent = false; + let mut body = None; + let deadline = Instant::now() + TIMEOUT; + while Instant::now() < deadline && body.is_none() { + let mut connected = false; + client.poll_with(|e| match e { + HttpEvent::Connected { .. } => connected = true, + HttpEvent::Response { response, .. } => body = Some(response.body.to_vec()), + _ => {} + }); + if connected && !sent { + assert!(client.request(token, "GET", "/", &[], b"")); + sent = true; + } + thread::sleep(Duration::from_millis(1)); + } + peer.join().unwrap(); + assert_eq!(body.unwrap(), b"hey!!"); +} + +#[test] +fn client_head_response_ignores_advisory_content_length() { + let listener = std::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).unwrap(); + let addr = listener.local_addr().unwrap(); + let peer = thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + let mut request = [0; 1024]; + let _ = stream.read(&mut request); + stream.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 999999\r\n\r\n").unwrap(); + }); + let mut client = HttpNetwork::default().with_max_body_bytes(8); + let token = client.connect(addr); + let deadline = Instant::now() + TIMEOUT; + let mut connected = false; + let mut response = None; + while Instant::now() < deadline && response.is_none() { + client.poll_with(|event| match event { + HttpEvent::Connected { token: event_token } if event_token == token => connected = true, + HttpEvent::Response { token: event_token, response: event_response } + if event_token == token => + { + response = Some((event_response.status, event_response.body.len())); + } + _ => {} + }); + if connected { + assert!(client.request(token, "HEAD", "/", &[], &[])); + connected = false; + } + thread::sleep(Duration::from_millis(1)); + } + peer.join().unwrap(); + assert_eq!(response, Some((200, 0))); +} + +#[test] +fn client_binary_bodies() { + let listener = std::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).unwrap(); + let addr = listener.local_addr().unwrap(); + let chunked_body = b"\n\x00\n\xff"; + let plain_body: Vec = (0..=255u8).collect(); + let plain = plain_body.clone(); + let peer = thread::spawn(move || { + let (mut s, _) = listener.accept().unwrap(); + let mut b = [0; 1024]; + let _ = s.read(&mut b); + s.write_all( + b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n4\r\n\n\x00\n\xff\r\n0\r\n\r\n", + ) + .unwrap(); + let _ = s.read(&mut b); + s.write_all( + format!("HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n", plain.len()).as_bytes(), + ) + .unwrap(); + s.write_all(&plain).unwrap(); + }); + let mut client = HttpNetwork::default(); + let token = client.connect(addr); + let mut sent = false; + let mut bodies = Vec::new(); + let deadline = Instant::now() + TIMEOUT; + while Instant::now() < deadline && bodies.len() < 2 { + let mut connected = false; + let mut respond_again = false; + client.poll_with(|e| match e { + HttpEvent::Connected { .. } => connected = true, + HttpEvent::Response { response, .. } => { + bodies.push(response.body.to_vec()); + respond_again = true; + } + _ => {} + }); + if connected && !sent { + assert!(client.request(token, "GET", "/", &[], b"")); + sent = true; + } + if respond_again && bodies.len() == 1 { + assert!(client.request(token, "GET", "/", &[], b"")); + } + thread::sleep(Duration::from_millis(1)); + } + peer.join().unwrap(); + assert_eq!(bodies, [chunked_body.to_vec(), plain_body]); +} + +#[test] +fn client_reconnect_after_close() { + let (mut server, addr) = server(); + let mut client = HttpNetwork::default(); + let token = client.connect(addr); + let mut connected = 0; + let mut disconnected = 0; + let mut responses = 0; + let deadline = Instant::now() + TIMEOUT; + while Instant::now() < deadline && responses < 2 { + let mut replies = Vec::new(); + server.poll_with(|e| { + if let HttpEvent::Request { token, .. } = e { + replies.push(token); + } + }); + for token in replies { + server.respond(token, 200, &[("Connection", "close")], b"ok"); + } + let mut request_again = false; + client.poll_with(|e| match e { + HttpEvent::Connected { .. } => { + connected += 1; + request_again = true; + } + HttpEvent::Disconnected { .. } => disconnected += 1, + HttpEvent::Response { .. } => responses += 1, + _ => {} + }); + if request_again { + assert!(client.request(token, "GET", "/", &[("Connection", "close")], b"")); + } + thread::sleep(Duration::from_millis(1)); + } + assert!(connected >= 2 && disconnected >= 1 && responses == 2); +} + +#[test] +fn smuggling_rejected() { + for (request, status, accepted) in [ + ( + b"POST / HTTP/1.1\r\nContent-Length: 1\r\nContent-Length: 2\r\n\r\nx".as_slice(), + 400, + false, + ), + ( + b"POST / HTTP/1.1\r\nContent-Length: 1\r\nContent-Length: 1\r\n\r\nx".as_slice(), + 200, + true, + ), + ( + b"POST / HTTP/1.1\r\nContent-Length: 1\r\nTransfer-Encoding: chunked\r\n\r\n0\r\n\r\n" + .as_slice(), + 501, + false, + ), + ] { + let (mut server, addr) = server(); + let mut client = std::net::TcpStream::connect(addr).unwrap(); + client.set_nonblocking(true).unwrap(); + client.write_all(request).unwrap(); + let deadline = Instant::now() + TIMEOUT; + let mut received = Vec::new(); + let mut requests = 0; + let mut closed = false; + while Instant::now() < deadline && + if accepted { response_len(&received).is_none() } else { !closed } + { + let mut replies = Vec::new(); + server.poll_with(|event| { + if let HttpEvent::Request { token, .. } = event { + requests += 1; + replies.push(token); + } + }); + for token in replies { + assert!(server.respond(token, 200, &[], b"ok")); + } + closed = read_available(&mut client, &mut received); + thread::sleep(Duration::from_millis(1)); + } + if accepted { + assert!(response_len(&received).is_some()); + } else { + assert!(closed); + } + assert!(std::str::from_utf8(&received).unwrap().starts_with(&format!("HTTP/1.1 {status}"))); + assert_eq!(requests > 0, accepted); + } +} + +#[test] +fn bare_lf_head_rejected() { + let (mut server, addr) = server(); + let mut client = std::net::TcpStream::connect(addr).unwrap(); + client.set_nonblocking(true).unwrap(); + client.write_all(b"GET / HTTP/1.1\nHost: x\n\n").unwrap(); + let deadline = Instant::now() + TIMEOUT; + let mut received = Vec::new(); + let mut closed = false; + while Instant::now() < deadline && !closed { + server.poll_with(|_| {}); + closed = read_available(&mut client, &mut received); + thread::sleep(Duration::from_millis(1)); + } + assert!(closed); + assert!(received.starts_with(b"HTTP/1.1 400")); +} + +fn assert_chunked_response_disconnect(response: &'static [u8], max_headers: usize) { + let listener = std::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).unwrap(); + let addr = listener.local_addr().unwrap(); + let peer = thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + let mut request = [0; 1024]; + let _ = stream.read(&mut request); + stream.write_all(response).unwrap(); + }); + let mut client = HttpNetwork::default().with_max_headers(max_headers); + let token = client.connect(addr); + let deadline = Instant::now() + TIMEOUT; + let mut sent = false; + let mut disconnected = 0; + while Instant::now() < deadline && disconnected == 0 { + let mut connected = false; + client.poll_with(|event| match event { + HttpEvent::Connected { .. } => connected = true, + HttpEvent::Disconnected { token: event_token } if event_token == token => { + disconnected += 1; + } + _ => {} + }); + if connected && !sent { + assert!(client.request(token, "GET", "/", &[], b"")); + sent = true; + } + thread::sleep(Duration::from_millis(1)); + } + peer.join().unwrap(); + assert_eq!(disconnected, 1); +} + +#[test] +fn client_chunked_malformed_trailers_rejected() { + assert_chunked_response_disconnect( + b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n0\r\ngarbage\r\n\r\n", + 64, + ); + assert_chunked_response_disconnect( + b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n0\r\nOne: 1\r\nTwo: 2\r\n\r\n", + 1, + ); +} + +#[test] +fn client_chunked_overflow() { + for response in [ + b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n10000000000000000\r\n".as_slice(), + b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n11\r\n".as_slice(), + ] { + let listener = std::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).unwrap(); + let addr = listener.local_addr().unwrap(); + let peer = thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + let mut request = [0; 1024]; + let _ = stream.read(&mut request); + stream.write_all(response).unwrap(); + }); + let mut client = HttpNetwork::default().with_max_body_bytes(16); + let token = client.connect(addr); + let deadline = Instant::now() + TIMEOUT; + let mut sent = false; + let mut disconnected = 0; + while Instant::now() < deadline && disconnected == 0 { + let mut connected = false; + client.poll_with(|event| match event { + HttpEvent::Connected { .. } => connected = true, + HttpEvent::Disconnected { token: event_token } if event_token == token => { + disconnected += 1; + } + _ => {} + }); + if connected && !sent { + assert!(client.request(token, "GET", "/", &[], b"")); + sent = true; + } + thread::sleep(Duration::from_millis(1)); + } + peer.join().unwrap(); + assert_eq!(disconnected, 1); + } +} + +#[test] +fn idle_timeout_disconnects() { + let addr = unused_addr(); + let mut server = HttpNetwork::default().with_idle_timeout(Duration::from_millis(200).into()); + server.listen(addr).unwrap(); + let mut client = std::net::TcpStream::connect(addr).unwrap(); + client.set_nonblocking(true).unwrap(); + let deadline = Instant::now() + Duration::from_secs(2); + let mut bytes = Vec::new(); + let mut closed = false; + while Instant::now() < deadline && !closed { + server.poll_with(|_| {}); + closed = read_available(&mut client, &mut bytes); + thread::sleep(Duration::from_millis(1)); + } + assert!(closed); + + let mut client = std::net::TcpStream::connect(addr).unwrap(); + client.set_nonblocking(true).unwrap(); + client.write_all(b"GET / HTTP/1.1\r\nHost: x\r\n\r\n").unwrap(); + let deadline = Instant::now() + TIMEOUT; + let mut response = Vec::new(); + while Instant::now() < deadline && response_len(&response).is_none() { + let mut replies = Vec::new(); + server.poll_with(|event| { + if let HttpEvent::Request { token, .. } = event { + replies.push(token); + } + }); + for token in replies { + assert!(server.respond(token, 200, &[], b"ok")); + } + read_available(&mut client, &mut response); + thread::sleep(Duration::from_millis(1)); + } + assert!(response_len(&response).is_some()); + let deadline = Instant::now() + Duration::from_secs(2); + let mut closed = false; + while Instant::now() < deadline && !closed { + server.poll_with(|_| {}); + closed = read_available(&mut client, &mut response); + thread::sleep(Duration::from_millis(1)); + } + assert!(closed); +} + +#[test] +fn pending_buffer_cap_disconnects() { + let addr = unused_addr(); + let mut server = HttpNetwork::default().with_max_head_bytes(64).with_max_body_bytes(64); + server.listen(addr).unwrap(); + let mut client = std::net::TcpStream::connect(addr).unwrap(); + client.set_nonblocking(true).unwrap(); + client.write_all(b"GET / HTTP/1.1\r\nHost: x\r\n\r\n").unwrap(); + let deadline = Instant::now() + TIMEOUT; + let mut accepted = false; + while Instant::now() < deadline && !accepted { + server.poll_with(|event| accepted |= matches!(event, HttpEvent::Request { .. })); + thread::sleep(Duration::from_millis(1)); + } + assert!(accepted); + client.write_all(&[b'x'; 129]).unwrap(); + let mut received = Vec::new(); + let mut closed = false; + while Instant::now() < deadline && !closed { + server.poll_with(|_| {}); + closed = read_available(&mut client, &mut received); + thread::sleep(Duration::from_millis(1)); + } + assert!(closed); +} + +#[test] +fn pipelined_binary_bodies() { + let (mut server, addr) = server(); + let first = b"\x00\n\xffone"; + let second = b"two\n\x00\xfe"; + let mut request = Vec::new(); + for (path, body) in [("/one", first.as_slice()), ("/two", second.as_slice())] { + write!( + request, + "POST {path} HTTP/1.1\r\nHost: x\r\nContent-Length: {}\r\n\r\n", + body.len() + ) + .unwrap(); + request.extend_from_slice(body); + } + let mut client = std::net::TcpStream::connect(addr).unwrap(); + client.set_nonblocking(true).unwrap(); + client.write_all(&request).unwrap(); + let deadline = Instant::now() + TIMEOUT; + let mut replies = Vec::new(); + let mut received = Vec::new(); + while Instant::now() < deadline && replies.len() < 2 { + let mut pending = Vec::new(); + server.poll_with(|event| { + if let HttpEvent::Request { token, request } = event { + pending.push((token, request.body.to_vec())); + } + }); + for (token, body) in pending { + replies.push(body.clone()); + assert!(server.respond(token, 200, &[], &body)); + } + read_available(&mut client, &mut received); + thread::sleep(Duration::from_millis(1)); + } + assert_eq!(replies, [first.to_vec(), second.to_vec()]); + let first_len = response_len(&received).unwrap(); + let second_len = response_len(&received[first_len..]).unwrap(); + assert_eq!(&received[first_len - first.len()..first_len], first); + assert_eq!(&received[first_len + second_len - second.len()..first_len + second_len], second); +} + +#[test] +fn client_remove_stops_reconnect() { + let (mut server, addr) = server(); + let mut client = HttpNetwork::default(); + let token = client.connect(addr); + let deadline = Instant::now() + TIMEOUT; + let mut connected = false; + while Instant::now() < deadline && !connected { + server.poll_with(|_| {}); + client.poll_with(|event| connected |= matches!(event, HttpEvent::Connected { token: event_token } if event_token == token)); + thread::sleep(Duration::from_millis(1)); + } + assert!(connected); + assert!(client.remove(token)); + assert!(!client.request(token, "GET", "/", &[], b"")); + let deadline = Instant::now() + Duration::from_millis(500); + let mut reconnects = 0; + while Instant::now() < deadline { + server.poll_with(|_| {}); + client.poll_with(|event| { + if matches!(event, HttpEvent::Connected { token: event_token } if event_token == token) + { + reconnects += 1; + } + }); + thread::sleep(Duration::from_millis(1)); + } + assert_eq!(reconnects, 0); +} + +#[test] +fn server_disconnect_kicks() { + let (mut server, addr) = server(); + let mut client = std::net::TcpStream::connect(addr).unwrap(); + client.set_nonblocking(true).unwrap(); + let deadline = Instant::now() + TIMEOUT; + let mut token = None; + while Instant::now() < deadline && token.is_none() { + server.poll_with(|event| { + if let HttpEvent::Accepted { token: connected, .. } = event { + token = Some(connected); + } + }); + thread::sleep(Duration::from_millis(1)); + } + let token = token.expect("server must accept client"); + assert!(server.disconnect(token)); + let mut bytes = Vec::new(); + let mut closed = false; + let mut disconnected = 0; + while Instant::now() < deadline && (!closed || disconnected == 0) { + server.poll_with(|event| { + if matches!(event, HttpEvent::Disconnected { token: event_token } if event_token == token) { + disconnected += 1; + } + }); + closed |= read_available(&mut client, &mut bytes); + thread::sleep(Duration::from_millis(1)); + } + assert!(closed); + assert_eq!(disconnected, 1); +} + +#[test] +fn wrong_role_calls_return_false() { + let addr = unused_addr(); + let mut http = HttpNetwork::default(); + http.listen(addr).unwrap(); + let outbound = http.connect(addr); + let stream = std::net::TcpStream::connect(addr).unwrap(); + stream.set_nonblocking(true).unwrap(); + let deadline = Instant::now() + TIMEOUT; + let mut accepted = None; + while Instant::now() < deadline && accepted.is_none() { + http.poll_with(|event| { + if let HttpEvent::Accepted { token, .. } = event { + accepted = Some(token); + } + }); + thread::sleep(Duration::from_millis(1)); + } + assert!(!http.respond(outbound, 200, &[], b"")); + assert!(!http.request(accepted.unwrap(), "GET", "/", &[], b"")); +} + +#[test] +fn single_instance_serves_itself() { + let addr = unused_addr(); + let mut http = HttpNetwork::default(); + http.listen(addr).unwrap(); + let outbound = http.connect(addr); + let deadline = Instant::now() + TIMEOUT; + let mut sent = false; + let mut body = None; + while Instant::now() < deadline && body.is_none() { + let mut respond = None; + let mut request = false; + http.poll_with(|event| match event { + HttpEvent::Connected { token } if token == outbound => request = true, + HttpEvent::Request { token, .. } => respond = Some(token), + HttpEvent::Response { token, response } if token == outbound => { + body = Some(response.body.to_vec()); + } + _ => {} + }); + if request && !sent { + assert!(http.request(outbound, "GET", "/", &[], b"")); + sent = true; + } + if let Some(token) = respond { + assert!(http.respond(token, 200, &[], b"self")); + } + thread::sleep(Duration::from_millis(1)); + } + assert_eq!(body.as_deref(), Some(b"self".as_slice())); +} diff --git a/crates/flux-network/tests/tcp_network_raw.rs b/crates/flux-network/tests/tcp_network_raw.rs new file mode 100644 index 0000000..776c8a8 --- /dev/null +++ b/crates/flux-network/tests/tcp_network_raw.rs @@ -0,0 +1,307 @@ +use std::{ + io::{self, Read, Write}, + net::{Ipv4Addr, SocketAddr}, + thread, + time::{Duration, Instant}, +}; + +use flux_network::tcp::{Framing, TcpEvent, TcpGroupConfig, TcpNetwork}; + +const TIMEOUT: Duration = Duration::from_secs(5); + +fn unused_addr() -> SocketAddr { + let listener = std::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).unwrap(); + let addr = listener.local_addr().unwrap(); + drop(listener); + addr +} + +fn raw_group(name: &'static str) -> TcpGroupConfig { + TcpGroupConfig { name, framing: Framing::Raw, ..TcpGroupConfig::default() } +} + +#[test] +fn raw_roundtrip() { + let request = b"raw request bytes"; + let addr = unused_addr(); + let mut network = TcpNetwork::default(); + let group = network.add_group(raw_group("raw-server")); + network.listen(group, addr).unwrap(); + + let mut client = std::net::TcpStream::connect(addr).unwrap(); + client.set_nonblocking(true).unwrap(); + let mut received = Vec::new(); + let mut echoed = false; + let deadline = Instant::now() + TIMEOUT; + client.write_all(request).unwrap(); + + while Instant::now() < deadline && received.len() < request.len() { + let mut echo = None; + network.poll_with(|event| { + if let TcpEvent::Message { group: event_group, token, payload, .. } = event { + assert_eq!(event_group, group); + echo = Some((token, payload.to_vec())); + } + }); + if let Some((token, payload)) = echo { + assert!(network.send_with(token, |buf| buf.extend_from_slice(&payload))); + echoed = true; + } + let mut buffer = [0; 128]; + match client.read(&mut buffer) { + Ok(read) => received.extend_from_slice(&buffer[..read]), + Err(err) if err.kind() == io::ErrorKind::WouldBlock => {} + Err(err) => panic!("client read failed: {err}"), + } + thread::sleep(Duration::from_millis(1)); + } + + assert!(echoed); + assert_eq!(received, request); +} + +#[test] +fn http_get_smoke() { + let request = b"GET /health HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n"; + let response = b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok"; + let addr = unused_addr(); + let mut network = TcpNetwork::default(); + let group = network.add_group(raw_group("http")); + network.listen(group, addr).unwrap(); + + let mut client = std::net::TcpStream::connect(addr).unwrap(); + client.set_nonblocking(true).unwrap(); + client.write_all(request).unwrap(); + let mut request_bytes = Vec::new(); + let mut response_token = None; + let mut received = Vec::new(); + let deadline = Instant::now() + TIMEOUT; + + while Instant::now() < deadline && response_token.is_none() { + let mut reply_to = None; + network.poll_with(|event| { + if let TcpEvent::Message { group: event_group, token, payload, .. } = event { + assert_eq!(event_group, group); + request_bytes.extend_from_slice(payload); + if request_bytes.windows(4).any(|bytes| bytes == b"\r\n\r\n") { + reply_to = Some(token); + } + } + }); + if let Some(token) = reply_to { + assert!(network.send_with(token, |buf| buf.extend_from_slice(response))); + response_token = Some(token); + } + thread::sleep(Duration::from_millis(1)); + } + let token = response_token.expect("HTTP request was not received"); + assert!(network.disconnect(token)); + + let deadline = Instant::now() + TIMEOUT; + loop { + let mut buffer = [0; 256]; + match client.read(&mut buffer) { + Ok(0) => break, + Ok(read) => received.extend_from_slice(&buffer[..read]), + Err(err) if err.kind() == io::ErrorKind::WouldBlock => { + assert!(Instant::now() < deadline, "HTTP response did not reach EOF"); + network.poll_with(|_| {}); + thread::sleep(Duration::from_millis(1)); + } + Err(err) => panic!("client read failed: {err}"), + } + } + + assert!(received.starts_with(b"HTTP/1.1 200 OK\r\n")); + assert!(received.ends_with(b"ok")); +} + +#[test] +fn raw_outbound_connect() { + let hello = b"raw hello"; + let message = b"raw message"; + let listener = std::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).unwrap(); + listener.set_nonblocking(true).unwrap(); + let addr = listener.local_addr().unwrap(); + let mut network = TcpNetwork::default(); + let group = network.add_group(TcpGroupConfig { + name: "raw-client", + framing: Framing::Raw, + on_connect_msg: Some(hello.to_vec()), + reconnect_interval: flux_timing::Duration::from_millis(1), + ..TcpGroupConfig::default() + }); + let token = network.connect(group, addr); + let mut peer = None; + let deadline = Instant::now() + TIMEOUT; + while Instant::now() < deadline && peer.is_none() { + network.poll_with(|_| {}); + match listener.accept() { + Ok((stream, _)) => { + stream.set_nonblocking(true).unwrap(); + peer = Some(stream); + } + Err(err) if err.kind() == io::ErrorKind::WouldBlock => {} + Err(err) => panic!("accept failed: {err}"), + } + thread::sleep(Duration::from_millis(1)); + } + let mut peer = peer.expect("outbound connection was not accepted"); + + let mut received = Vec::new(); + let deadline = Instant::now() + TIMEOUT; + while Instant::now() < deadline && received.len() < hello.len() { + network.poll_with(|_| {}); + let mut buffer = [0; 128]; + match peer.read(&mut buffer) { + Ok(read) => received.extend_from_slice(&buffer[..read]), + Err(err) if err.kind() == io::ErrorKind::WouldBlock => {} + Err(err) => panic!("peer read failed: {err}"), + } + thread::sleep(Duration::from_millis(1)); + } + assert_eq!(received, hello); + + assert!(network.send_with(token, |buf| buf.extend_from_slice(message))); + let deadline = Instant::now() + TIMEOUT; + while Instant::now() < deadline && received.len() < hello.len() + message.len() { + network.poll_with(|_| {}); + let mut buffer = [0; 128]; + match peer.read(&mut buffer) { + Ok(read) => received.extend_from_slice(&buffer[..read]), + Err(err) if err.kind() == io::ErrorKind::WouldBlock => {} + Err(err) => panic!("peer read failed: {err}"), + } + thread::sleep(Duration::from_millis(1)); + } + assert_eq!(received, [hello.as_slice(), message.as_slice()].concat()); +} + +#[test] +fn framed_and_raw_coexist() { + let framed_addr = unused_addr(); + let raw_addr = unused_addr(); + let mut server = TcpNetwork::default(); + let framed_server = + server.add_group(TcpGroupConfig { name: "framed-server", ..Default::default() }); + let raw_server = server.add_group(raw_group("raw-server")); + server.listen(framed_server, framed_addr).unwrap(); + server.listen(raw_server, raw_addr).unwrap(); + + let mut framed_client = TcpNetwork::default(); + let framed_client_group = framed_client.add_group(TcpGroupConfig { + name: "framed-client", + reconnect_interval: flux_timing::Duration::from_millis(1), + ..Default::default() + }); + let framed_client_token = framed_client.connect(framed_client_group, framed_addr); + let mut raw_client = std::net::TcpStream::connect(raw_addr).unwrap(); + raw_client.set_nonblocking(true).unwrap(); + raw_client.write_all(b"raw").unwrap(); + + let mut framed_server_token = None; + let mut framed_connected = false; + let mut framed_sent = false; + let mut framed_reply = Vec::new(); + let mut raw_reply = Vec::new(); + let deadline = Instant::now() + TIMEOUT; + while Instant::now() < deadline && + (framed_server_token.is_none() || + !framed_connected || + framed_reply != b"framed" || + raw_reply != b"raw") + { + let mut echoes = Vec::new(); + server.poll_with(|event| match event { + TcpEvent::Accepted { group, token, .. } if group == framed_server => { + framed_server_token = Some(token); + } + TcpEvent::Message { group, token, payload, .. } + if group == framed_server || group == raw_server => + { + echoes.push((token, payload.to_vec())); + } + _ => {} + }); + for (token, payload) in echoes { + assert!(server.send_with(token, |buf| buf.extend_from_slice(&payload))); + } + framed_client.poll_with(|event| match event { + TcpEvent::Connected { token, .. } => { + assert_eq!(token, framed_client_token); + framed_connected = true; + } + TcpEvent::Message { payload, .. } => framed_reply.extend_from_slice(payload), + _ => {} + }); + if framed_connected && framed_server_token.is_some() && !framed_sent { + assert!( + framed_client + .send_with(framed_client_token, |buf| buf.extend_from_slice(b"framed")) + ); + framed_sent = true; + } + let mut buffer = [0; 32]; + match raw_client.read(&mut buffer) { + Ok(read) => raw_reply.extend_from_slice(&buffer[..read]), + Err(err) if err.kind() == io::ErrorKind::WouldBlock => {} + Err(err) => panic!("raw client read failed: {err}"), + } + thread::sleep(Duration::from_millis(1)); + } + + assert_eq!(framed_reply, b"framed"); + assert_eq!(raw_reply, b"raw"); +} + +#[test] +fn raw_disconnect_when_drained_flushes_queue() { + let addr = unused_addr(); + let payload = vec![0xA5; 8 * 1024 * 1024]; + let mut network = TcpNetwork::default(); + let group = network.add_group(TcpGroupConfig { + name: "raw-drain", + framing: Framing::Raw, + socket_buf_size: Some(1024), + max_frame_size: payload.len(), + ..TcpGroupConfig::default() + }); + network.listen(group, addr).unwrap(); + + let mut client = std::net::TcpStream::connect(addr).unwrap(); + client.set_nonblocking(true).unwrap(); + let mut token = None; + let deadline = Instant::now() + TIMEOUT; + while Instant::now() < deadline && token.is_none() { + network.poll_with(|event| { + if let TcpEvent::Accepted { group: event_group, token: accepted, .. } = event { + assert_eq!(event_group, group); + token = Some(accepted); + } + }); + thread::sleep(Duration::from_millis(1)); + } + let token = token.expect("raw connection was not accepted"); + + assert!(network.send_with(token, |buf| buf.extend_from_slice(&payload))); + assert!(network.disconnect_when_drained(token)); + assert!(!network.send_with(token, |buf| buf.extend_from_slice(b"late response"))); + + let mut received = Vec::with_capacity(payload.len()); + let deadline = Instant::now() + TIMEOUT; + loop { + let mut buffer = [0; 16 * 1024]; + match client.read(&mut buffer) { + Ok(0) => break, + Ok(read) => received.extend_from_slice(&buffer[..read]), + Err(err) if err.kind() == io::ErrorKind::WouldBlock => { + assert!(Instant::now() < deadline, "raw response did not reach EOF"); + network.poll_with(|_| {}); + thread::sleep(Duration::from_millis(1)); + } + Err(err) => panic!("client read failed: {err}"), + } + } + + assert_eq!(received, payload); +}