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

//! A contribution of uncompressed essence held to its frame rate, end to end on loopback
//! (WAI `extensions/essence-contribution.md` §10.3 and §12): the gateway's real session, a
//! real-time encoder behind it, this crate's own RTP sender pacing whole 4:2:2 10-bit frames
//! of a broadcast size at the stream's rate on ST 2110-21's linear schedule, four packets at an
//! instant and several a system call where the host can (`st2110::pace`), and a subscriber
//! at the relay that verifies every receipt. `tests/essence_rx.rs` measures the way in alone.
//!
//! **Slow is a failure here, not a note.** For each real-time encoder this host has
//! (`va`, `svt-av1`), the test sends 1280 × 720 and then 1920 × 1080 at 30 frames a second
//! for 20 seconds and fails unless
//!
//! - every frame sent reached the encode stage whole (none lost on the way in: that would
//!   be a measurement of the sender or the socket, and is reported as such),
//! - the stage dropped none (`frames_not_encoded_queue_full` is 0, and the session counted
//!   no frame behind the encoder),
//! - every frame was published, in order, each object verified against the gateway's
//!   receipts, and every keyframe within the video floor,
//! - and, where the host has libdav1d, every temporal unit decoded there to a picture of
//!   the stream's depth that is the picture sent (its luma within a stated distance of the
//!   source's), coded at the picture's size or at the next the engine's alignment gives,
//!   with the picture's size stated. The first frames are decoded a second time by this
//!   repository's own decoder (`wai-av1`) and compared sample for sample with the first
//!   decode: what it says is printed, and is not what the test passes or fails on.
//!
//! **A loaded host is not measured.** The test reads the host's load first and refuses to
//! measure above `WAI_RT_MAX_LOAD_PER_CORE` (default 0.15 of a core's worth a core over the
//! last minute), saying so: with `WAI_RT_REQUIRE=1` that refusal fails the test, otherwise
//! it is printed and the test ends without a result. Nothing here hides load.
//!
//! These tests are ignored unless asked for: they need a real-time encoder, 20 seconds a
//! size and a quiet host. Every figure they print is that host's, that day's.
//!
//! ```text
//! WAI_SVT_AV1_LIBRARY=/path/libSvtAv1Enc.so.4 cargo test --release -p joule-moq-ingest \
//!     --features st2110 --test essence_rt -- --ignored --nocapture --test-threads 1
//! ```
//!
//! `WAI_RT_SOURCE` names a YUV4MPEG2 file of 8-bit 4:2:0 pictures at least as large as the
//! stream's, whose centre is sent (at 10 bits, its chroma rows doubled to 4:2:2); without
//! it the pictures are computed: a panning texture with a little noise, which no encoder
//! finds easy. `WAI_RT_SECONDS` (default 20) is how long each size is sent.
//!
//! `WAI_RT_MAXUDP` (default 1460, ST 2110-10's Standard UDP Size Limit) is the largest UDP
//! datagram sent, header and RTP header included; above 1460, up to the Extended UDP Size
//! Limit of 8960 (ST 2110-10 §6.4), the description states it as `MAXUDP` (§8.6). A run at
//! the extended size sends about a sixth of the datagrams: it measures the encoder where a
//! host's loopback cannot carry the standard size's datagram rate, and says so.

mod common;

use common::dav1d::Dav1d;
use common::*;
use joule_moq_ingest::frame::Codec;
use joule_moq_ingest::room::media;
use joule_moq_ingest::st2110::encode::{self, Kind};
use joule_moq_ingest::st2110::pace::Pacer;
use joule_moq_ingest::st2110::sdp::{Sampling, VideoFormat};
use joule_moq_ingest::st2110::send::{description, video_format, video_packets, Clocks, Media, Out};
use serde_json::{json, Value};
use std::sync::Arc;
use std::time::{Duration, Instant};

/// The host's one-minute load and its logical processors.
fn load() -> Option<(f64, usize)> {
    let cores = std::thread::available_parallelism().ok()?.get();
    let text = match std::fs::read_to_string("/proc/loadavg") {
        Ok(t) => t,
        Err(_) => String::from_utf8(std::process::Command::new("sysctl").args(["-n", "vm.loadavg"]).output().ok()?.stdout).ok()?,
    };
    let one = text.split_whitespace().find_map(|t| t.parse::<f64>().ok())?;
    Some((one, cores))
}

