From 780150d7ae3f1b0d18356c13c361d042f094e237 Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 30 Sep 2026 09:23:12 +0900 Subject: [PATCH] fix: bound HTTP intake and response lifetimes --- Cargo.lock | 37 --- crates/wip-fs-server/Cargo.toml | 1 - crates/wip-fs-server/README.md | 16 +- crates/wip-fs-server/src/lib.rs | 6 +- crates/wip-fs-server/src/server.rs | 340 +++++++++++++++++++---- crates/wip-fs-server/tests/end_to_end.rs | 70 ++++- 6 files changed, 364 insertions(+), 106 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 5df8602..7deeae2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,12 +2,6 @@ # It is not intended for manual editing. version = 4 -[[package]] -name = "ascii" -version = "1.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d92bec98840b8f03a5ff5413de5293bfcd8bf96467cf5452609f939ec6f5de16" - [[package]] name = "base64" version = "0.22.1" @@ -32,12 +26,6 @@ version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4e7648175b45a9a48536d676f68d918270699102aa8dab5496df06904c914600" -[[package]] -name = "chunked_transfer" -version = "1.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6e4de3bc4ea267985becf712dc6d9eed8b04c953b3fcfb339ebc87acd9804901" - [[package]] name = "displaydoc" version = "0.2.7" @@ -113,12 +101,6 @@ dependencies = [ "itoa", ] -[[package]] -name = "httpdate" -version = "1.0.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" - [[package]] name = "icu_collections" version = "2.3.0" @@ -257,12 +239,6 @@ version = "0.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "47d9d19d1d6efa0109d2f65ff4c85cddd50bd572e5a00127ab10987290bcefae" -[[package]] -name = "log" -version = "0.4.34" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6" - [[package]] name = "memchr" version = "2.8.3" @@ -451,18 +427,6 @@ dependencies = [ "winapi-util", ] -[[package]] -name = "tiny_http" -version = "0.12.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "389915df6413a2e74fb181895f933386023c71110878cd0825588928e64cdc82" -dependencies = [ - "ascii", - "chunked_transfer", - "httpdate", - "log", -] - [[package]] name = "tinystr" version = "0.8.4" @@ -588,7 +552,6 @@ dependencies = [ "http", "rustix", "tempfile", - "tiny_http", "wip-host", "wip-https", "wip-interface", diff --git a/crates/wip-fs-server/Cargo.toml b/crates/wip-fs-server/Cargo.toml index e0707f5..1aa9210 100644 --- a/crates/wip-fs-server/Cargo.toml +++ b/crates/wip-fs-server/Cargo.toml @@ -10,7 +10,6 @@ publish = false [dependencies] http = "1" rustix = { version = "1.1", features = ["fs"] } -tiny_http = "0.12" wip-host = { path = "../wip-host" } wip-https = { path = "../wip-https" } wip-interface = { path = "../wip-interface" } diff --git a/crates/wip-fs-server/README.md b/crates/wip-fs-server/README.md index b2685b9..00f151a 100644 --- a/crates/wip-fs-server/README.md +++ b/crates/wip-fs-server/README.md @@ -125,11 +125,17 @@ standard `POST /wip/v1/batch` route. `call_operation` is never batchable. ## Resource and failure behavior -`ResourceLimits` bounds request/response bytes, JSON nesting, batch items, file -read/write bytes, tree depth/node count, pinned handles, concurrent requests, and -a cooperative deadline. Advisory file-lock acquisition uses a nonblocking loop and -checks that deadline/cancellation rather than allowing a contended lock to stall -a worker. Tree traversal is breadth-first and opens only requested entries. Once a limit or depth boundary is reached it reads only far enough to +`ResourceLimits` bounds request/response bytes, HTTP request-head bytes, JSON +nesting, batch items, file read/write bytes, tree depth/node count, pinned handles, +and active connections/requests. One absolute request deadline covers header and +body intake, dispatch, cooperative filesystem work, and response transmission; +stalled clients are closed and cannot occupy a worker beyond that bound. Workers +accept connections directly, so the listener never moves more than the configured +concurrency into active userspace service. Advisory file-lock acquisition uses a +nonblocking loop and checks the same deadline/cancellation rather than allowing a +contended lock to stall a worker. Tree traversal is breadth-first and opens only +requested entries. Once a limit or depth boundary is reached it reads only far +enough to prove whether an additional visible child exists, so `truncated` is true only when something was actually omitted. I/O failure and cancellation remain errors, not clean exhaustion. diff --git a/crates/wip-fs-server/src/lib.rs b/crates/wip-fs-server/src/lib.rs index cb5124b..c76b93e 100644 --- a/crates/wip-fs-server/src/lib.rs +++ b/crates/wip-fs-server/src/lib.rs @@ -57,9 +57,11 @@ pub struct ResourceLimits { pub max_tree_nodes: u32, /// Maximum simultaneously pinned projected filesystem handles. pub max_open_handles: usize, + /// Maximum bytes accepted for one HTTP request head, including its terminator. + pub max_http_header_bytes: usize, /// Maximum requests executing concurrently. pub max_concurrent_requests: usize, - /// Deadline applied to request dispatch and cooperative filesystem work. + /// End-to-end deadline for HTTP intake, dispatch, and response transmission. pub request_timeout: Duration, } @@ -73,6 +75,7 @@ impl Default for ResourceLimits { max_tree_depth: 32, max_tree_nodes: 1024, max_open_handles: 2048, + max_http_header_bytes: 16 * 1024, max_concurrent_requests: 32, request_timeout: Duration::from_secs(30), } @@ -124,6 +127,7 @@ impl ServerConfig { || self.limits.max_file_write_bytes == 0 || self.limits.max_tree_nodes == 0 || self.limits.max_open_handles == 0 + || self.limits.max_http_header_bytes == 0 || self.limits.max_concurrent_requests == 0 || self.limits.request_timeout.is_zero() { diff --git a/crates/wip-fs-server/src/server.rs b/crates/wip-fs-server/src/server.rs index ae36a62..dbb581a 100644 --- a/crates/wip-fs-server/src/server.rs +++ b/crates/wip-fs-server/src/server.rs @@ -1,5 +1,5 @@ -use std::io::Read; -use std::net::SocketAddr; +use std::io::{Read, Write}; +use std::net::{SocketAddr, TcpListener, TcpStream}; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::thread; @@ -62,6 +62,14 @@ impl WipFsServer { /// Listener integrations must enforce the same body limit while collecting bytes. #[must_use] pub fn handle_http(&self, request: Request>) -> Response> { + self.handle_http_with_deadline(request, Instant::now() + self.config.limits.request_timeout) + } + + fn handle_http_with_deadline( + &self, + request: Request>, + deadline: Instant, + ) -> Response> { let Some(route) = self.route(request.uri().path()) else { return empty_response(StatusCode::NOT_FOUND); }; @@ -69,7 +77,7 @@ impl WipFsServer { return self.metadata_error(route, error); } let context = ExecutionContext::new() - .with_deadline(Instant::now() + self.config.limits.request_timeout) + .with_deadline(deadline) .with_resource_budget(u64::from(self.config.limits.max_tree_nodes)); match route { Route::Fetch => self.fetch(request.body(), &context), @@ -85,16 +93,18 @@ impl WipFsServer { /// The listener is cleartext by design. Configuration must explicitly state /// whether TLS is terminated by a reverse proxy or development HTTP is in use. pub fn bind(self) -> Result { - let server = tiny_http::Server::http(self.config.listen) + let listener = TcpListener::bind(self.config.listen) + .map_err(|error| ServerBuildError::Listener(error.to_string()))?; + listener + .set_nonblocking(true) + .map_err(|error| ServerBuildError::Listener(error.to_string()))?; + let local_addr = listener + .local_addr() .map_err(|error| ServerBuildError::Listener(error.to_string()))?; - let local_addr = server - .server_addr() - .to_ip() - .ok_or_else(|| ServerBuildError::Listener("expected an IP listener".into()))?; Ok(BoundServer { local_addr, service: Arc::new(self), - listener: Arc::new(server), + listener: Arc::new(listener), }) } @@ -356,7 +366,7 @@ impl WipFsServer { pub struct BoundServer { local_addr: SocketAddr, service: Arc, - listener: Arc, + listener: Arc, } impl BoundServer { @@ -367,6 +377,11 @@ impl BoundServer { } /// Serves requests with a fixed worker count until `shutdown` becomes true. + /// + /// Workers accept connections directly, so at most + /// `max_concurrent_requests` sockets are active in userspace. Each connection + /// is closed after one request and shares one absolute deadline across HTTP + /// intake, dispatch, and response transmission. pub fn serve_until(&self, shutdown: Arc) -> Result<(), std::io::Error> { let mut workers = Vec::with_capacity(self.service.config.limits.max_concurrent_requests); for _ in 0..self.service.config.limits.max_concurrent_requests { @@ -375,9 +390,12 @@ impl BoundServer { let shutdown = Arc::clone(&shutdown); workers.push(thread::spawn(move || { while !shutdown.load(Ordering::Acquire) { - match listener.recv_timeout(Duration::from_millis(25)) { - Ok(Some(request)) => serve_request(&service, request), - Ok(None) => {} + match listener.accept() { + Ok((stream, _)) => serve_connection(&service, stream), + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { + thread::sleep(Duration::from_millis(5)); + } + Err(error) if error.kind() == std::io::ErrorKind::Interrupted => {} Err(error) => return Err(error), } } @@ -398,60 +416,260 @@ impl BoundServer { } } -fn serve_request(service: &WipFsServer, mut request: tiny_http::Request) { - let maximum = service.config.limits.wire.max_request_bytes(); - let mut body = Vec::new(); - let read_result = request - .as_reader() - .take(maximum.saturating_add(1) as u64) - .read_to_end(&mut body); - let route = service.route(request.url().split('?').next().unwrap_or_default()); - let response = if read_result.is_err() { - empty_response(StatusCode::BAD_REQUEST) - } else if body.len() > maximum { - route.map_or_else( - || empty_response(StatusCode::PAYLOAD_TOO_LARGE), - |route| { - if route == Route::Batch { - service.batch_top_error( - &resource_error(), - Some(ResourceLimitKind::RepresentationBytes), - ) - } else { - service.protocol_error( - interaction(route), - &resource_error(), - Some(ResourceLimitKind::RepresentationBytes), - ) - } - }, - ) - } else { - let mut builder = Request::builder() - .method(request.method().as_str()) - .uri(request.url()); - for header in request.headers() { - if let (Ok(name), Ok(value)) = ( - http::header::HeaderName::from_bytes(header.field.as_str().as_bytes()), - http::HeaderValue::from_bytes(header.value.as_bytes()), - ) { - builder = builder.header(name, value); - } +#[derive(Debug)] +struct ParsedRequestHead { + method: http::Method, + target: String, + headers: Vec<(http::header::HeaderName, http::HeaderValue)>, + content_length: usize, +} + +#[derive(Debug)] +enum IntakeError { + Timeout, + HeaderTooLarge, + PayloadTooLarge { target: String }, + BadRequest, + Socket, +} + +fn serve_connection(service: &WipFsServer, mut stream: TcpStream) { + let _ = stream.set_nonblocking(false); + let started = Instant::now(); + let deadline = started + .checked_add(service.config.limits.request_timeout) + .unwrap_or(started); + let response = match read_http_request( + &mut stream, + service.config.limits.max_http_header_bytes, + service.config.limits.wire.max_request_bytes(), + deadline, + ) { + Ok(request) => Some(service.handle_http_with_deadline(request, deadline)), + Err(IntakeError::HeaderTooLarge) => { + Some(empty_response(StatusCode::REQUEST_HEADER_FIELDS_TOO_LARGE)) } - match builder.body(body) { - Ok(request) => service.handle_http(request), - Err(_) => empty_response(StatusCode::BAD_REQUEST), + Err(IntakeError::PayloadTooLarge { target }) => { + Some(payload_too_large_response(service, &target)) } + Err(IntakeError::BadRequest) => Some(empty_response(StatusCode::BAD_REQUEST)), + Err(IntakeError::Timeout | IntakeError::Socket) => None, }; - let status = tiny_http::StatusCode(response.status().as_u16()); - let mut outgoing = - tiny_http::Response::from_data(response.body().clone()).with_status_code(status); + if let Some(response) = response { + let _ = write_http_response(&mut stream, &response, deadline); + } + let _ = stream.shutdown(std::net::Shutdown::Both); +} + +fn read_http_request( + stream: &mut TcpStream, + maximum_header: usize, + maximum_body: usize, + deadline: Instant, +) -> Result>, IntakeError> { + let mut received = Vec::with_capacity(maximum_header.min(8 * 1024)); + let header_end = loop { + if let Some(position) = received.windows(4).position(|window| window == b"\r\n\r\n") { + let end = position + 4; + if end > maximum_header { + return Err(IntakeError::HeaderTooLarge); + } + break end; + } + if received.len() >= maximum_header { + return Err(IntakeError::HeaderTooLarge); + } + let mut chunk = [0_u8; 4 * 1024]; + let available = (maximum_header - received.len()).min(chunk.len()); + let count = read_with_deadline(stream, &mut chunk[..available], deadline)?; + if count == 0 { + return Err(IntakeError::BadRequest); + } + received.extend_from_slice(&chunk[..count]); + }; + + let parsed = parse_request_head(&received[..header_end])?; + if parsed.content_length > maximum_body { + return Err(IntakeError::PayloadTooLarge { + target: parsed.target, + }); + } + let mut body = Vec::with_capacity(parsed.content_length); + let buffered = &received[header_end..]; + body.extend_from_slice(&buffered[..buffered.len().min(parsed.content_length)]); + while body.len() < parsed.content_length { + let mut chunk = [0_u8; 8 * 1024]; + let wanted = (parsed.content_length - body.len()).min(chunk.len()); + let count = read_with_deadline(stream, &mut chunk[..wanted], deadline)?; + if count == 0 { + return Err(IntakeError::BadRequest); + } + body.extend_from_slice(&chunk[..count]); + } + + let mut builder = Request::builder().method(parsed.method).uri(parsed.target); + for (name, value) in parsed.headers { + builder = builder.header(name, value); + } + builder.body(body).map_err(|_| IntakeError::BadRequest) +} + +fn parse_request_head(bytes: &[u8]) -> Result { + let text = std::str::from_utf8(bytes).map_err(|_| IntakeError::BadRequest)?; + let mut lines = text + .strip_suffix("\r\n\r\n") + .ok_or(IntakeError::BadRequest)? + .split("\r\n"); + let mut request_line = lines + .next() + .ok_or(IntakeError::BadRequest)? + .split_ascii_whitespace(); + let method = request_line.next().ok_or(IntakeError::BadRequest)?; + let target = request_line.next().ok_or(IntakeError::BadRequest)?; + let version = request_line.next().ok_or(IntakeError::BadRequest)?; + if request_line.next().is_some() + || !matches!(version, "HTTP/1.0" | "HTTP/1.1") + || !target.starts_with('/') + { + return Err(IntakeError::BadRequest); + } + let method = + http::Method::from_bytes(method.as_bytes()).map_err(|_| IntakeError::BadRequest)?; + let _: http::Uri = target.parse().map_err(|_| IntakeError::BadRequest)?; + + let mut headers = Vec::new(); + let mut content_length = None; + for line in lines { + if line.is_empty() || line.starts_with([' ', '\t']) { + return Err(IntakeError::BadRequest); + } + let (name, value) = line.split_once(':').ok_or(IntakeError::BadRequest)?; + let name = http::header::HeaderName::from_bytes(name.as_bytes()) + .map_err(|_| IntakeError::BadRequest)?; + let value = + http::HeaderValue::from_str(value.trim()).map_err(|_| IntakeError::BadRequest)?; + if name == http::header::CONTENT_LENGTH { + if content_length.is_some() { + return Err(IntakeError::BadRequest); + } + content_length = Some( + value + .to_str() + .ok() + .and_then(|value| value.parse::().ok()) + .ok_or(IntakeError::BadRequest)?, + ); + } + if name == http::header::TRANSFER_ENCODING { + return Err(IntakeError::BadRequest); + } + headers.push((name, value)); + } + + Ok(ParsedRequestHead { + method, + target: target.to_owned(), + headers, + content_length: content_length.unwrap_or(0), + }) +} + +fn read_with_deadline( + stream: &mut TcpStream, + buffer: &mut [u8], + deadline: Instant, +) -> Result { + let remaining = deadline + .checked_duration_since(Instant::now()) + .ok_or(IntakeError::Timeout)?; + stream + .set_read_timeout(Some(remaining.max(Duration::from_millis(1)))) + .map_err(|_| IntakeError::Socket)?; + stream.read(buffer).map_err(|error| match error.kind() { + std::io::ErrorKind::TimedOut | std::io::ErrorKind::WouldBlock => IntakeError::Timeout, + _ => IntakeError::Socket, + }) +} + +fn payload_too_large_response(service: &WipFsServer, target: &str) -> Response> { + let route = service.route(target.split('?').next().unwrap_or_default()); + route.map_or_else( + || empty_response(StatusCode::PAYLOAD_TOO_LARGE), + |route| { + if route == Route::Batch { + service.batch_top_error( + &resource_error(), + Some(ResourceLimitKind::RepresentationBytes), + ) + } else { + service.protocol_error( + interaction(route), + &resource_error(), + Some(ResourceLimitKind::RepresentationBytes), + ) + } + }, + ) +} + +fn write_http_response( + stream: &mut TcpStream, + response: &Response>, + deadline: Instant, +) -> Result<(), std::io::Error> { + let mut encoded = Vec::new(); + write!( + encoded, + "HTTP/1.1 {} {}\r\n", + response.status().as_u16(), + response.status().canonical_reason().unwrap_or("") + )?; for (name, value) in response.headers() { - if let Ok(header) = tiny_http::Header::from_bytes(name.as_str(), value.as_bytes()) { - outgoing.add_header(header); + if name == http::header::CONTENT_LENGTH + || name == http::header::CONNECTION + || name == http::header::TRANSFER_ENCODING + { + continue; + } + encoded.extend_from_slice(name.as_str().as_bytes()); + encoded.extend_from_slice(b": "); + encoded.extend_from_slice(value.as_bytes()); + encoded.extend_from_slice(b"\r\n"); + } + write!( + encoded, + "Content-Length: {}\r\nConnection: close\r\n\r\n", + response.body().len() + )?; + encoded.extend_from_slice(response.body()); + write_all_with_deadline(stream, &encoded, deadline) +} + +fn write_all_with_deadline( + stream: &mut TcpStream, + mut bytes: &[u8], + deadline: Instant, +) -> Result<(), std::io::Error> { + while !bytes.is_empty() { + let remaining = deadline + .checked_duration_since(Instant::now()) + .ok_or_else(|| { + std::io::Error::new(std::io::ErrorKind::TimedOut, "request timed out") + })?; + stream.set_write_timeout(Some(remaining.max(Duration::from_millis(1))))?; + match stream.write(bytes) { + Ok(0) => { + return Err(std::io::Error::new( + std::io::ErrorKind::WriteZero, + "socket closed", + )); + } + Ok(count) => bytes = &bytes[count..], + Err(error) if error.kind() == std::io::ErrorKind::Interrupted => {} + Err(error) => return Err(error), } } - let _ = request.respond(outgoing); + stream.flush() } fn interaction(route: Route) -> ProtocolInteraction { diff --git a/crates/wip-fs-server/tests/end_to_end.rs b/crates/wip-fs-server/tests/end_to_end.rs index 71bcc5b..3de1f6b 100644 --- a/crates/wip-fs-server/tests/end_to_end.rs +++ b/crates/wip-fs-server/tests/end_to_end.rs @@ -4,7 +4,7 @@ use std::net::{SocketAddr, TcpStream}; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::thread; -use std::time::Duration; +use std::time::{Duration, Instant}; use tempfile::tempdir; use wip_fs_server::{AccessMode, ServerConfig, TransportSecurity, WipFsServer}; @@ -202,6 +202,74 @@ fn real_listener_serves_four_core_interactions_batch_and_write() { .expect("listener exits cleanly"); } +#[test] +fn stalled_request_body_times_out_and_releases_the_only_worker() { + let temporary = tempdir().expect("temporary root"); + let mut config = ServerConfig::for_root(temporary.path()); + config.transport_security = TransportSecurity::DevelopmentHttp; + config.limits.max_concurrent_requests = 1; + config.limits.request_timeout = Duration::from_millis(200); + let limits = config.limits.wire; + let bound = Arc::new( + WipFsServer::new(config) + .expect("build server") + .bind() + .expect("bind"), + ); + let address = bound.local_addr(); + let shutdown = Arc::new(AtomicBool::new(false)); + let worker = { + let bound = Arc::clone(&bound); + let shutdown = Arc::clone(&shutdown); + thread::spawn(move || bound.serve_until(shutdown)) + }; + + let mut stalled = TcpStream::connect(address).expect("connect stalled client"); + stalled + .set_read_timeout(Some(Duration::from_secs(1))) + .expect("set stalled read timeout"); + write!( + stalled, + "POST /wip/v1/fetch HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nAccept: application/json\r\nContent-Length: 100\r\n\r\n{{" + ) + .expect("write partial request"); + stalled.flush().expect("flush partial request"); + thread::sleep(Duration::from_millis(25)); + + let endpoint = Endpoint::parse(&format!("https://localhost:{}/wip", address.port())) + .expect("client endpoint"); + let fetch = FetchRequest { path: "/".into() }; + let started = Instant::now(); + let response = exchange( + address, + encode_fetch_request(&endpoint, &fetch, limits).expect("encode fetch"), + ); + let _root = success(decode_fetch_response(&fetch, &response, limits).expect("decode fetch")); + assert!( + started.elapsed() < Duration::from_secs(1), + "the valid request remained starved after the intake deadline" + ); + + let mut byte = [0_u8; 1]; + match stalled.read(&mut byte) { + Ok(0) => {} + Err(error) + if matches!( + error.kind(), + std::io::ErrorKind::ConnectionReset + | std::io::ErrorKind::ConnectionAborted + | std::io::ErrorKind::BrokenPipe + ) => {} + outcome => panic!("stalled connection was not closed after its deadline: {outcome:?}"), + } + + shutdown.store(true, Ordering::Release); + worker + .join() + .expect("join listener") + .expect("listener exits cleanly"); +} + #[test] fn batch_keeps_known_malformed_arguments_item_local() { let temporary = tempdir().expect("temporary root");