//! A start refused part way returns its error with every port it bound free, and no task of //! its left running, so a start that follows binds them at once: a gateway restarting on fixed //! ports whose RTMP, RTMPS, SRT or SCTE 104 port is not yet to be had (another process's copy //! of a socket, a stop elsewhere still under way), or whose RTMPS identity does not read. Each //! check holds one of the gateway's ports, or spoils the identity, starts the gateway on all of //! them, and binds every other at once when the start returns; then a start with nothing held //! comes up on every one. The 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`) //! cannot have let its socket go until the caller yields, and one it did not end at all, an //! accept loop that waits for a shutdown that never comes, holds its port for the life of the //! process. mod common; use common::*; use ed25519_dalek::SigningKey; use joule_moq_ingest::room::{build, Descriptor}; use joule_moq_ingest::{GatewayConfig, GatewayServer}; use std::net::SocketAddr; use std::time::Duration; use tokio::time::timeout; const ROUNDS: usize = 5; const INJECTOR_SEED: [u8; 32] = [0x71; 32]; fn at(port: u16) -> SocketAddr { SocketAddr::from(([127, 0, 0, 1], port)) } /// Every listener a gateway can have, each on a fixed port of its own below the range handed /// out for port 0 (`low_tcp`, `low_udp`): HTTP, RTMP, RTMPS, SRT, and the SCTE 104 injector (its /// key, its grant from the room's authority, seed 0x51 as `World`'s). fn every_listener(c: &mut GatewayConfig) { c.http.listen = at(low_tcp()); c.rtmp.listen = Some(at(low_tcp())); c.rtmp.tls_listen = Some(at(low_tcp())); let id = joule_moq_relay::cert::RelayIdentity::dev_self_signed(&["localhost".to_string()]).unwrap(); c.rtmp.tls_der = Some((id.chain.iter().map(|c| c.to_vec()).collect(), id.key.secret_der().to_vec())); c.contrib.srt.listen = Some(at(low_udp())); let authority = SigningKey::from_bytes(&[0x51; 32]); let d = Descriptor::read(c.rooms[0].descriptor.as_ref().unwrap()).unwrap(); let key = SigningKey::from_bytes(&INJECTOR_SEED); let s = now_us() / 1_000_000; let g = build::grant(&d, &authority, &key.verifying_key().to_bytes(), 0, s - 60, s + 3600); c.injector.listen = Some(at(low_tcp())); c.injector.room = hex::encode(d.room_id()); c.injector.key_seed = Some(INJECTOR_SEED); // The principal's chain, kept across a restart of this gateway (a key given in code needs a // place for it). c.injector.state_path = Some(std::env::temp_dir().join(format!("wai-ingest-injector-ports-{}-{}.chain.json", std::process::id(), std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos()))); c.injector.grant = Some(g.to_bytes()); } /// The ports a gateway listens on, TCP (HTTP, RTMP, RTMPS, the injector's) and UDP (SRT). struct Ports { http: SocketAddr, rtmp: SocketAddr, rtmps: SocketAddr, injector: SocketAddr, srt: SocketAddr, } /// The port a check holds. #[derive(Clone, Copy, PartialEq)] enum Held { Nothing, Tcp(SocketAddr), Srt, } /// Bind each of `p`'s ports but the one `held` at once, as a start following the refused one /// does first. fn free(p: &Ports, held: Held, why: &str) { for a in [p.http, p.rtmp, p.rtmps, p.injector] { if held != Held::Tcp(a) && let Err(e) = std::net::TcpListener::bind(a) { panic!("{why}: TCP {a} is still bound when the refused start returns: {e}"); } } if held != Held::Srt && let Err(e) = std::net::UdpSocket::bind(p.srt) { panic!("{why}: UDP {} is still bound when the refused start returns: {e}", p.srt); } } async fn refused(cfg: GatewayConfig, why: &str) -> String { match GatewayServer::start(cfg).await { Ok(_) => panic!("{why}: the gateway 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 { let World { relay, gateway, .. } = World::start(every_listener).await; assert!(gateway.injector.is_some(), "the injector runs"); // The ports every start names. let p = Ports { http: gateway.addr, rtmp: gateway.rtmp_addr.unwrap(), rtmps: gateway.rtmps_addr.unwrap(), injector: gateway.injector_addr.unwrap(), srt: gateway.srt_addr.unwrap() }; let cfg = gateway.gateway.cfg.clone(); gateway.shutdown().await; for round in 0..ROUNDS { for (name, at) in [("RTMP", p.rtmp), ("RTMPS", p.rtmps), ("SCTE 104", p.injector)] { let why = format!("round {round}, the {name} port held"); let held = std::net::TcpListener::bind(at).unwrap(); let e = refused(cfg.clone(), &why).await; assert!(e.contains(&at.to_string()), "{why}: the refusal names the port: {e}"); free(&p, Held::Tcp(at), &why); drop(held); } let why = format!("round {round}, the SRT port held"); let held = std::net::UdpSocket::bind(p.srt).unwrap(); let e = refused(cfg.clone(), &why).await; assert!(e.contains(&p.srt.to_string()), "{why}: the refusal names the port: {e}"); free(&p, Held::Srt, &why); drop(held); // An RTMPS identity whose key does not read. let why = format!("round {round}, the RTMPS identity spoiled"); let mut c = cfg.clone(); if let Some((_, key)) = c.rtmp.tls_der.as_mut() { *key = vec![0x30, 0x03, 0x02, 0x01, 0x00]; } let e = refused(c, &why).await; assert!(e.contains("RTMPS identity"), "{why}: the refusal names the identity: {e}"); free(&p, Held::Nothing, &why); // Nothing held: a start on every one of the ports comes up at once. let g = GatewayServer::start(cfg.clone()).await.unwrap_or_else(|e| panic!("round {round}: the gateway starts on the ports a refused start bound: {e}")); assert_eq!((g.addr, g.rtmp_addr, g.rtmps_addr, g.injector_addr, g.srt_addr), (p.http, Some(p.rtmp), Some(p.rtmps), Some(p.injector), Some(p.srt))); g.shutdown().await; free(&p, Held::Nothing, &format!("round {round}, the shutdown")); } relay.shutdown().await; }); timeout(Duration::from_secs(300), t).await.expect("in time").unwrap(); }