fn env_or<T: std::str::FromStr>(name: &str, default: T) -> T {
    std::env::var(name).ok().and_then(|v| v.parse().ok()).unwrap_or(default)
}

/// `Ok` with the load when the host is quiet enough to measure; `Err` with why not.
fn quiet() -> Result<String, String> {
    let limit: f64 = env_or("WAI_RT_MAX_LOAD_PER_CORE", 0.15);
    let (one, cores) = load().ok_or("this host's load cannot be read")?;
    let said = format!("one-minute load {one:.2} on {cores} logical processors");
    if one > limit * cores as f64 { Err(format!("{said}, above {limit} a processor: not measured")) } else { Ok(said) }
}

fn host() -> String {
    let name = std::process::Command::new("hostname").output().ok().and_then(|o| String::from_utf8(o.stdout).ok()).unwrap_or_default();
    let cpu = std::fs::read_to_string("/proc/cpuinfo").ok().and_then(|c| c.lines().find(|l| l.starts_with("model name")).and_then(|l| l.split(':').nth(1)).map(|s| s.trim().to_string())).unwrap_or_else(|| std::env::consts::ARCH.to_string());
    format!("{} ({cpu}, {})", name.trim(), std::env::consts::OS)
}

/// Today's date, from the clock.
fn today() -> String {
    joule_moq_ingest::package::rfc3339(now_us())[..10].to_string()
}

/// A computed picture no encoder finds easy: a texture that pans two samples a picture,
/// with a little noise that changes every picture, in the narrow range at 10 bits.
fn computed(f: &VideoFormat, n: u32) -> [Vec<u16>; 3] {
    let (w, h, cw) = (f.width, f.height, f.width.div_ceil(2));
    let noise = |x: u32, y: u32, k: u32| {
        let mut v = x.wrapping_mul(0x9e37_79b1) ^ y.wrapping_mul(0x85eb_ca6b) ^ k.wrapping_mul(0xc2b2_ae35);
        v ^= v >> 15;
        v = v.wrapping_mul(0x2c1b_3c6d);
        v ^ (v >> 12)
    };
    let mut y = Vec::with_capacity((w * h) as usize);
    for py in 0..h {
        for px in 0..w {
            let (sx, sy) = (px + 2 * n, py + n);
            let texture = ((sx / 4) ^ (sy / 4)) & 0x7f;
            let ramp = (sx + sy) / 8 % 256;
            y.push((100 + texture * 3 + ramp + (noise(px, py, n) & 15)) as u16);
        }
    }
    let chroma = |seed: u32| -> Vec<u16> {
        let mut c = Vec::with_capacity((cw * h) as usize);
        for py in 0..h {
            for px in 0..cw {
                let (sx, sy) = (px + n, py + n);
                c.push((300 + ((sx / 16 + seed * 7) ^ (sy / 16)) % 64 * 6 + (noise(px, py, n + seed) & 7)) as u16);
            }
        }
        c
    };
    [y, chroma(1), chroma(2)]
}

