//! 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(); }