redirector-host: reproduce the oracle's connection lifecycle, not just its bytes

Two live gate attempts failed with "An error occurred downloading the FUT
Squad Update" while the redirect response was verified byte-identical to the
Python oracle. Rolling back to the Python redirector fixed it, so the response
bytes were never the whole contract.

Log archaeology found the discriminator: the client polls
/fifa17/fut/rosterupdate.xml ~4x/min in every successful FUT session, and the
only gap in 300 recorded fetches is 03:47-03:58 -- exactly the two
Rust-redirector sessions. The Blaze RPC sequence over those sessions is
identical (msgNum 0-53), so the divergence is entirely outside Blaze.

A differential lifecycle probe against both redirectors found the two
behaviours this host never reproduced:

  * the oracle drains the request body per Content-Length; this host stopped
    at the header terminator, leaving unread data in the receive queue, which
    makes Linux close with RST rather than FIN
  * the oracle holds the connection open ~300ms before closing
    (time.sleep(0.3)); this host closed at 0ms

Both are now reproduced. The dwell is a named constant, ORACLE_CLOSE_DWELL,
overridable only so the causal experiment -- set it to 0, confirm the failure
returns -- can be run without a rebuild.

The suite could not have caught either: it sent Content-Length: 0, so there
was never a body to drain. It now POSTs a body, and asserts a split-write body
is fully consumed.

Testing the drain via client-visible symptoms does NOT work -- verified by
mutation: with the drain removed the client still reads the buffered response
and sees close_notify before any reset. So the host records a per-connection
ConnOutcome and the test asserts on that. Both mutations (no-dwell, no-drain)
are now each caught by exactly one test.

This does not yet prove causation for the FUT Squad Update failure; it removes
the only two measured divergences. Gate 6 is the test.

openfut-redirector-host/src/lib.rs

@@ -113,21 +113,57 @@ pub struct Server {
     listener: TcpListener,
     acceptor: Arc<SslAcceptor>,
     cfg: Arc<RedirectorConfig>,
+    outcomes: Outcomes,
 }
 
+/// What the host did on a connection, beyond the bytes it sent.
+///
+/// Recorded because the failure that cost two live gate attempts — a request
+/// body left unread — is invisible to any comparison of the response. The log
+/// line carries the same facts for a live run; this makes them assertable.
+#[derive(Debug, Clone)]
+pub struct ConnOutcome {
+    pub id: u64,
+    pub request_bytes: usize,
+    pub body: BodyRead,
+    pub response_bytes: usize,
+}
+
+pub type Outcomes = Arc<std::sync::Mutex<Vec<ConnOutcome>>>;
+
+/// Keep the record bounded: a long-running redirector must not accumulate one
+/// entry per connection forever.
+const OUTCOME_HISTORY: usize = 64;
+
 impl Server {
+    /// A handle to the connection record, obtainable before [`Server::run`]
+    /// consumes the server.
+    pub fn outcomes(&self) -> Outcomes {
+        self.outcomes.clone()
+    }
+
     pub fn run(self) -> std::io::Result<()> {
         let counter = AtomicU64::new(0);
         for incoming in self.listener.incoming() {
             let Ok(stream) = incoming else { continue };
             let id = counter.fetch_add(1, Ordering::Relaxed) + 1;
             let (acceptor, cfg) = (self.acceptor.clone(), self.cfg.clone());
-            std::thread::spawn(move || handle(stream, id, &acceptor, &cfg));
+            let outcomes = self.outcomes.clone();
+            std::thread::spawn(move || handle(stream, id, &acceptor, &cfg, &outcomes));
         }
         Ok(())
     }
 }
 
+fn record(outcomes: &Outcomes, o: ConnOutcome) {
+    if let Ok(mut v) = outcomes.lock() {
+        if v.len() >= OUTCOME_HISTORY {
+            v.remove(0);
+        }
+        v.push(o);
+    }
+}
+
 /// Build TLS, rehearse the retail handshake, and bind.
 pub fn bind(cfg: RedirectorConfig) -> std::io::Result<Server> {
     let acceptor = tls::build_acceptor(&cfg.tls)
@@ -154,6 +190,7 @@ pub fn bind(cfg: RedirectorConfig) -> std::io::Result<Server> {
         listener,
         acceptor: Arc::new(acceptor),
         cfg: Arc::new(cfg),
+        outcomes: Outcomes::default(),
     })
 }
 