/// The pictures to send: from `WAI_RT_SOURCE` where it is set, computed otherwise. With
/// what they are, for the record.
fn pictures(f: &VideoFormat, count: usize) -> (Vec<[Vec<u16>; 3]>, String) {
    let Ok(path) = std::env::var("WAI_RT_SOURCE") else {
        return ((0..count as u32).map(|n| computed(f, n)).collect(), "computed pictures (a panning texture with noise)".into());
    };
    let bytes = std::fs::read(&path).unwrap_or_else(|e| panic!("WAI_RT_SOURCE {path}: {e}"));
    let end = bytes.iter().position(|b| *b == b'\n').expect("a YUV4MPEG2 header");
    let head = String::from_utf8_lossy(&bytes[..end]).into_owned();
    assert!(head.starts_with("YUV4MPEG2 "), "{path}: not YUV4MPEG2");
    let field = |c: char| head.split(' ').find_map(|t| t.strip_prefix(c).map(str::to_string));
    let (sw, sh): (usize, usize) = (field('W').unwrap().parse().unwrap(), field('H').unwrap().parse().unwrap());
    let colour = field('C').unwrap_or_else(|| "420".into());
    assert!(colour.starts_with("420") && !colour.starts_with("420p"), "{path}: {colour}, and 8-bit 4:2:0 is read");
    let (w, h) = (f.width as usize, f.height as usize);
    assert!(sw >= w && sh >= h && sw.is_multiple_of(2) && sh.is_multiple_of(2), "{path}: {sw} x {sh} for a stream of {w} x {h}");
    let (x0, y0) = (((sw - w) / 2) & !1, ((sh - h) / 2) & !1);
    let (mut at, mut out) = (end + 1, Vec::new());
    while out.len() < count {
        let Some(nl) = bytes[at..].iter().position(|b| *b == b'\n') else { break };
        at += nl + 1;
        if at + sw * sh * 3 / 2 > bytes.len() {
            break;
        }
        let (ys, us, vs) = (&bytes[at..at + sw * sh], &bytes[at + sw * sh..at + sw * sh * 5 / 4], &bytes[at + sw * sh * 5 / 4..at + sw * sh * 3 / 2]);
        at += sw * sh * 3 / 2;
        let mut y = Vec::with_capacity(w * h);
        for r in 0..h {
            y.extend(ys[(y0 + r) * sw + x0..][..w].iter().map(|v| (*v as u16) << 2));
        }
        // Each 4:2:0 chroma row stands for two rows of 4:2:2.
        let chroma = |p: &[u8]| -> Vec<u16> {
            let mut c = Vec::with_capacity(w / 2 * h);
            for r in 0..h {
                c.extend(p[(y0 + r) / 2 * (sw / 2) + x0 / 2..][..w.div_ceil(2)].iter().map(|v| (*v as u16) << 2));
            }
            c
        };
        out.push([y, chroma(us), chroma(vs)]);
    }
    assert!(out.len() >= 2, "{path}: fewer than two pictures");
    let said = format!("{} pictures of {} ({sw} x {sh}, its centre {w} x {h}, at 10 bits with its chroma rows doubled)", out.len(), std::path::Path::new(&path).file_name().unwrap().to_string_lossy());
    (out, said)
}

/// What a paced sender did.
struct Sent {
    frames: u32,
    packets: u64,
    /// How far behind its own schedule the sender ran at worst.
    late_us: u64,
    /// The system calls that wrote, and how.
    calls: u64,
    mechanism: &'static str,
}

/// Send `frames` frames of `packed` (in order, then back, so no cut is made where they
/// repeat) at `fps` on ST 2110-21's linear schedule, four packets at an instant
/// (`st2110::pace`), with the sequence numbers and timestamps of one stream.
fn send_paced(sock: std::net::UdpSocket, packed: Arc<Vec<Vec<Vec<u8>>>>, frames: u32, fps: (u32, u32), go: Instant) -> Sent {
    let ticks = 90_000 * fps.1 / fps.0;
    let (k, mut seq) = (packed.len(), 65_000u32);
    let mut pacer = Pacer::new(sock, 4);
    let mechanism = pacer.mechanism();
    let p = pacer
        .run(frames, fps, go, |n, data, lens| {
            let at = (n as usize) % (2 * k - 2).max(1);
            for p in &packed[if at < k { at } else { 2 * k - 2 - at }] {
                let s = data.len();
                data.extend_from_slice(p);
                data[s + 2..s + 4].copy_from_slice(&(seq as u16).to_be_bytes());
                data[s + 4..s + 8].copy_from_slice(&(1_000u32.wrapping_add(n.wrapping_mul(ticks))).to_be_bytes());
                data[s + 12..s + 14].copy_from_slice(&((seq >> 16) as u16).to_be_bytes());
                seq = seq.wrapping_add(1);
                lens.push(p.len());
            }
        })
        .expect("the sender writes");
    Sent { frames: p.frames, packets: p.packets, late_us: p.late_us, calls: p.calls, mechanism }
}

/// The mean squared distance of a decoded luma plane from the luma sent, as a PSNR in
/// tenths of a decibel at the stream's depth.
fn psnr_tenths(sent: &[u16], got: &[u16], depth: u32) -> u64 {
    let sse: u64 = sent.iter().zip(got).map(|(a, b)| (*a as i64 - *b as i64).pow(2) as u64).sum();
    if sse == 0 {
        return 999;
    }
    let peak = ((1u64 << depth) - 1) as f64;
    (100.0 * (peak * peak * sent.len() as f64 / sse as f64).log10()) as u64
}

