OpenFUT/openfut-blaze-host/src/conn.rs

//! Per-connection Fire2 stream handling.
//!
//! This is the layer fixtures cannot test: a socket delivers bytes, not frames.
//! Frames arrive split across reads and coalesced into one, several arrive per
//! connection, and the connection outlives every individual RPC.
//!
//! The loop mirrors the Python oracle's exactly — read a 16-byte header, derive
//! the total length from it, read the rest, consume, dispatch, write every
//! returned frame in order — because that behaviour is part of what was proven
//! against the real client.

use std::io::{self, Read, Write};
use std::net::TcpStream;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use openfut_adapter_fifa17::blaze::{Adapter, Session};
use openfut_protocol_blaze::fire2::{Frame, Header, HEADER_LEN};
use openfut_protocol_blaze::heat2::{self, Struct};

use crate::capture::{Capture, Direction};
use crate::config::HostConfig;
use crate::trace::{self, Tracer};
use crate::Hooks;

/// Why a connection ended. Logged so a live run can be compared with Python's.
#[derive(Debug, PartialEq, Eq)]
pub enum CloseReason {
    /// Client closed cleanly between frames.
    ClientClosed,
    /// Stream ended part-way through a frame.
    EofMidFrame { want: usize, have: usize },
    /// A header claimed a payload larger than the configured ceiling.
    AbsurdPayload { claimed: u32 },
    /// No bytes within the idle timeout.
    IdleTimeout,
    /// Socket error.
    Io(String),
}

impl std::fmt::Display for CloseReason {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            CloseReason::ClientClosed => write!(f, "client closed"),
            CloseReason::EofMidFrame { want, have } => {
                write!(f, "EOF mid-frame (want {want}, have {have})")
            }
            CloseReason::AbsurdPayload { claimed } => {
                write!(f, "absurd payload_len {claimed}, dropping connection")
            }
            CloseReason::IdleTimeout => write!(f, "idle timeout"),
            CloseReason::Io(e) => write!(f, "io error: {e}"),
        }
    }
}

pub(crate) fn unix_now() -> i64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs() as i64)
        .unwrap_or(0)
}

/// Mint a Blaze-shaped session key.
///
/// The client never validates the format, so this only has to be stable within
/// a connection and distinct between them.
pub(crate) fn mint_session_key() -> String {
    use rand::Rng;
    let mut rng = rand::thread_rng();
    const ALPHA: &[u8] = b"abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789$*";
    let tail: String = (0..44)
        .map(|_| ALPHA[rng.gen_range(0..ALPHA.len())] as char)
        .collect();
    openfut_adapter_fifa17::blaze::session::format_session_key(rng.gen::<u64>(), &tail)
}

