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

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