/// How many frames the second decoder is asked for: it is written for exactness, not speed.
const SECOND_DECODER_FRAMES: usize = 8;

/// The repository's own AV1 decoder (`wai/wai-av1`, written from the specification's text)
/// over the first temporal units, held sample for sample to what libdav1d made of them.
/// What it says is stated, not required: it refuses a coding tool no committed stream holds
/// it to, and is then asked again with those tools allowed, for which it claims nothing.
fn second_decoder(units: &[Vec<u8>], reference: &[Vec<u16>]) -> String {
    let run = |allow_unverified: bool| -> Result<(usize, Vec<&'static str>), wai_av1::Error> {
        let mut d = wai_av1::Decoder::new(wai_av1::Options { allow_unverified, ..wai_av1::Options::default() });
        let mut same = 0;
        for (u, want) in units.iter().zip(reference) {
            // A temporal unit begins with its delimiter, which the track's framing carries.
            let shown = d.decode_temporal_unit(&[&[0x12, 0x00][..], u].concat())?;
            let luma: Vec<u16> = shown.first().map(|p| (0..p.planes[0].height).flat_map(|y| p.planes[0].data[y * p.planes[0].stride..][..p.planes[0].width].to_vec()).collect()).unwrap_or_default();
            same += (shown.len() == 1 && luma == *want) as usize;
        }
        Ok((same, d.tools_used().iter().copied().filter(|t| !wai_av1::frame::VERIFIED_TOOLS.contains(t)).collect()))
    };
    let n = units.len();
    match run(false) {
        Ok((same, _)) if same == n => format!("wai-av1 decoded the first {n} to the same luma, sample for sample"),
        Ok((same, _)) => format!("wai-av1 DISAGREES with libdav1d on {} of the first {n} frames", n - same),
        Err(e) if e.code == "refused_unverified_tool" => match run(true) {
            Ok((same, tools)) if same == n => format!("wai-av1 refuses the stream by default ({}) and, with the tools it holds to no reference allowed ({}), decoded the first {n} to the same luma, sample for sample", e.detail, tools.join(", ")),
            Ok((same, tools)) => format!("wai-av1 refuses the stream by default ({}); with its unverified tools allowed ({}) it DISAGREES with libdav1d on {} of the first {n} frames", e.detail, tools.join(", "), n - same),
            Err(e2) => format!("wai-av1 refuses the stream by default ({}), and with its unverified tools allowed: {}: {}", e.detail, e2.code, e2.detail),
        },
        Err(e) => format!("wai-av1 does not decode the stream: {}: {}", e.code, e.detail),
    }
}

