open-standards/joule-code/joule-code-rs/crates/joule-moq-relay/tests/start_ports.rs

//! A start refused part way returns its error with every port it bound free, so a start that
//! follows binds them at once: a relay restarting on fixed ports (a rollover) whose
//! WebSocket port, control port or control socket is not yet to be had. The relay binds its
//! UDP endpoint, then its WebSocket listener, then its control surface; each check holds one of
//! those, starts the relay on all of them, and binds every other at once when the start
//! returns. Each test runs in a task on a runtime of one worker, as a restart from a spawned
//! task does: there a task a refused start only told to go (`JoinHandle::abort`), or a QUIC
//! endpoint whose driver has yet to see its last handle go, cannot have let its socket go
//! until the caller yields, and one it did not end at all holds its port for the life of the
//! process. The control surface is held alone as well: refused at its socket, it closes the TCP
//! listener it had bound.

use joule_moq_relay::{RelayConfig, RelayServer};
use std::net::SocketAddr;
use std::time::Duration;
use tokio::time::timeout;

const ROUNDS: usize = 10;

/// Bind each address at once, as a start following the refused one does first.
fn free(tcp: &[SocketAddr], udp: &[SocketAddr], why: &str) {
    for &a in tcp {
        if let Err(e) = std::net::TcpListener::bind(a) {
            panic!("{why}: TCP {a} is still bound when the refused start returns: {e}");
        }
    }
    for &a in udp {
        if let Err(e) = std::net::UdpSocket::bind(a) {
            panic!("{why}: UDP {a} is still bound when the refused start returns: {e}");
        }
    }
}

/// A port below the range a system hands out for port 0 (from 32768 on Linux, 49152 on
/// macOS) that `free` finds free now. A port handed out for port 0 can go to any socket on the
/// host the moment it is let go, so a test that lets a port go and checks at once that it is
/// free, or binds it again, takes one of these instead. Each call starts its search at a place
/// of its own, so tests of one binary running at once are not handed the same port. Even ports
/// only, so a pair (P, P + 1) can be asked for.
fn low_port(free: impl Fn(u16) -> bool) -> u16 {
    static CALLS: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0);
    let first = std::process::id() % 2_000 + 64 * CALLS.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
    (0..6_000u32)
        .map(|i| (20_000 + 2 * ((first + i) % 6_000)) as u16)
        .find(|&p| free(p))
        .expect("a free port below the range handed out for port 0")
}

fn low_tcp() -> SocketAddr {
    SocketAddr::from(([127, 0, 0, 1], low_port(|p| std::net::TcpListener::bind(("127.0.0.1", p)).is_ok())))
}

fn low_udp() -> SocketAddr {
    SocketAddr::from(([127, 0, 0, 1], low_port(|p| std::net::UdpSocket::bind(("127.0.0.1", p)).is_ok())))
}

async fn refused(cfg: RelayConfig, why: &str) -> String {
    match RelayServer::start(cfg).await {
        Ok(_) => panic!("{why}: the relay starts"),
        Err(e) => e,
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
async fn a_start_refused_part_way_frees_every_port_it_bound() {
    let t = tokio::spawn(async {
        // Fixed ports below the range handed out for port 0, for every start.
        let mut cfg = RelayConfig::loopback();
        let (quic, ws, control) = (low_udp(), low_tcp(), low_tcp());
        (cfg.quic_listen, cfg.websocket_listen, cfg.control.listen) = (Some(quic), Some(ws), Some(control));
        let dir = std::env::temp_dir().join(format!("relay-start-ports-{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        for round in 0..ROUNDS {
            // The WebSocket port held: the UDP endpoint was bound before it.
            let held = std::net::TcpListener::bind(ws).unwrap();
            let e = refused(cfg.clone(), "the WebSocket port held").await;
            assert!(e.contains(&ws.to_string()), "round {round}: the refusal names the WebSocket port: {e}");
            free(&[control], &[quic], &format!("round {round}, the WebSocket port held"));
            drop(held);
            // The control port held: the UDP endpoint and the WebSocket listener were bound
            // before it.
            let held = std::net::TcpListener::bind(control).unwrap();
            let e = refused(cfg.clone(), "the control port held").await;
            assert!(e.contains(&control.to_string()), "round {round}: the refusal names the control port: {e}");
            free(&[ws], &[quic], &format!("round {round}, the control port held"));
            drop(held);
            // A file that is not a socket where the control socket goes: the control surface's
            // TCP listener was bound before it, and the file is left as it was.
            #[cfg(unix)]
            {
                let path = dir.join("control.sock");
                std::fs::write(&path, b"not a socket").unwrap();
                let mut c = cfg.clone();
                c.control.socket = Some(path.clone());
                let e = refused(c, "a file in the control socket's place").await;
                assert!(e.contains("not a socket"), "round {round}: the refusal names the control socket: {e}");
                free(&[ws, control], &[quic], &format!("round {round}, a file in the control socket's place"));
                assert_eq!(std::fs::read(&path).unwrap(), b"not a socket", "round {round}: the file is left as it was");
                std::fs::remove_file(&path).unwrap();
            }
            // Nothing held: a start on every one of the ports comes up at once.
            let server = RelayServer::start(cfg.clone()).await.unwrap_or_else(|e| panic!("round {round}: the relay starts on the ports a refused start bound: {e}"));
            assert_eq!((server.quic_addr, server.websocket_addr, server.control_addr), (Some(quic), Some(ws), Some(control)));
            server.shutdown().await;
            free(&[ws, control], &[quic], &format!("round {round}, the shutdown"));
        }
        let _ = std::fs::remove_dir_all(&dir);
    });
    timeout(Duration::from_secs(300), t).await.expect("in time").unwrap();
}

/// The control surface alone, refused part way (a file that is not a socket where its socket
/// goes): the TCP listener it bound first is closed when the refusal returns, its task ended, not
/// only told to go.
#[cfg(unix)]
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
async fn a_control_surface_refused_part_way_frees_its_port() {
    let t = tokio::spawn(async {
        let dir = std::env::temp_dir().join(format!("relay-control-ports-{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        let path = dir.join("control.sock");
        std::fs::write(&path, b"not a socket").unwrap();
        let at = low_tcp();
        let mut cfg = RelayConfig::loopback();
        cfg.control.listen = Some(at);
        cfg.control.socket = Some(path.clone());
        let relay = joule_moq_relay::Relay::new(cfg).unwrap();
        for round in 0..ROUNDS * 2 {
            match joule_moq_relay::control::Control::start(&relay).await {
                Ok(_) => panic!("round {round}: the control surface starts"),
                Err(e) => assert!(e.contains("not a socket"), "round {round}: the refusal names the control socket: {e}"),
            }
            free(&[at], &[], &format!("round {round}, the control surface refused"));
        }
        assert_eq!(std::fs::read(&path).unwrap(), b"not a socket", "the file is left as it was");
        let _ = std::fs::remove_dir_all(&dir);
    });
    timeout(Duration::from_secs(120), t).await.expect("in time").unwrap();
}