@@ -162,7 +199,13 @@ pub fn serve(cfg: RedirectorConfig) -> std::io::Result<()> {
     bind(cfg)?.run()
 }
 
-fn handle(stream: TcpStream, id: u64, acceptor: &SslAcceptor, cfg: &RedirectorConfig) {
+fn handle(
+    stream: TcpStream,
+    id: u64,
+    acceptor: &SslAcceptor,
+    cfg: &RedirectorConfig,
+    outcomes: &Outcomes,
+) {
     let peer = stream
         .peer_addr()
         .map(|a| a.to_string())
@@ -215,9 +258,25 @@ fn handle(stream: TcpStream, id: u64, acceptor: &SslAcceptor, cfg: &RedirectorCo
         }
     }
 
+    // Drain the request body, exactly as the oracle does. This is NOT cosmetic:
+    // answering and closing while the client is still sending leaves unread data
+    // in the receive queue, and Linux turns that close into an RST rather than a
+    // FIN. The oracle has always drained; the first version of this host stopped
+    // at the header terminator, which is a behaviour difference invisible to any
+    // byte-comparison of the response.
+    let body_read = drain_body(&mut tls, &mut buf);
+
     let head = String::from_utf8_lossy(&buf);
     let line0 = head.lines().next().unwrap_or("").to_string();
-    log(&format!("conn-{id:04} {peer} REQ {line0}"));
+    log(&format!(
+        "conn-{id:04} {peer} REQ {line0} req_bytes={} body={}",
+        buf.len(),
+        match body_read {
+            BodyRead::None => "none".to_string(),
+            BodyRead::Complete(n) => format!("{n}B complete"),
+            BodyRead::Short { got, want } => format!("{got}B of {want}B SHORT"),
+        }
+    ));
     if !line0.is_empty() && !redirector::is_get_server_instance(&line0) {
         log(&format!(
             "conn-{id:04} {peer} NOTE: unexpected request line; answering anyway (oracle behaviour)"
@@ -237,10 +296,67 @@ fn handle(stream: TcpStream, id: u64, acceptor: &SslAcceptor, cfg: &RedirectorCo
         cfg.adapter.endpoints.blaze_port
     ));
 
-    // The oracle closes after responding; the redirector is a one-shot hop.
+    // The oracle holds the connection open before closing. That dwell is the
+    // only other measured difference between the two implementations, and a
+    // redirector that closes at 0ms is not the behaviour FIFA 17 was proven
+    // against — so it is reproduced rather than assumed harmless.
+    record(
+        outcomes,
+        ConnOutcome {
+            id,
+            request_bytes: buf.len(),
+            body: body_read,
+            response_bytes: response.len(),
+        },
+    );
+
+    std::thread::sleep(cfg.close_dwell);
     let _ = tls.shutdown();
 }
 
+/// What happened when the request body was drained. Reported per connection
+/// because "the client sent a body we never read" is exactly the class of
+/// divergence a response-bytes comparison cannot see.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum BodyRead {
+    /// No `Content-Length` header — nothing to drain.
+    None,
+    Complete(usize),
+    Short { got: usize, want: usize },
+}
+
+/// Parse `Content-Length` and read the remainder of the body into `buf`.
+///
+/// Mirrors the oracle: header lookup is case-insensitive, and a client that
+/// stops early ends the read rather than hanging until the timeout.
+fn drain_body<S: Read>(stream: &mut S, buf: &mut Vec<u8>) -> BodyRead {
+    let Some(sep) = buf.windows(4).position(|w| w == b"\r\n\r\n") else {
+        return BodyRead::None;
+    };
+    let head = String::from_utf8_lossy(&buf[..sep]).to_string();
+    let Some(want) = head
+        .lines()
+        .find(|l| l.to_ascii_lowercase().starts_with("content-length"))
+        .and_then(|l| l.split(':').nth(1))
+        .and_then(|v| v.trim().parse::<usize>().ok())
+    else {
+        return BodyRead::None;
+    };
+
+    let mut got = buf.len() - (sep + 4);
+    let mut chunk = [0u8; 4096];
+    while got < want {
+        match stream.read(&mut chunk) {
+            Ok(0) | Err(_) => return BodyRead::Short { got, want },
+            Ok(n) => {
+                buf.extend_from_slice(&chunk[..n]);
+                got += n;
+            }
+        }
+    }
+    BodyRead::Complete(got)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;