/// One measurement: a session of `w` × `h` at `fps` through `kind` for `seconds`. `Ok`
/// with the line to record when the rate was held; `Err` with what was not.
async fn hold(kind: Kind, w: u32, h: u32, fps: (u32, u32), seconds: u32) -> Result<String, String> {
    let before = quiet()?;
    let frames = seconds * fps.0 / fps.1;
    let f = video_format(Sampling::YCbCr422, 10, w, h, fps);
    // The pictures, packed once before anything is timed.
    // The datagram size: the UDP header (8 octets) and the RTP header (12) are in it.
    let maxudp: usize = env_or("WAI_RT_MAXUDP", 1_460usize).clamp(1_460, 8_960);
    let (pics, source) = tokio::task::spawn_blocking({
        let f = f.clone();
        move || pictures(&f, 48)
    })
    .await
    .unwrap();
    let packed: Arc<Vec<Vec<Vec<u8>>>> = Arc::new(
        tokio::task::spawn_blocking({
            let (f, pics) = (f.clone(), pics.clone());
            move || pics.iter().map(|p| video_packets(&f, p, 0, &mut Out::new(96, 0x7274_0001, 0), maxudp - 20)).collect()
        })
        .await
        .unwrap(),
    );
    let world = World::start(|c| {
        c.st2110.enabled = true;
        c.st2110.bind_ip = "127.0.0.1".parse().unwrap();
        c.st2110.encoder = kind.name().into();
        c.st2110.queue_packets = 65_536;
        encode::from_env(&mut c.st2110);
        c.contrib.idle_ms = 1_500;
        // A frame of 1920 x 1080 may be large at a keyframe; the session's own limits stand.
    })
    .await;
    let chosen = world.gateway.gateway.essence_encoder().map_err(|e| format!("no encoder: {e}"))?;
    let auth = world.auth(&world.publisher, 0);
    let mut text = description(&[Media::Video(f.clone())], "127.0.0.1", &[30_000], Clocks::Local, None);
    if maxudp > 1_460 {
        text = text.replace("; TP=2110TPW", &format!("; TP=2110TPW; MAXUDP={maxudp}"));
    }
    let r = auth.send("POST", &format!("{}/st2110", world.base()), &[("content-type", "application/json")], &serde_json::to_vec(&json!({"sdp": text})).unwrap()).await;
    if r.status != 201 {
        return Err(format!("POST /st2110: {} {}", r.status, r.text()));
    }
    let s: Value = serde_json::from_slice(&r.body).unwrap();
    assert_eq!((s["encoder"]["backend"].as_str(), s["encoder"]["fallback"].is_null()), (Some(kind.name()), true), "the encoder asked for encodes, under its own name");
    let id = s["session"].as_str().unwrap().to_string();
    assert!(contrib_live(&world, &id).await, "the session publishes its tracks");
    let watcher = watch_kinds(&world, &[media::VIDEO]).await;
    let sock = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
    sock.connect(("127.0.0.1", s["streams"][0]["port"].as_u64().unwrap() as u16)).unwrap();
    let go = Instant::now() + Duration::from_millis(200);
    let sent = tokio::task::spawn_blocking({
        let packed = packed.clone();
        move || send_paced(sock, packed, frames, fps, go)
    })
    .await
    .unwrap();
    let during = load().map(|(l, _)| l).unwrap_or(0.0);
    if !until(Duration::from_secs(60), || world.gateway.gateway.records().iter().any(|r| r.record.protocol == "st2110")).await {
        return Err("the session signed no record".into());
    }
    let rec = world.gateway.gateway.records().into_iter().find(|r| r.record.protocol == "st2110").unwrap();
    rec.verify().unwrap();
    assert!(until(STEP, || watcher.ended.lock().unwrap().len() == 1).await, "PUBLISH_DONE on the video track");
    settle(&watcher, Duration::from_millis(500)).await;
    let d = &rec.record.detail;
    let (sv, stage) = (&d["streams"][0], &d["encoder"]["stage"]);
    let n = |v: &Value| v.as_u64().unwrap_or(0);
    let size = if maxudp > 1_460 { format!("datagrams of at most {maxudp} octets, the extended UDP size limit's, stated as MAXUDP") } else { format!("datagrams of at most {maxudp} octets") };
    let what = format!("{} {} ({}), {w} x {h} at {}/{} for {seconds} s, {} frames in {} {size}, {source}", kind.name(), chosen.identity.version, stage["settings"].as_str().unwrap_or(""), fps.0, fps.1, sent.frames, sent.packets);

    // The way in: every packet arrived and every frame reached the stage whole. Otherwise
    // this run measured the sender or the socket, not the encoder.
    let whole = n(&sv["read"]["frames"]) == frames as u64 && n(&sv["read"]["frames_incomplete"]) == 0 && n(&sv["lost"]) == 0 && n(&d["packets_dropped_queue_full"]) == 0;
    // How the way in went: the sender's writes, the receive thread's reads, the arrivals
    // against the description's sender type.
    let (rx, at) = (&sv["receive"], &sv["arrival_timing"]);
    let way_in = format!(
        "sent by {} in {} calls, at worst {} ms behind its schedule; read by {} in {} reads ({} coalesced buffers), the receive thread {} ms of processor time, {} buffers dropped by the kernel at a full receive buffer of {} octets ({} ms of the stream); arrivals against a {} sender's model: C_MAX {}, C_INST at most {}, {} of {} frames outside",
        sent.mechanism,
        sent.calls,
        sent.late_us / 1000,
        rx["mechanism"].as_str().unwrap_or("?"),
        rx["reads"],
        rx["coalesced_buffers"],
        n(&rx["thread_cpu_us"]) / 1000,
        rx["kernel_dropped_buffers"].as_u64().map_or("(a count the host does not give)".to_string(), |v| v.to_string()),
        n(&sv["receive_buffer_octets"]),
        n(&rx["buffer_us"]) / 1000,
        at["sender_type"].as_str().unwrap_or("?"),
        at["cmax"],
        at["cinst_max"],
        at["frames_outside"],
        at["frames_read"],
    );
    if !whole || sent.late_us > 20_000 {
        return Err(format!(
            "{what}: the way in did not hold, so the encoder was not measured: {} of {} packets counted, {} lost, {} frames read, {} incomplete, {} datagrams found the session's queue full; {way_in}",
            n(&sv["packets"]),
            sent.packets,
            n(&sv["lost"]),
            n(&sv["read"]["frames"]),
            n(&sv["read"]["frames_incomplete"]),
            n(&d["packets_dropped_queue_full"]),
        ));
    }

    // The encoder: every frame offered, none dropped, every one published and verified.
    let video = watcher.tracks.iter().find(|t| t.kind == media::VIDEO).unwrap();
    let (units, groups) = verified(&watcher, video, &world.gateway_key, Codec::Av1);
    let env = envelopes(&watcher, video);
    let dropped = n(&stage["frames_not_encoded_queue_full"]);
    let behind = n(&sv["not_published"]["frames_encoder_behind_not_published"]);
    let figures = format!(
        "{} frames encoded of {} offered, {dropped} dropped behind the encoder, {} published in {groups} groups; a frame took {:.1} ms on average and {:.1} at most of {:.1}; {} frames over their interval; queue {} deep at most of {}; rungs {} to {} of {} ({} moves faster, {} slower); {} kbit/s made",
        n(&stage["frames_encoded"]),
        n(&stage["frames_offered"]),
        units.len(),
        n(&stage["us_a_frame"]["encode_mean"]) as f64 / 1000.0,
        n(&stage["us_a_frame"]["encode_max"]) as f64 / 1000.0,
        n(&stage["frame_interval_us"]) as f64 / 1000.0,
        n(&stage["frames_over_their_interval"]),
        n(&stage["queue_depth_max"]),
        n(&stage["queue"]),
        n(&stage["rungs"]["first"]),
        n(&stage["rungs"]["last"]),
        n(&stage["rungs"]["count"]),
        n(&stage["rungs"]["moved_faster"]),
        n(&stage["rungs"]["moved_slower"]),
        n(&stage["octets"]) * 8 * fps.0 as u64 / (fps.1 as u64 * frames as u64 * 1000),
    );
    if n(&stage["frames_offered"]) != frames as u64 || dropped != 0 || behind != 0 || n(&stage["frames_encoded"]) != frames as u64 || units.len() != frames as usize {
        return Err(format!("{what}: NOT HELD: {figures}; ended {}", rec.record.ended));
    }
    if rec.record.ended != "idle" {
        return Err(format!("{what}: the session ended {}", rec.record.ended));
    }

    // The floor, the order and the times: as every essence session's.
    assert!(env[0].key, "a group begins at a keyframe");
    for e in env.iter().filter(|e| e.key) {
        joule_moq_ingest::video_floor::check_payload(&e.payload).unwrap_or_else(|o| panic!("an object outside the video floor: {}", o.detail()));
    }
    let seq = joule_moq_ingest::frame::av1_obus(&units[0]).unwrap().into_iter().find(|o| o.0 == 1).map(|o| joule_moq_ingest::video_floor::read_sequence_header(o.2).unwrap()).expect("a sequence header");
    assert_eq!(seq.seq_profile, 0);
    assert!(seq.operating_points.iter().all(|(level, tier)| *level <= 9 && *tier == 0), "{:?}", seq.operating_points);
    // The picture every keyframe states is the stream's. The frames are coded at that size,
    // or at the next the engine's block alignment gives, and the record says which.
    let sizes = encode::check_unit(&units[0], w, h, fps.0, fps.1).expect("a keyframe within the floor");
    let (cw, ch) = sizes.coded;
    assert_eq!(sizes.render, (w, h));
    assert_eq!(stage["coded_frame"], json!([cw, ch]));
    assert_eq!(stage["picture"], json!([w, h]));
    for (e, u) in env.iter().zip(&units).filter(|(e, _)| e.key) {
        assert_eq!(encode::check_unit(u, w, h, fps.0, fps.1), Ok(sizes), "every keyframe");
        let dim = |i: usize| u32::from_le_bytes(e.payload[4 * i..4 * i + 4].try_into().unwrap());
        assert_eq!((dim(1), dim(2)), (ch, cw), "the framing states the size the frames are coded at");
    }
    let rtps: Vec<u64> = env.iter().map(|e| e.manifest["live"]["rtp"].as_u64().unwrap()).collect();
    let ticks = 90_000 * fps.1 as u64 / fps.0 as u64;
    assert!(rtps.windows(2).all(|x| x[1] == x[0] + ticks), "every frame is published, in order, with its own timestamp");
    assert_eq!(world.relay.relay().stats().refused_at_door, 0, "the relay's door verified every object");

    // The oracle: every temporal unit decodes with libdav1d to the picture sent.
    let oracle = match Dav1d::open() {
        Ok(mut dec) => {
            let (mut worst, mut first) = (999u64, 0u64);
            // The first frames as libdav1d decoded them, for the second decoder below.
            let mut reference: Vec<Vec<u16>> = Vec::new();
            for (i, u) in units.iter().enumerate() {
                let shown = dec.decode(u).unwrap_or_else(|e| panic!("libdav1d refuses temporal unit {i}: {e}"));
                assert_eq!(shown.len(), 1, "temporal unit {i} shows one picture");
                let p = &shown[0];
                assert_eq!((p.w, p.h, p.bpc, p.layout), (cw, ch, 10, common::dav1d::LAYOUT_I420), "temporal unit {i}");
                if reference.len() < SECOND_DECODER_FRAMES {
                    reference.push(p.luma.clone());
                }
                let at = i % (2 * pics.len() - 2);
                let src = &pics[if at < pics.len() { at } else { 2 * pics.len() - 2 - at }];
                // The picture is the frame's first `h` rows of `w` samples.
                let picture: Vec<u16> = p.luma.chunks_exact(cw as usize).take(h as usize).flat_map(|row| row[..w as usize].iter().copied()).collect();
                let q = psnr_tenths(&src[0], &picture, 10);
                worst = worst.min(q);
                if i == 0 {
                    first = q;
                }
            }
            // A picture of the wrong frame, or one with its planes or its samples misplaced,
            // lies below 15 dB of the picture sent; an encode of it at these rates far above.
            assert!(worst >= 200, "a decoded picture is not the picture sent: {:.1} dB at worst", worst as f64 / 10.0);
            let second = tokio::task::spawn_blocking({
                let units: Vec<Vec<u8>> = units.iter().take(SECOND_DECODER_FRAMES).cloned().collect();
                move || second_decoder(&units, &reference)
            })
            .await
            .unwrap();
            format!("libdav1d {} decoded all {} temporal units to {cw} x {ch} 4:2:0 at 10 bits (the picture {w} x {h}), luma PSNR {:.1} dB for the first and {:.1} dB at worst; {second}", dec.version, units.len(), first as f64 / 10.0, worst as f64 / 10.0)
        }
        Err(why) => {
            assert!(std::env::var_os("WAI_RT_REQUIRE_DAV1D").is_none(), "{why}");
            format!("not decoded here: {why}")
        }
    };
    world.stop().await;
    let after = load().map(|(l, _)| l).unwrap_or(0.0);
    Ok(format!("HELD {} on {}: {what}: {figures}; {oracle}; the way in: {way_in}; {before} before, {during:.2} as the sender ended, {after:.2} after", today(), host()))
}

