fix: bound HTTP intake and response lifetimes

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",

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" }

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
+`ResourceLimits` bounds request/response bytes, HTTP request-head bytes, JSON
-read/write bytes, tree depth/node count, pinned handles, concurrent requests, and
+nesting, batch items, file read/write bytes, tree depth/node count, pinned handles,
-a cooperative deadline. Advisory file-lock acquisition uses a nonblocking loop and
+and active connections/requests. One absolute request deadline covers header and
-checks that deadline/cancellation rather than allowing a contended lock to stall
+body intake, dispatch, cooperative filesystem work, and response transmission;
-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
+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.

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()
         {

crates/wip-fs-server/src/server.rs

@@ -1,5 +1,5 @@
-use std::io::Read;
+use std::io::{Read, Write};
-use std::net::SocketAddr;
+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<Vec<u8>>) -> Response<Vec<u8>> {
+        self.handle_http_with_deadline(request, Instant::now() + self.config.limits.request_timeout)
+    }
+
+    fn handle_http_with_deadline(
+        &self,
+        request: Request<Vec<u8>>,
+        deadline: Instant,
+    ) -> Response<Vec<u8>> {
         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<BoundServer, ServerBuildError> {
-        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<WipFsServer>,
-    listener: Arc<tiny_http::Server>,
+    listener: Arc<TcpListener>,
 }
 
 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<AtomicBool>) -> 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)) {
+                    match listener.accept() {
-                        Ok(Some(request)) => serve_request(&service, request),
+                        Ok((stream, _)) => serve_connection(&service, stream),
-                        Ok(None) => {}
+                        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) {
+#[derive(Debug)]
-    let maximum = service.config.limits.wire.max_request_bytes();
+struct ParsedRequestHead {
-    let mut body = Vec::new();
+    method: http::Method,
-    let read_result = request
+    target: String,
-        .as_reader()
+    headers: Vec<(http::header::HeaderName, http::HeaderValue)>,
-        .take(maximum.saturating_add(1) as u64)
+    content_length: usize,
-        .read_to_end(&mut body);
+}
-    let route = service.route(request.url().split('?').next().unwrap_or_default());
+
-    let response = if read_result.is_err() {
+#[derive(Debug)]
-        empty_response(StatusCode::BAD_REQUEST)
+enum IntakeError {
-    } else if body.len() > maximum {
+    Timeout,
-        route.map_or_else(
+    HeaderTooLarge,
-            || empty_response(StatusCode::PAYLOAD_TOO_LARGE),
+    PayloadTooLarge { target: String },
-            |route| {
+    BadRequest,
-                if route == Route::Batch {
+    Socket,
-                    service.batch_top_error(
+}
-                        &resource_error(),
+
-                        Some(ResourceLimitKind::RepresentationBytes),
+fn serve_connection(service: &WipFsServer, mut stream: TcpStream) {
-                    )
+    let _ = stream.set_nonblocking(false);
-                } else {
+    let started = Instant::now();
-                    service.protocol_error(
+    let deadline = started
-                        interaction(route),
+        .checked_add(service.config.limits.request_timeout)
-                        &resource_error(),
+        .unwrap_or(started);
-                        Some(ResourceLimitKind::RepresentationBytes),
+    let response = match read_http_request(
-                    )
+        &mut stream,
-                }
+        service.config.limits.max_http_header_bytes,
-            },
+        service.config.limits.wire.max_request_bytes(),
-        )
+        deadline,
-    } else {
+    ) {
-        let mut builder = Request::builder()
+        Ok(request) => Some(service.handle_http_with_deadline(request, deadline)),
-            .method(request.method().as_str())
+        Err(IntakeError::HeaderTooLarge) => {
-            .uri(request.url());
+            Some(empty_response(StatusCode::REQUEST_HEADER_FIELDS_TOO_LARGE))
-        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);
-            }
         }
-        match builder.body(body) {
+        Err(IntakeError::PayloadTooLarge { target }) => {
-            Ok(request) => service.handle_http(request),
+            Some(payload_too_large_response(service, &target))
-            Err(_) => empty_response(StatusCode::BAD_REQUEST),
         }
+        Err(IntakeError::BadRequest) => Some(empty_response(StatusCode::BAD_REQUEST)),
+        Err(IntakeError::Timeout | IntakeError::Socket) => None,
     };
-    let status = tiny_http::StatusCode(response.status().as_u16());
+    if let Some(response) = response {
-    let mut outgoing =
+        let _ = write_http_response(&mut stream, &response, deadline);
-        tiny_http::Response::from_data(response.body().clone()).with_status_code(status);
+    }
+    let _ = stream.shutdown(std::net::Shutdown::Both);
+}
+
+fn read_http_request(
+    stream: &mut TcpStream,
+    maximum_header: usize,
+    maximum_body: usize,
+    deadline: Instant,
+) -> Result<Request<Vec<u8>>, 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<ParsedRequestHead, IntakeError> {
+    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::<usize>().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<usize, IntakeError> {
+    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<Vec<u8>> {
+    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<Vec<u8>>,
+    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()) {
+        if name == http::header::CONTENT_LENGTH
-            outgoing.add_header(header);
+            || 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 {

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");