/// Serve one connection until it closes.
pub fn handle(
    mut stream: TcpStream,
    conn_id: u64,
    cfg: &HostConfig,
    adapter: &Adapter,
    tracer: &Tracer,
    hooks: &Hooks,
    capture: &Capture,
) -> CloseReason {
    let peer = stream
        .peer_addr()
        .map(|a| a.to_string())
        .unwrap_or_else(|_| "<unknown>".into());

    let _ = stream.set_read_timeout(Some(Duration::from_secs(cfg.idle_timeout_secs)));
    // Blaze is request/response with server pushes; Nagle would add latency to
    // every burst for no benefit.
    let _ = stream.set_nodelay(true);

    let mut session = Session::new((hooks.session_key)(), cfg.adapter.identity.account_locale);
    trace::log(&format!(
        "conn-{conn_id:04} CONNECT from {peer} (session key minted: [REDACTED])"
    ));
    tracer.write_line(&format!("conn-{conn_id:04} OPEN"));

    let mut buf: Vec<u8> = Vec::with_capacity(16 * 1024);
    let mut frame_no: u64 = 0;

    let reason = loop {
        // ---- header
        match fill_to(&mut stream, &mut buf, HEADER_LEN) {
            Ok(true) => {}
            Ok(false) => {
                break if buf.is_empty() {
                    CloseReason::ClientClosed
                } else {
                    CloseReason::EofMidFrame {
                        want: HEADER_LEN,
                        have: buf.len(),
                    }
                };
            }
            Err(e) => break classify(e),
        }

        let header = match Header::parse(&buf[..HEADER_LEN]) {
            Ok(h) => h,
            // Unreachable: fill_to guaranteed 16 bytes. Treated as a close
            // rather than a panic because this is a network path.
            Err(e) => break CloseReason::Io(e.to_string()),
        };

        if header.payload_len > cfg.max_payload_bytes {
            trace::log(&format!(
                "conn-{conn_id:04} REJECT payload_len {} exceeds {} — {}",
                header.payload_len,
                cfg.max_payload_bytes,
                hex_prefix(&buf)
            ));
            break CloseReason::AbsurdPayload {
                claimed: header.payload_len,
            };
        }

        // ---- body
        let total = header.frame_len();
        match fill_to(&mut stream, &mut buf, total) {
            Ok(true) => {}
            Ok(false) => {
                break CloseReason::EofMidFrame {
                    want: total,
                    have: buf.len(),
                }
            }
            Err(e) => break classify(e),
        }

        let (frame, used) = match Frame::parse(&buf[..total]) {
            Ok(v) => v,
            Err(e) => break CloseReason::Io(e.to_string()),
        };
        buf.drain(..used);

        frame_no += 1;
        // Capture the exact bytes as received, before anything interprets them.
        // `&buf[..total]` was already consumed above, so re-encode from the
        // parsed frame — which is byte-identical by construction and covered by
        // the protocol crate's round-trip tests.
        capture.record(conn_id, Direction::Rx, &frame.encode());
        trace::log(&format!(
            "conn-{conn_id:04} RX #{frame_no} {}",
            trace::describe(&frame.header)
        ));
        tracer.frame(conn_id, "RX", &frame_no.to_string(), &frame);

        // A body that will not decode is NOT fatal: the oracle logs it and
        // dispatches with no fields, letting the RPC fall back to defaults.
        // An empty Struct is equivalent to the oracle's `None` on every path
        // dispatch takes (verified: every read is guarded or defaulted).
        let body = if frame.payload.is_empty() {
            Struct::new()
        } else {
            match heat2::decode(&frame.payload) {
                Ok(s) => s,
                Err(e) => {
                    trace::log(&format!(
                        "conn-{conn_id:04} RX #{frame_no} TDF DECODE FAILED: {e}"
                    ));
                    Struct::new()
                }
            }
        };

        let out = adapter.dispatch(&frame.header, &body, &mut session, (hooks.now)());

        for (k, resp) in out.iter().enumerate() {
            let bytes = resp.encode();
            if let Err(e) = stream.write_all(&bytes) {
                trace::log(&format!("conn-{conn_id:04} TX #{frame_no}.{k} FAILED: {e}"));
                break;
            }
            // AFTER the write: a TX record means these bytes were sent, not
            // merely intended. Capturing here also keeps the client's latency
            // untouched — it already has the bytes.
            //
            // NOT COVERED BY A TEST: moving this above the write is
            // indistinguishable while writes succeed, and only diverges when
            // one fails (it would record a frame that never reached the
            // client). Mutation-tested and confirmed undetected. The invariant
            // is held by this placement and this comment; do not move it.
            capture.record(conn_id, Direction::Tx, &bytes);
            trace::log(&format!(
                "conn-{conn_id:04} TX #{frame_no}.{k} {} ({}B total)",
                trace::describe(&resp.header),
                bytes.len()
            ));
            tracer.frame(conn_id, "TX", &format!("{frame_no}.{k}"), resp);
        }
        if let Err(e) = stream.flush() {
            break CloseReason::Io(e.to_string());
        }
    };

    trace::log(&format!(
        "conn-{conn_id:04} CLOSE after {frame_no} frame(s): {reason}"
    ));
    tracer.write_line(&format!("conn-{conn_id:04} CLOSE frames={frame_no}"));
    reason
}

/// Read until `buf` holds at least `n` bytes. `Ok(false)` on clean EOF.
fn fill_to(stream: &mut TcpStream, buf: &mut Vec<u8>, n: usize) -> io::Result<bool> {
    let mut chunk = [0u8; 65536];
    while buf.len() < n {
        match stream.read(&mut chunk) {
            Ok(0) => return Ok(false),
            Ok(got) => buf.extend_from_slice(&chunk[..got]),
            Err(ref e) if e.kind() == io::ErrorKind::Interrupted => continue,
            Err(e) => return Err(e),
        }
    }
    Ok(true)
}

fn classify(e: io::Error) -> CloseReason {
    match e.kind() {
        io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut => CloseReason::IdleTimeout,
        _ => CloseReason::Io(e.to_string()),
    }
}

fn hex_prefix(buf: &[u8]) -> String {
    buf.iter()
        .take(16)
        .map(|b| format!("{b:02x}"))
        .collect::<Vec<_>>()
        .join(" ")
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn minted_keys_have_the_blaze_shape_and_differ() {
        let a = mint_session_key();
        let b = mint_session_key();
        assert_eq!(a.len(), 16 + 1 + 44);
        assert_ne!(a, b, "a session key must be distinct per connection");
        assert!(a[..16].chars().all(|c| c.is_ascii_hexdigit()));
        assert_eq!(&a[16..17], "_");
    }

    #[test]
    fn close_reasons_render_readably() {
        assert!(CloseReason::ClientClosed.to_string().contains("closed"));
        assert!(CloseReason::EofMidFrame { want: 20, have: 5 }
            .to_string()
            .contains("want 20"));
        assert!(CloseReason::AbsurdPayload { claimed: 99 }
            .to_string()
            .contains("99"));
    }
}