/// The real-time encoders this host has, by their probes.
fn here() -> Vec<(Kind, Result<String, String>)> {
    let mut c = joule_moq_ingest::config::St2110Config { calibrate_ms: 0, ..Default::default() };
    encode::from_env(&mut c);
    [Kind::Va, Kind::SvtAv1].into_iter().map(|k| (k, encode::probe(k, &c).map(|i| format!("{} {}", i.version, i.detail)))).collect()
}

async fn sizes(list: &[(u32, u32, (u32, u32))]) {
    let seconds: u32 = env_or("WAI_RT_SECONDS", 20);
    let require = std::env::var_os("WAI_RT_REQUIRE").is_some();
    let only = std::env::var("WAI_RT_ENCODER").ok();
    let (mut ran, mut failed) = (0, Vec::new());
    for (kind, probe) in tokio::task::spawn_blocking(here).await.unwrap() {
        if only.as_deref().is_some_and(|o| o != kind.name()) {
            continue;
        }
        match probe {
            Err(why) => println!("essence_rt: {} is not on this host: {why}", kind.name()),
            Ok(id) => {
                println!("essence_rt: {} is here: {id}", kind.name());
                for (w, h, fps) in list {
                    ran += 1;
                    match hold(kind, *w, *h, *fps, seconds).await {
                        Ok(line) => println!("essence_rt: {line}"),
                        Err(why) if why.ends_with("not measured") => {
                            println!("essence_rt: REFUSED: {} {w} x {h}: {why}", kind.name());
                            if require {
                                failed.push(why);
                            }
                        }
                        Err(why) => {
                            println!("essence_rt: FAILED: {why}");
                            failed.push(why);
                        }
                    }
                }
            }
        }
    }
    assert!(ran > 0 || !require, "no real-time encoder is on this host: only the fallback, which holds no broadcast size");
    assert!(failed.is_empty(), "{} of {ran} not held:\n{}", failed.len(), failed.join("\n"));
}

