wai/changes/2026-11-05-origin-refusal-ends-cleanly.md

@@ -0,0 +1,6 @@
+### `wai http origin`: a refusal ends cleanly
+
+- The origin answered 503 and closed with the request unread, so the connection ended in a reset
+  after the answer, and a client reading to the close saw an error despite a valid 503. For a client
+  past `--max-per-peer` or `--max-connections`, the origin now stops sending and reads a little more
+  before closing, as it already did for a 431.

wai/wai-rs/src/bin/wai_cli/http.rs

@@ -528,25 +528,30 @@ fn respond(s: &TcpStream, until: Instant, status: u16, headers: &[(&str, String)
     Ok(())
 }
 
+/// Stop sending and read a little more before closing. Closing with the request unread makes the
+/// kernel send a reset after the answer, which a client reading to the close takes for a failure,
+/// and which some network stacks let discard what the client has not yet read.
+fn drain_before_close(mut stream: &TcpStream) {
+    let _ = stream.shutdown(std::net::Shutdown::Write);
+    let _ = stream.set_read_timeout(Some(Duration::from_millis(200)));
+    let mut sink = [0u8; 64 * 1024];
+    let (t0, mut drained) = (Instant::now(), 0usize);
+    while t0.elapsed() < Duration::from_secs(1) && drained < 1 << 20 {
+        match stream.read(&mut sink) {
+            Ok(n) if n > 0 => drained += n,
+            _ => break,
+        }
+    }
+}
+
 fn serve_one(cfg: &OriginConfig, stream: TcpStream, accepted: Instant) -> std::io::Result<()> {
     let until = accepted + cfg.lifetime;
     let req = match read_request(&stream, (accepted + cfg.read_timeout).min(until)) {
         Read1::Request(r) => r,
         Read1::TimedOut => return respond(&stream, until, 408, &[], b"request head not received in time\n", false),
         Read1::TooLarge => {
-            // Answer, then stop sending and read a little more before closing, so the
-            // answer is not lost to a reset while the client is still writing.
             respond(&stream, until, 431, &[], b"request head too large\n", false)?;
-            let _ = stream.shutdown(std::net::Shutdown::Write);
-            let _ = stream.set_read_timeout(Some(Duration::from_millis(200)));
-            let mut sink = [0u8; 64 * 1024];
-            let (t0, mut drained) = (Instant::now(), 0usize);
-            while t0.elapsed() < Duration::from_secs(1) && drained < 1 << 20 {
-                match (&stream).read(&mut sink) {
-                    Ok(n) if n > 0 => drained += n,
-                    _ => break,
-                }
-            }
+            drain_before_close(&stream);
             return Ok(());
         }
         Read1::Bad => return respond(&stream, until, 400, &[], b"bad request\n", false),
@@ -642,6 +647,7 @@ fn origin_cmd(args: &[String]) -> Result<(), String> {
             use std::sync::atomic::Ordering;
             let busy = |stream: &TcpStream| {
                 let _ = respond(stream, Instant::now() + cfg.read_timeout, 503, &[("Retry-After", "1".to_string())], b"busy\n", false);
+                drain_before_close(stream);
             };
             // At most `max_connections` in all and, when `--max-per-peer` is given, that
             // many from one address; one more is told so and closed.

wai/wai-rs/tests/cli_http.rs

@@ -377,14 +377,40 @@ fn origin_cuts_a_slow_reader_at_the_lifetime() {
 fn origin_per_peer_limit() {
     let root = scratch("origin-peer");
     std::fs::write(root.join("x.json"), b"{}").unwrap();
-    let origin = Origin::start(&root, &["--max-per-peer", "2", "--read-timeout-ms", "2000"]);
-    let idle: Vec<TcpStream> = (0..2).map(|_| TcpStream::connect(("127.0.0.1", origin.port)).unwrap()).collect();
-    std::thread::sleep(std::time::Duration::from_millis(200));
-    let r = raw(origin.port, b"GET /x.json HTTP/1.1\r\nHost: a\r\n\r\n");
+    let origin = Origin::start(&root, &["--max-per-peer", "2", "--read-timeout-ms", "60000"]);
+    let request = b"GET /x.json HTTP/1.1\r\nHost: a\r\n\r\n";
+    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
+    // The origin counts each connection on a thread of its own, so a third can overtake the idle
+    // two and one of them is refused instead. An attempt that went that way starts again.
+    let (idle, r) = loop {
+        let idle: Vec<TcpStream> = (0..2).map(|_| TcpStream::connect(("127.0.0.1", origin.port)).unwrap()).collect();
+        std::thread::sleep(std::time::Duration::from_millis(100));
+        let r = raw(origin.port, request);
+        if r.starts_with("HTTP/1.1 503") || std::time::Instant::now() > deadline {
+            break (idle, r);
+        }
+    };
     assert!(r.starts_with("HTTP/1.1 503"), "{r:?}");
+    // A refusal ends cleanly: closing with the request unread would end the connection in a reset,
+    // which a client reading to the close takes for a failure, after a valid answer. So the read
+    // itself must succeed here, and no attempt is retried.
+    for i in 0..200 {
+        let mut s = TcpStream::connect(("127.0.0.1", origin.port)).unwrap();
+        s.set_read_timeout(Some(std::time::Duration::from_secs(10))).unwrap();
+        let mut out = Vec::new();
+        let read = s.write_all(request).and_then(|()| s.read_to_end(&mut out));
+        let r = String::from_utf8_lossy(&out);
+        assert!(read.is_ok() && r.starts_with("HTTP/1.1 503") && r.contains("Retry-After: 1"), "refusal {i}: {read:?} {r:?}");
+    }
     drop(idle);
-    std::thread::sleep(std::time::Duration::from_millis(200));
-    assert!(raw(origin.port, b"GET /x.json HTTP/1.1\r\nHost: a\r\n\r\n").starts_with("HTTP/1.1 200"));
+    let r = loop {
+        let r = raw(origin.port, request);
+        if r.starts_with("HTTP/1.1 200") || std::time::Instant::now() > deadline {
+            break r;
+        }
+        std::thread::sleep(std::time::Duration::from_millis(50));
+    };
+    assert!(r.starts_with("HTTP/1.1 200"), "{r:?}");
 }
 
 /// N3: `wai http bench` times the base from the session's `base_ready`: with the origin