/// 1280 × 720 and 1920 × 1080 at 30 frames a second, 20 seconds each, with every real-time
/// encoder this host has: no frame dropped behind the encoder, or the test fails.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore = "20 seconds a size on a quiet host with a real-time encoder: run on request"]
async fn each_real_time_encoder_here_holds_hd_at_30_frames_a_second() {
    sizes(&[(1280, 720, (30, 1)), (1920, 1080, (30, 1))]).await;
}

/// 1920 × 1080 at 60 frames a second, the top of level 4.1: measured and held to the
/// same, where it is asked for.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore = "a measurement at the top of the floor: run on request"]
async fn hd_at_60_frames_a_second_as_measured() {
    sizes(&[(1920, 1080, (60, 1))]).await;
}

/// The fallback encoder at a broadcast size, as measured: it is not held to the rate (it
/// does not hold it on the hosts measured), but to the count: every frame it did not
/// encode is counted, and the rest are published.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore = "a measurement of the fallback: run on request"]
async fn the_fallback_at_hd_drops_and_counts() {
    let seconds: u32 = env_or("WAI_RT_SECONDS", 10);
    match hold(Kind::Rav1e, 1280, 720, (30, 1), seconds).await {
        Ok(line) => println!("essence_rt: the fallback held it: {line}"),
        Err(why) if why.contains("NOT HELD") => println!("essence_rt: the fallback, as measured on {} on {}: {why}", host(), today()),
        Err(why) => println!("essence_rt: the fallback was not measured: {why}"),
    }
}

/// 3840 × 2160 is past level 4.1's picture size at any rate: a session of it is refused
/// before any encoder is asked, so no encoder here is measured at it.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn ultra_high_definition_is_outside_the_floor_at_any_rate() {
    let world = World::start(|c| {
        c.st2110.enabled = true;
        c.st2110.bind_ip = "127.0.0.1".parse().unwrap();
    })
    .await;
    for (i, fps) in [(30u32, 1u32), (25, 1), (60, 1)].into_iter().enumerate() {
        let who = ed25519_dalek::SigningKey::from_bytes(&[0x90 + i as u8; 32]);
        let text = description(&[Media::Video(video_format(Sampling::YCbCr422, 10, 3840, 2160, fps))], "127.0.0.1", &[30_000], Clocks::Local, None);
        let r = world.auth(&who, 0).send("POST", &format!("{}/st2110", world.base()), &[("content-type", "application/json")], &serde_json::to_vec(&json!({"sdp": text})).unwrap()).await;
        let v: Value = serde_json::from_slice(&r.body).unwrap_or_default();
        assert_eq!((r.status, v["reason"].as_str()), (422, Some("essence_outside_floor")), "{}", r.text());
    }
    world.stop().await;
}