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

//! Uncompressed essence into a room's tracks, end to end on loopback (WAI
//! `extensions/essence-contribution.md` §10): a publisher POSTs `/st2110` with its grant, the
//! description of a video stream, a sound stream and an ancillary data stream, and the source
//! it states; this crate's own sender sends them over UDP, from pictures and sound it
//! computes, with one video packet and one sound packet never sent. A subscriber at the relay
//! receives every group of both tracks and verifies every receipt under the gateway's key:
//! every video object is AV1 within the video floor, and the sound decodes to the samples
//! sent, bit for bit, less the packet that was lost. The session's record states each
//! stream, counts the loss and the frame not published for it, says where each capture
//! instant came from, counts the ancillary data not carried, and holds an origin claim for
//! each track; the SCTE 104 message in the ancillary data became the room's signed cues.
//! Then the refusals, a multicast group joined on loopback, and an encoder asked for that is
//! not here: the gateway does not start, or starts on the next one and says so in the
//! session's answer and its record. The tests that hold a session to its frame rate at
//! broadcast sizes are `tests/essence_rt.rs`.
//!
//! The sender here holds no PTP clock. It counts its RTP timestamps from the system clock
//! on the PTP timescale, as a sender whose media clock is direct with offset 0 would, and
//! its description says so: a fixture for the path that reads a capture instant from an RTP
//! timestamp, not a claim about this host's clock.

mod common;

use common::*;
use ed25519_dalek::{SigningKey, Verifier};
use joule_moq::draft22::{Parameters, StreamHeader, UniEvent};
use joule_moq::TrackName;
use joule_moq_ingest::catalog;
use joule_moq_ingest::frame::Codec;
use joule_moq_ingest::room::{build, media, Descriptor};
use joule_moq_ingest::st2110::sdp::{AudioFormat, Sampling};
use joule_moq_ingest::st2110::send::{self, anc_packet, audio_packet, description, scte104_anc, video_format, video_packets, Clocks, Media, Out};
use joule_moq_ingest::st2110::{flac, session};
use serde_json::{json, Value};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::net::UdpSocket;
use tokio::time::timeout;

const INJECTOR_SEED: [u8; 32] = [0x72; 32];
const LEAP: u64 = 37;

async fn world(f: impl FnOnce(&mut joule_moq_ingest::GatewayConfig)) -> World {
    // The relay's door holds the media tracks to their receipts; an events track carries none.
    let door = |rc: &mut joule_moq_relay::RelayConfig, room: &[u8; 32], gk: &str| {
        rc.receipts.door = ["audio", "video"]
            .iter()
            .map(|k| joule_moq_relay::config::DoorRule { prefix: vec![hex::encode(room), "*".into(), k.to_string()], trust: vec![gk.to_string()], trust_field: None, hold_ms: 10_000 })
            .collect();
    };
    World::start_with(door, |c| {
        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("127.0.0.1:0".parse().unwrap());
        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-st2110-{}-{}.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());
        c.st2110.enabled = true;
        // The encoder every host has: what a session does is the same on each. The
        // real-time encoders are held to their rates in `tests/essence_rt.rs`.
        c.st2110.encoder = "rav1e".into();
        c.st2110.bind_ip = "127.0.0.1".parse().unwrap();
        c.st2110.interface = "127.0.0.1".parse().unwrap();
        c.contrib.idle_ms = 1_500;
        f(c);
    })
    .await
}

async fn open(w: &World, auth: &Auth, body: &Value) -> Resp {
    auth.send("POST", &format!("{}/st2110", w.base()), &[("content-type", "application/json")], &serde_json::to_vec(body).unwrap()).await
}

/// The count of a `rate` clock on the PTP timescale at `unix_us`.
fn ticks(unix_us: u64, rate: u64) -> u32 {
    ((unix_us as u128 + LEAP as u128 * 1_000_000) * rate as u128 / 1_000_000) as u32
}

/// A `multiple_operation_message()` of one immediate splice-out of 1.5 s with auto-return
/// (SCTE 104 §7, Table 9-5), as `tests/scte104_injection.rs` builds it.
fn splice_message() -> Vec<u8> {
    let mut d = vec![2u8];
    d.extend_from_slice(&0x10u32.to_be_bytes());
    d.extend_from_slice(&[0, 9]);
    d.extend_from_slice(&0u16.to_be_bytes());
    d.extend_from_slice(&15u16.to_be_bytes());
    d.extend_from_slice(&[1, 2, 1, 0]);
    let mut o = vec![0xff, 0xff, 0, 0, 0, 3, 2, 0, 1, 0, 0, 1];
    o.extend_from_slice(&0x0101u16.to_be_bytes());
    o.extend_from_slice(&(d.len() as u16).to_be_bytes());
    o.extend_from_slice(&d);
    let len = o.len() as u16;
    o[2..4].copy_from_slice(&len.to_be_bytes());
    o
}

/// Each object of a track with the capture instant and the hop time its properties state.
type Stamps = Arc<Mutex<Vec<(u64, u64, Option<u64>, Option<u64>)>>>;

async fn watch_stamps(w: &World, kind: u8) -> (Stamps, Vec<joule_moq_relay::wire::ReqTx>) {
    let (c, ep) = relay_client(w).await;
    let c = Arc::new(c);
    let ns = catalog::media_namespace(&w.room, &w.owner(), kind);
    let (_, tx, _rx) = c.subscribe(ns, TrackName::from_str("0"), Parameters::new()).await.expect("SUBSCRIBE");
    let out: Stamps = Arc::default();
    let o2 = out.clone();
    tokio::spawn(async move {
        let _ep = ep;
        while let Some(mut r) = c.next_stream(Duration::from_secs(30)).await {
            let o2 = o2.clone();
            tokio::spawn(async move {
                let Ok(Ok(h)) = timeout(STEP, r.start()).await else { return };
                while let Ok(Ok(Some(e))) = timeout(STEP, r.next()).await {
                    if let (StreamHeader::Subgroup(sh), UniEvent::SubgroupObject(o)) = (&h, e) {
                        let one = |p: &joule_moq::draft22::kvp::Properties, ty: u64| p.0.iter().find(|k| k.ty == ty).and_then(|k| k.as_bytes()).and_then(|b| joule_moq::draft22::varint::get(b, "a timestamp").ok()).map(|v| v.0);
                        let inner = o.properties.immutable().ok().flatten().unwrap_or_default();
                        o2.lock().unwrap().push((sh.group_id, o.object_id, one(&inner, joule_moq::tempo::CAPTURE_TIMESTAMP), one(&o.properties, joule_moq::tempo::HOP_TIME)));
                    }
                }
            });
        }
    });
    (out, vec![tx])
}

/// The (group, object) of every object a watcher holds of a track.
fn w_objects(wt: &Watcher, t: &Watched) -> Vec<(u64, u64)> {
    wt.objects.lock().unwrap().keys().filter(|k| k.0 == t.media_alias).map(|k| (k.1, k.2)).collect()
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn uncompressed_essence_reaches_a_relay_subscriber_with_every_receipt() {
    timeout(Duration::from_secs(240), run()).await.expect("the test finishes in time");
}

async fn run() {
    let w = world(|_| {}).await;
    let auth = w.auth(&w.publisher, 0);
    const FRAMES: u32 = 50;
    let vf = video_format(Sampling::YCbCr422, 10, 128, 72, (25, 1));
    let af = AudioFormat { bits: 24, rate: 48_000, channels: 2, ptime_us: Some(1_000), channel_order: Some("SMPTE2110.(ST)".into()) };
    let text = description(&[Media::Video(vf.clone()), Media::Audio(af.clone()), Media::Anc(vec![(0x41, 0x07), (0x61, 0x01), (0x60, 0x60), (0x61, 0x02)])], "127.0.0.1", &[30_000, 30_002, 30_004], Clocks::Traceable, Some("SAMP"));
    let r = open(&w, &auth, &json!({"sdp": text, "source": "digitalCapture", "captions": {"lang": "en", "origin": "human"}})).await;
    assert_eq!(r.status, 201, "POST /st2110: {}", r.text());
    let s: Value = serde_json::from_slice(&r.body).unwrap();
    let id = s["session"].as_str().unwrap().to_string();
    assert_eq!(r.header("location"), Some(format!("/st2110/s/{id}").as_str()));
    assert_eq!((s["streams"][0]["clock"].as_str(), s["streams"][0]["tsmode"].as_str(), s["origin_claim"].as_bool()), (Some("ptp_traceable_stated"), Some("SAMP"), Some(true)));
    // The answer names the encoder: the one asked for, software, and no fallback.
    assert_eq!((s["encoder"]["backend"].as_str(), s["encoder"]["asked"].as_str(), s["encoder"]["hardware"].as_bool(), s["encoder"]["fallback"].is_null()), (Some("rav1e"), Some("rav1e"), Some(false), true), "{}", s["encoder"]);
    let ports: Vec<u16> = s["streams"].as_array().unwrap().iter().map(|x| x["port"].as_u64().unwrap() as u16).collect();
    assert!(contrib_live(&w, &id).await, "the session publishes its tracks");
    let watcher = watch(&w).await;
    let (stamps, _held) = watch_stamps(&w, media::VIDEO).await;
    let cat = watch_tracks(&w, vec![(0, catalog::catalog_namespace(&w.room, &w.owner()), "catalog".into())]).await;
    let injector = w.gateway.injector.clone().expect("the injector runs");

    // The sender: one socket a stream, a frame every 40 ms with its sound and its ancillary data.
    let mut socks = Vec::new();
    for p in &ports {
        let sk = UdpSocket::bind("127.0.0.1:0").await.unwrap();
        sk.connect(("127.0.0.1", *p)).await.unwrap();
        socks.push(sk);
    }
    let (mut vo, mut ao, mut no) = (Out::new(96, 0x5a5a_0001, 65_530), Out::new(97, 0x5a5a_0002, 65_530), Out::new(98, 0x5a5a_0003, 0));
    // Each frame's time code (VITC at 25 frames a second, 10:00:SS:FF), a CDP stating it, and
    // a pop-on caption in 608 data on field 1, a pair a frame, shown at frame 19 (its first
    // End of Caption) and erased at frame 30.
    let label = |n: u32| [10u8, 0, (n / 25) as u8, (n % 25) as u8];
    // CTA-708 in the CDPs (essence-contribution §8.5), the caption service directory of each
    // CDP stating service 1 in French and service 2 in Spanish, and service 3 not at all.
    // Service 2 shows a window at frame 12 and deletes it at frame 22. Services 1 and 3 each
    // show one at frame 32 and delete it at frame 44: service 1, the primary service, is
    // carried in the publisher's English (the directory's French counted as differing), and
    // service 3, with no stated language, is not carried. Each a caption channel packet of
    // a service block a service, its constructs first.
    let packet = |seq: u8, blocks: &[(u8, &[u8])]| -> Vec<[u8; 3]> {
        let mut p = vec![0u8];
        for (service, data) in blocks {
            p.push(service << 5 | data.len() as u8);
            p.extend_from_slice(data);
        }
        if p.len() % 2 == 1 {
            p.push(0);
        }
        p[0] = seq << 6 | (p.len() / 2) as u8;
        p.chunks(2).enumerate().map(|(i, c)| [if i == 0 { 0xff } else { 0xfe }, c[0], c[1]]).collect()
    };
    let window = |text: &[u8]| -> Vec<u8> { [&[0x98, 0x38, 74, 104, 0x70, 31, 0x19][..], text, &[0x03][..]].concat() };
    let shown = packet(0, &[(2, &window(b"HOLA")[..])]);
    let deleted = packet(1, &[(2, &[0x8c, 0xff][..])]);
    let primary = packet(2, &[(1, &window(b"UNO")[..]), (3, &window(b"TRES")[..])]);
    let gone = packet(3, &[(1, &[0x8c, 0xff][..]), (3, &[0x8c, 0xff][..])]);
    let dtvcc_constructs = (shown.len() + deleted.len() + primary.len() + gone.len()) as u64;
    let cdp = |n: u32| {
        let mut cc = match n {
            12 => shown.clone(),
            22 => deleted.clone(),
            32 => primary.clone(),
            44 => gone.clone(),
            _ => Vec::new(),
        };
        cc.resize(24, [0xfa, 0x00, 0x00]);
        let directory = vec![[0x81, b'f', b'r', b'e', 0xc1, 0x7f, 0xff], [0x82, b's', b'p', b'a', 0xc2, 0x7f, 0xff]];
        send::cdp(&send::CdpParts { frame_rate: 3, sequence: n as u16, time: Some((label(n), 0, false)), cc: Some(cc), active: true, svc: Some((0x50, directory)) })
    };
    // Two pop-on captions in 608: HELLO, shown at frame 19 and erased at 30, carried; BYE,
    // shown at frame 42 once service 1 has shown a caption, counted and not carried
    // (CTA-708-E §9.24), and erased at 46.
    let mut cc: Vec<[u8; 2]> = vec![[0x14, 0x20], [0x14, 0x20], [0x14, 0x2e], [0x14, 0x2e], [0x14, 0x70], [0x14, 0x70], *b"HE", *b"LL", [b'O', 0], [0x14, 0x2f], [0x14, 0x2f]];
    cc.extend([[0, 0]; 9]);
    cc.extend([[0x14, 0x2c], [0x14, 0x2c]]);
    cc.extend([[0, 0]; 2]);
    cc.extend([[0x14, 0x20], [0x14, 0x20], [0x14, 0x2e], [0x14, 0x2e], [0x14, 0x70], [0x14, 0x70], *b"BY", [b'E', 0], [0x14, 0x2f], [0x14, 0x2f]]);
    cc.extend([[0, 0]; 2]);
    cc.extend([[0x14, 0x2c], [0x14, 0x2c]]);
    let pair = |n: u32| -> [u8; 2] {
        let p = if (10..10 + cc.len() as u32).contains(&n) { cc[(n - 10) as usize] } else { [0, 0] };
        [joule_moq_ingest::st2110::captions::odd(p[0]), joule_moq_ingest::st2110::captions::odd(p[1])]
    };
    let start = now_us();
    let audio_rtp0 = ticks(start, 48_000);
    let mut pcm_sent: Vec<i32> = Vec::new();
    let started = std::time::Instant::now();
    for n in 0..FRAMES {
        let at = start + n as u64 * 40_000;
        let rtp = ticks(at, 90_000);
        for (k, p) in video_packets(&vf, &send::picture(&vf, n), rtp, &mut vo, 1_440).into_iter().enumerate() {
            // One packet of frame 7 is never sent.
            if n == 7 && k == 1 {
                continue;
            }
            socks[0].send(&p).await.unwrap();
        }
        let mut ancs = vec![send::atc(label(n), &[], [0; 8], 1, 0), cdp(n), send::cc608(1, 12, pair(n))];
        if n == 5 {
            ancs.extend(scte104_anc(&splice_message(), false));
        }
        socks[2].send(&anc_packet(&ancs, 0, rtp, true, &mut no)).await.unwrap();
        for k in 0..40u32 {
            let first = (n * 40 + k) as u64 * 48;
            let samples = send::sound(&af, first, 48);
            let p = audio_packet(&af, &samples, audio_rtp0.wrapping_add(first as u32), &mut ao);
            // One packet of sound is never sent.
            if n == 9 && k == 3 {
                continue;
            }
            pcm_sent.extend_from_slice(&samples);
            socks[1].send(&p).await.unwrap();
        }
        tokio::time::sleep_until((started + Duration::from_millis(40 * (n as u64 + 1))).into()).await;
    }

    // The session ends when nothing has come for its idle time, and signs its record.
    assert!(until(Duration::from_secs(120), || w.gateway.gateway.records().iter().any(|r| r.record.protocol == "st2110")).await, "the session's record is signed");
    assert!(until(STEP, || watcher.ended.lock().unwrap().len() == 2).await, "PUBLISH_DONE on both media tracks");
    settle(&watcher, Duration::from_millis(500)).await;
    let rec = w.gateway.gateway.records().into_iter().find(|r| r.record.protocol == "st2110").unwrap();
    rec.verify().unwrap();
    let d = &rec.record.detail;
    assert_eq!(rec.record.ended, "idle");

    // Video: every object verifies, and is AV1 within the video floor.
    let video = watcher.tracks.iter().find(|t| t.kind == media::VIDEO).unwrap();
    let audio = watcher.tracks.iter().find(|t| t.kind == media::AUDIO).unwrap();
    let (units, groups) = verified(&watcher, video, &w.gateway_key, Codec::Av1);
    assert!(groups >= 1 && !units.is_empty(), "the video track carries a group");
    let env = envelopes(&watcher, video);
    for e in &env {
        // A keyframe's object holds the sequence header: the Main profile, the Main tier, level
        // 4.1 or below, and a framing within the level. The frames after it are of its sequence.
        if 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 dim = |i: usize| u32::from_le_bytes(e.payload[4 * i..4 * i + 4].try_into().unwrap());
        assert_eq!((dim(1), dim(2)), (72, 128), "the framing states the picture's size");
        assert_eq!(e.manifest["model_requirement"]["capability"], "wai.video.av1");
        assert_eq!(e.manifest["live"]["rate"], 90_000);
    }
    assert!(env[0].key, "a group begins at a keyframe");
    let h = joule_moq_ingest::frame::av1_header(&units[0]).expect("a keyframe holds a sequence header");
    assert_eq!((h.width, h.height), (128, 72));
    let rtps: Vec<u64> = env.iter().map(|e| e.manifest["live"]["rtp"].as_u64().unwrap()).collect();
    assert!(rtps.windows(2).all(|x| x[1] > x[0] && (x[1] - x[0]).is_multiple_of(3_600)), "each object's RTP timestamp is its frame's: {rtps:?}");
    // Every frame sent is accounted for: published, not published for a missing sample, or
    // not published because the encoder was behind. Nothing is drawn for the lost packet.
    let sv = &d["streams"][0];
    let behind = sv["not_published"]["frames_encoder_behind_not_published"].as_u64().unwrap_or(0);
    assert_eq!((sv["lost"].as_u64(), sv["read"]["frames"].as_u64(), sv["read"]["frames_incomplete"].as_u64(), sv["not_published"]["frames_incomplete_not_published"].as_u64()), (Some(1), Some(FRAMES as u64), Some(1), Some(1)));
    assert_eq!(units.len() as u64 + 1 + behind, FRAMES as u64, "every frame is published or counted: {} published, {behind} behind the encoder", units.len());
    let frame7 = ticks(start + 7 * 40_000, 90_000) as u64;
    assert!(rtps.iter().all(|r| *r as u32 as u64 != frame7), "the frame with a missing row is not published");
    // The stage's own counts say the same: every whole frame was offered to it, and each
    // was encoded or counted as finding its queue full. Nothing else is dropped.
    let enc = &d["encoder"];
    let stage = &enc["stage"];
    assert_eq!((enc["backend"].as_str(), enc["asked"].as_str(), enc["fallback"].is_null()), (Some("rav1e"), Some("rav1e"), true));
    assert_eq!((stage["frames_offered"].as_u64(), stage["frames_not_encoded_queue_full"].as_u64(), stage["frames_encoded"].as_u64()), (Some(FRAMES as u64 - 1), Some(behind), Some(units.len() as u64)), "{stage}");
    assert_eq!(stage["drop_policy"], joule_moq_ingest::st2110::encode::DROP_POLICY);
    assert!(stage["keyframes"].as_u64().unwrap() >= 1 && stage["rungs"]["count"].as_u64().unwrap() >= 1);
    println!("essence_st2110: {} of {} frames of 128 x 72 at 25 a second published in this build by {} ({} found the encoder's queue full): {}", units.len(), FRAMES, enc["backend"], behind, stage);

    // Sound: FLAC objects that decode to the samples sent, bit for bit; the lost packet's
    // 48 sample frames are counted missing and none is made for them.
    let (streams, _) = verified(&watcher, audio, &w.gateway_key, Codec::Flac);
    let mut got: Vec<i32> = Vec::new();
    for st in &streams {
        let (samples, ch, bits, rate) = flac::read(st).expect("a FLAC stream of the session's writer");
        assert_eq!((ch, bits, rate), (2, 24, 48_000));
        got.extend(samples);
    }
    assert_eq!(got.len(), pcm_sent.len());
    assert!(got == pcm_sent, "the sound is the sound sent");
    let sa = &d["streams"][1];
    assert_eq!((sa["lost"].as_u64(), sa["read"]["timestamp_jumps"].as_u64(), sa["read"]["sample_frames_missing"].as_u64()), (Some(1), Some(1), Some(48)));
    let aenv = envelopes(&watcher, audio);
    let lens: Vec<u64> = aenv.windows(2).map(|x| x[1].manifest["live"]["rtp"].as_u64().unwrap() - x[0].manifest["live"]["rtp"].as_u64().unwrap()).collect();
    assert!(lens.iter().filter(|l| **l != 960).count() <= 2, "an object every 20 ms, but at the break: {lens:?}");
    assert_eq!(w.relay.relay().stats().refused_at_door, 0, "the relay's door verified every object");

    // Time: each video object states a capture instant read from its RTP timestamp, within
    // the frame's own time, and a hop time; the record says where the instants came from.
    let st = stamps.lock().unwrap().clone();
    assert_eq!(st.len(), units.len(), "the stamped subscriber saw every object");
    for (_, _, capture, hop) in &st {
        let (c, hp) = (capture.expect("a CaptureTimestamp"), hop.expect("a HopTime"));
        assert!(c + 1 >= start / 1000 && c <= (start + FRAMES as u64 * 40_000) / 1000 + 1, "a capture instant within the frames sent: {c}");
        assert!(hp >= c, "the hop time is the relay's or the gateway's, after the capture");
    }
    assert!(sv["capture_instants"]["from_rtp_timestamp"].as_u64().unwrap() >= FRAMES as u64 - 1);
    assert_eq!((sv["capture_instants"]["from_arrival"].as_u64(), sv["capture_instants"]["rtp_disagreed_with_gateway_clock"].as_u64()), (Some(0), Some(0)));
    assert_eq!(d["reference_clock"], "none held by the gateway: no lock is claimed");
    assert_eq!((sv["statement"]["clock"].as_str(), sv["statement"]["tsmode"].as_str(), sv["statement"]["sender_type_checked"].as_str()), (Some("ptp_traceable_stated"), Some("SAMP"), Some("no")));
    assert_eq!(sv["statement"]["ssrc"], 0x5a5a_0001u32);

    // The catalog states the tracks' playout timing.
    settle(&cat, Duration::from_millis(300)).await;
    let last = {
        let objs = cat.objects.lock().unwrap();
        objs.iter().rfind(|(k, _)| k.0 == cat.tracks[0].media_alias).map(|(_, b)| serde_json::from_slice::<Value>(b).unwrap()).expect("a catalog")
    };
    let wai: Vec<&Value> = last["tracks"].as_array().unwrap().iter().filter(|t| t["packaging"] == "wai").collect();
    assert_eq!(wai.len(), 2);
    assert!(wai.iter().all(|t| t["wai_tempo"] == json!({"max_delay_ms": 0, "playout_delay_ms": 0})));
    assert!(wai.iter().any(|t| t["wai_capabilities"] == json!(["wai.video.av1"])) && wai.iter().any(|t| t["wai_capabilities"] == json!(["wai.audio.flac"])));

    // Ancillary data: every packet is read. The SCTE 104 message became the room's cues,
    // signed by its injector; the time codes went onto the video objects; the CDPs were read
    // and their cc_data decoded; the first 608 caption and the windows of 708 services 2 and 1
    // each became a partial and a final revision of a caption event, signed by the same
    // principal. Service 3, which the directory gives no language, is decoded and not
    // carried; service 1's French in the directory, beside the publisher's English, is
    // counted; and the 608 caption after service 1 has shown one is not carried (CTA-708-E
    // §9.24 and §4.5.1, essence-contribution §10.6).
    assert_eq!(d["ancillary_not_carried"], json!({}));
    assert_eq!(d["scte104"], json!({"injected": 1}));
    let a = &d["time_code_and_captions"];
    assert_eq!((a["time_codes_read"].as_u64(), a["cdps_read"].as_u64(), a["cdp_constructs_without_data"].as_u64(), a["cc608_packets_field_1"].as_u64()), (Some(FRAMES as u64), Some(FRAMES as u64), Some(24 * FRAMES as u64 - dtvcc_constructs), Some(FRAMES as u64)), "{a}");
    assert_eq!((a["cdp_constructs_dtvcc"].as_u64(), a["dtvcc_packets"].as_u64()), (Some(dtvcc_constructs), Some(4)), "{a}");
    assert_eq!((a["captions_decoded"].as_u64(), a["captions_carried"].as_u64(), a["captions_finished"].as_u64()), (Some(4), Some(3), Some(3)), "{a}");
    assert_eq!((a["captions_not_carried:service_language_unstated"].as_u64(), a["captions_not_carried:cc608_beside_cta708"].as_u64(), a["caption_service_1_language_differs"].as_u64()), (Some(1), Some(1), Some(1)), "{a}");
    assert!(a.get("time_codes_discontinuous").is_none() && a.get("cdps_discontinuous").is_none(), "{a}");
    assert_eq!(d["captions_stated"], json!({"lang": "en", "origin": 1, "model": null}));
    let video = watcher.tracks.iter().find(|t| t.kind == media::VIDEO).unwrap();
    let env = envelopes(&watcher, video);
    let rtp0 = ticks(start, 90_000);
    assert!(!env.is_empty());
    for e in &env {
        let n = (e.manifest["live"]["rtp"].as_u64().unwrap() as u32).wrapping_sub(rtp0) / 3600;
        let [h, m, s, f] = label(n);
        assert_eq!(e.manifest["live"]["tc"], json!({"label": format!("{h:02}:{m:02}:{s:02}:{f:02}"), "drop": false, "interpolated": false}), "the time code of frame {n}");
    }
    assert_eq!(a["video_objects_with_time_code"].as_u64(), Some(env.len() as u64), "{a}");
    assert!(until(STEP, || injector.events().len() >= 22).await, "an out and its in, each after its injection record, and two revisions of each of three captions, each with its annotation and its window: {}", injector.events().len());
    let events = injector.events();
    let anchored: Vec<&Vec<u8>> = events.iter().filter(|e| e[117] == 1).collect();
    assert_eq!(anchored.len(), 18, "the captions' revisions, their annotations and their windows are anchored");
    let video_track = catalog::track_digest(&catalog::media_namespace(&w.room, &w.owner(), media::VIDEO), "0");
    let published: Vec<(u64, u64)> = w_objects(&watcher, video);
    for e in &anchored {
        assert_eq!(&e[118..150], &video_track[..], "anchored to the video track");
        let (g, o) = (u64::from_le_bytes(e[150..158].try_into().unwrap()), u64::from_le_bytes(e[158..166].try_into().unwrap()));
        assert!(published.contains(&(g, o)), "anchored to an object published: ({g}, {o})");
    }
    let payload = |e: &[u8]| e[169..169 + u16::from_le_bytes([e[167], e[168]]) as usize].to_vec();
    // Each caption event is followed by its annotation and its window (timed-text §4.7); the
    // 708 window's first, shown before the 608 caption and gone before it.
    assert!(anchored.chunks(3).all(|p| (p[0][166], p[1][166], p[2][166]) == (0x0a, 0x01, 0x01)));
    let of = |text: &[u8]| -> Vec<&Vec<u8>> { anchored.chunks(3).filter(|p| payload(p[0]).ends_with(text)).flat_map(|p| [p[0], p[1], p[2]]).collect() };
    let (hola, uno, anchored) = (of(b"HOLA"), of(b"UNO"), of(b"HELLO"));
    assert_eq!((hola.len(), uno.len(), anchored.len()), (6, 6, 6));
    assert!(of(b"TRES").is_empty() && of(b"BYE").is_empty(), "service 3 and the 608 caption beside service 1 are not carried");
    // A final's duration_ms is how long it is shown from its own t (timed-text §4.1): the
    // time left to the instant its caption left, the partial's t plus the media time from its
    // frame to the frame that removed it, and at least 1. So the cue it resolves to (§4.5)
    // ends where the caption left the screen, or 1 ms after the final's t where the final
    // was signed later, and is never shown again for the time it was shown.
    let t_of = |e: &[u8]| u64::from_le_bytes(e[109..117].try_into().unwrap());
    let ends_where_it_left = |c: &[&Vec<u8>], text: &[u8], lang: &[u8], shown_ms: u64, what: &str| {
        let (tp, tf) = (t_of(c[0]) / 1000, t_of(c[3]) / 1000);
        let d = u32::from_le_bytes(payload(c[3])[..4].try_into().unwrap()) as u64;
        assert_eq!(payload(c[3])[4..], [&[1, lang.len() as u8][..], lang, &(text.len() as u16).to_le_bytes(), text].concat()[..], "{what}: the final's text");
        assert!(d >= 1, "{what}: a final states a duration");
        assert_eq!(tf + d, (tp + shown_ms).max(tf + 1), "{what}: shown {shown_ms} ms from the partial's t ({tp} ms), the final signed at {tf} ms with duration_ms {d}: the cue ends where the caption left");
    };
    assert_eq!(payload(uno[0]), [&0u32.to_le_bytes()[..], &[1, 2], b"en", &3u16.to_le_bytes(), b"UNO"].concat(), "service 1 in the publisher's language, not the directory's");
    ends_where_it_left(&uno, b"UNO", b"en", 480, "shown from frame 32 to frame 44");
    assert_eq!(payload(hola[0]), [&0u32.to_le_bytes()[..], &[1, 9], b"es-x-svc2", &4u16.to_le_bytes(), b"HOLA"].concat(), "service 2 in the directory's language, its number a private-use subtag");
    ends_where_it_left(&hola, b"HOLA", b"es-x-svc2", 400, "shown from frame 12 to frame 22");
    assert_eq!(&payload(hola[4])[5..37], &joule_moq_ingest::room::sha256(&[&hola[0][..hola[0].len() - 64]])[..], "the window's final supersedes its partial");
    // The windows: CTA-708 service 2 and service 1, window 0 of each, and CTA-608 data
    // channel 1, after both revisions of each caption.
    for (c, w) in [(&hola, [2u8, 2, 0]), (&uno, [2, 1, 0]), (&anchored, [1, 1, 0])] {
        assert_eq!((payload(c[2]), payload(c[5])), ([&b"WTW1\x01"[..], &w].concat(), [&b"WTW1\x01"[..], &w].concat()), "a caption's window, timed-text §4.7");
    }
    let (partial, final_) = (payload(anchored[0]), payload(anchored[3]));
    // caption: duration_ms, origin 1 (no model), "en", the text.
    assert_eq!(partial, [&0u32.to_le_bytes()[..], &[1, 2], b"en", &5u16.to_le_bytes(), b"HELLO"].concat());
    assert_eq!(final_[4..], [&[1, 2][..], b"en", &5u16.to_le_bytes(), b"HELLO"].concat()[..]);
    ends_where_it_left(&anchored, b"HELLO", b"en", 440, "shown from frame 19 to frame 30");
    // The annotations: partial, then final superseding the partial's caption event.
    let (a0, a1) = (payload(anchored[1]), payload(anchored[4]));
    assert_eq!((&a0[..5], a0[37], &a1[..5], a1[37]), (&b"WTX1
"[..], 0, &b"WTX1
"[..], 1));
    assert_eq!(&a1[5..37], &joule_moq_ingest::room::sha256(&[&anchored[0][..anchored[0].len() - 64]])[..], "the final supersedes the partial");

    // The origin claims: one a track, signed by the gateway, of step capture, each naming
    // its stream's statement.
    let claims = d["origin_claims"].as_array().unwrap();
    assert_eq!(claims.len(), 2);
    for c in claims {
        let mut body = c.clone();
        let sig = body.as_object_mut().unwrap().remove("sig").unwrap();
        let sig = ed25519_dalek::Signature::from_slice(&hex::decode(sig.as_str().unwrap()).unwrap()).unwrap();
        // Over RFC 8785 of the claim without `sig`: its members in key order, no whitespace.
        let jcs = serde_json::to_string(&joule_moq::json_order::sorted(&body)).unwrap();
        w.gateway_key.verifying_key().verify(&[b"wai:synthesis\x01".as_slice(), jcs.as_bytes()].concat(), &sig).expect("the claim verifies under the gateway's key");
        assert_eq!((c["class"].as_str(), c["step"].as_str(), c["source"].as_str()), (Some("captured"), Some("capture"), Some("digitalCapture")));
        let i = if c["media"] == "video" { 0 } else { 1 };
        assert_eq!(c["record"], d["streams"][i]["statement_digest"]);
        assert_eq!(hex::encode(session::statement_digest(&d["streams"][i]["statement"])), d["streams"][i]["statement_digest"].as_str().unwrap());
        let kind = if i == 0 { media::VIDEO } else { media::AUDIO };
        assert_eq!(c["subject"]["track"].as_str().unwrap(), hex::encode(catalog::track_digest(&catalog::media_namespace(&w.room, &w.owner(), kind), "0")));
    }
    let first_group = stamps.lock().unwrap().iter().map(|s| s.0).min().unwrap();
    assert_eq!(claims.iter().find(|c| c["media"] == "video").unwrap()["subject"]["from_group"], first_group);

    // The session's resource is gone.
    assert_eq!(auth.send("DELETE", &format!("{}/st2110/s/{id}", w.base()), &[], b"").await.status, 404);
    w.stop().await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_description_that_is_not_carried_is_refused_with_its_reason() {
    timeout(Duration::from_secs(120), refusals()).await.expect("in time");
}

async fn refusals() {
    let w = world(|_| {}).await;
    let sdp_of = |m: &[Media], dest: &str, clocks: Clocks| description(m, dest, &[30_000, 30_002, 30_004][..m.len()], clocks, None);
    let base = video_format(Sampling::YCbCr422, 10, 128, 72, (25, 1));
    let good = sdp_of(&[Media::Video(base.clone())], "127.0.0.1", Clocks::Local);
    let reason = |r: &Resp| serde_json::from_slice::<Value>(&r.body).ok().and_then(|v| v["reason"].as_str().map(str::to_string)).unwrap_or_else(|| r.text());
    let cases: Vec<(&str, Value, u16, &str)> = vec![
        ("no description", json!({}), 400, "body_malformed"),
        ("a source that is no capture's", json!({"sdp": good, "source": "trainedAlgorithmicMedia"}), 400, "body_malformed"),
        ("interlaced video", json!({"sdp": good.replace("PM=2110GPM", "PM=2110GPM; interlace")}), 422, "sdp_scan_unsupported"),
        ("no depth", json!({"sdp": good.replace(" depth=10;", "")}), 422, "sdp_fmtp_missing"),
        ("RGB", json!({"sdp": sdp_of(&[Media::Video(video_format(Sampling::Rgb, 10, 128, 72, (25, 1)))], "127.0.0.1", Clocks::Local)}), 422, "essence_format_not_carried"),
        ("12 bits", json!({"sdp": sdp_of(&[Media::Video(video_format(Sampling::YCbCr422, 12, 128, 72, (25, 1)))], "127.0.0.1", Clocks::Local)}), 422, "essence_format_not_carried"),
        ("2160 lines at 60 a second", json!({"sdp": sdp_of(&[Media::Video(video_format(Sampling::YCbCr422, 10, 3840, 2160, (60, 1)))], "127.0.0.1", Clocks::Local)}), 422, "essence_outside_floor"),
        ("nine channels", json!({"sdp": sdp_of(&[Media::Audio(AudioFormat { bits: 24, rate: 48_000, channels: 9, ptime_us: Some(125), channel_order: None })], "127.0.0.1", Clocks::Local)}), 422, "essence_format_not_carried"),
        ("two video streams", json!({"sdp": sdp_of(&[Media::Video(base.clone()), Media::Video(base.clone())], "127.0.0.1", Clocks::Local)}), 422, "essence_streams_unsupported"),
        ("a multicast group, where the gateway joins none", json!({"sdp": sdp_of(&[Media::Video(base.clone())], "239.255.21.10", Clocks::Local)}), 403, "multicast_not_permitted"),
    ];
    for (i, (what, body, status, why)) in cases.into_iter().enumerate() {
        // A grant opens one session's request; each case takes its own.
        let who = SigningKey::from_bytes(&[0x80 + i as u8; 32]);
        let r = open(&w, &w.auth(&who, 0), &body).await;
        assert_eq!((r.status, reason(&r).as_str()), (status, why), "{what}: {}", r.text());
    }
    // A stream whose clock is not traceable is carried, and says so; a DELETE ends it, and
    // its record states the instants as arrivals.
    let auth = w.auth(&w.publisher, 0);
    let r = open(&w, &auth, &json!({"sdp": good})).await;
    assert_eq!(r.status, 201, "{}", r.text());
    let s: Value = serde_json::from_slice(&r.body).unwrap();
    assert_eq!((s["streams"][0]["clock"].as_str(), s["streams"][0]["tsmode"].as_str(), s["origin_claim"].as_bool()), (Some("not_traceable"), Some("NEW"), Some(false)));
    let id = s["session"].as_str().unwrap();
    assert!(contrib_live(&w, id).await);
    // The same principal again, while its session runs.
    let again = open(&w, &w.auth(&w.publisher, 0), &json!({"sdp": good})).await;
    assert_eq!((again.status, reason(&again).as_str()), (409, "session_conflict"));
    // One frame, from another address than the first packet's after it: only the first source's is read.
    let port = s["streams"][0]["port"].as_u64().unwrap() as u16;
    let (a, b) = (UdpSocket::bind("127.0.0.1:0").await.unwrap(), UdpSocket::bind("127.0.0.1:0").await.unwrap());
    let mut out = Out::new(96, 7, 0);
    let packets = video_packets(&base, &send::picture(&base, 0), 1_000, &mut out, 1_440);
    for p in &packets {
        a.send_to(p, ("127.0.0.1", port)).await.unwrap();
    }
    tokio::time::sleep(Duration::from_millis(100)).await;
    b.send_to(&packets[0], ("127.0.0.1", port)).await.unwrap();
    tokio::time::sleep(Duration::from_millis(300)).await;
    let loc = format!("{}/st2110/s/{id}", w.base());
    assert_eq!(auth.send("DELETE", &loc, &[], b"").await.status, 200);
    assert!(until(Duration::from_secs(60), || w.gateway.gateway.records().iter().any(|r| r.record.protocol == "st2110")).await);
    let rec = w.gateway.gateway.records().into_iter().find(|r| r.record.protocol == "st2110").unwrap();
    let sv = &rec.record.detail["streams"][0];
    assert_eq!(rec.record.ended, "deleted");
    assert_eq!((sv["statement"]["clock"].as_str(), sv["from_other_sources"].as_u64(), sv["read"]["frames"].as_u64(), sv["capture_instants"]["from_rtp_timestamp"].as_u64(), sv["capture_instants"]["from_arrival"].as_u64()), (Some("not_traceable"), Some(1), Some(1), Some(0), Some(1)));
    assert_eq!(rec.record.detail["origin_claims"], json!([]), "no source was stated, so no claim is signed");
    w.stop().await;

    // A gateway that does not take essence.
    let w = World::start(|_| {}).await;
    let r = open(&w, &w.auth(&w.publisher, 0), &json!({"sdp": good})).await;
    assert_eq!(r.status, 404);
    w.stop().await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_multicast_group_is_joined_on_loopback() {
    timeout(Duration::from_secs(60), multicast()).await.expect("in time");
}

/// As far as a loopback group allows: with the configuration's line, the description's group
/// is joined on the loopback interface at the description's own port, and the session is
/// opened; whether a datagram sent to the group comes back on this host's loopback is the
/// host's routing, which this test does not hold.
async fn multicast() {
    let w = world(|c| c.st2110.multicast = true).await;
    let port = low_udp();
    let f = video_format(Sampling::YCbCr422, 10, 128, 72, (25, 1));
    let text = description(&[Media::Video(f)], "239.255.21.10", &[port], Clocks::Local, None);
    let auth = w.auth(&w.publisher, 0);
    let r = open(&w, &auth, &json!({"sdp": text})).await;
    assert_eq!(r.status, 201, "{}", r.text());
    let s: Value = serde_json::from_slice(&r.body).unwrap();
    assert_eq!(s["streams"][0]["port"], port, "a group's stream is received at the description's port");
    let loc = format!("{}/st2110/s/{}", w.base(), s["session"].as_str().unwrap());
    assert!(contrib_live(&w, s["session"].as_str().unwrap()).await);
    assert_eq!(auth.send("DELETE", &loc, &[], b"").await.status, 200);
    w.stop().await;
}

/// An encoder asked for by name that is not on the host: the gateway does not start, and
/// says why; with `encoder_fallback` it starts on the next one, and the session's answer and
/// its record say which encoder was asked for, which encodes, and why.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn an_encoder_that_is_not_here_is_a_refusal_or_a_stated_fallback() {
    timeout(Duration::from_secs(180), absent_encoder()).await.expect("in time");
}

async fn absent_encoder() {
    // A hardware encoder through a device that does not exist, and a software encoder
    // through a library that does not: neither is on any host.
    fn absent(c: &mut joule_moq_ingest::GatewayConfig, name: &str) {
        c.st2110.enabled = true;
        c.st2110.encoder = name.into();
        c.st2110.va_device = "/dev/dri/no-such-render-node".into();
        c.st2110.svt_av1_library = "/no/such/libSvtAv1Enc.so".into();
    }
    for name in ["va", "svt-av1"] {
        let relay = joule_moq_relay::RelayServer::start(joule_moq_relay::RelayConfig::loopback()).await.unwrap();
        let mut gc = joule_moq_ingest::GatewayConfig::loopback(relay.quic_addr.unwrap());
        absent(&mut gc, name);
        let refused = joule_moq_ingest::GatewayServer::start(gc).await.err().expect("the gateway does not start on an encoder that is not here");
        assert!(refused.contains(&format!("encoder = \"{name}\" is not here")) && refused.contains("Nothing is used in its place"), "{refused}");
        relay.shutdown().await;

        let w = world(|c| {
            absent(c, name);
            c.st2110.encoder_fallback = true;
        })
        .await;
        let auth = w.auth(&w.publisher, 0);
        let f = video_format(Sampling::YCbCr422, 10, 128, 72, (25, 1));
        let text = description(&[Media::Video(f.clone())], "127.0.0.1", &[30_000], Clocks::Local, None);
        let r = open(&w, &auth, &json!({"sdp": text})).await;
        assert_eq!(r.status, 201, "{}", r.text());
        let s: Value = serde_json::from_slice(&r.body).unwrap();
        let said = |e: &Value| {
            assert_eq!((e["asked"].as_str(), e["hardware"].as_bool()), (Some(name), Some(false)), "{e}");
            assert_ne!(e["backend"].as_str(), Some(name), "it is not used in the name of the encoder asked for");
            assert!(e["fallback"].as_str().is_some_and(|x| x.starts_with(&format!("{name} was asked for and is not here: "))), "{e}");
        };
        said(&s["encoder"]);
        let id = s["session"].as_str().unwrap();
        assert!(contrib_live(&w, id).await);
        let sock = UdpSocket::bind("127.0.0.1:0").await.unwrap();
        let mut out = Out::new(96, 7, 0);
        for n in 0..3u32 {
            for p in video_packets(&f, &send::picture(&f, n), 1_000 + n * 3_600, &mut out, 1_440) {
                sock.send_to(&p, ("127.0.0.1", s["streams"][0]["port"].as_u64().unwrap() as u16)).await.unwrap();
            }
            tokio::time::sleep(Duration::from_millis(40)).await;
        }
        tokio::time::sleep(Duration::from_millis(300)).await;
        assert_eq!(auth.send("DELETE", &format!("{}/st2110/s/{id}", w.base()), &[], b"").await.status, 200);
        assert!(until(Duration::from_secs(90), || w.gateway.gateway.records().iter().any(|r| r.record.protocol == "st2110")).await);
        let rec = w.gateway.gateway.records().into_iter().find(|r| r.record.protocol == "st2110").unwrap();
        rec.verify().unwrap();
        said(&rec.record.detail["encoder"]);
        let stage = &rec.record.detail["encoder"]["stage"];
        assert_eq!(stage["frames_offered"], 3);
        assert_eq!(stage["frames_encoded"].as_u64().unwrap() + stage["frames_not_encoded_queue_full"].as_u64().unwrap(), 3, "{stage}");
        w.stop().await;
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_sender_that_bursts_whole_frames_is_read_outside_a_narrow_senders_model() {
    timeout(Duration::from_secs(120), narrow()).await.expect("in time");
}

/// Planted: a sender that sends each whole frame at once, against a description that says
/// its sender is narrow (`TP=2110TPN`, ST 2110-21 §7.1.2). The session reads its arrivals
/// against that type's Network Compatibility Model and its record counts every frame read
/// outside it. The control: the same frames on the linear schedule, four packets at an
/// instant (`st2110::pace`), whose bucket held a small part of what the burst's did. Both
/// runs' frames are read whole: the model is read, not enforced.
async fn narrow() {
    let w = world(|_| {}).await;
    // A small picture in small payloads: about fifty packets a frame, and an encoder that
    // finishes its frames in a moment on any host.
    let f = video_format(Sampling::YCbCr422, 10, 128, 72, (25, 1));
    let text = description(&[Media::Video(f.clone())], "127.0.0.1", &[30_000], Clocks::Local, None).replace("TP=2110TPW", "TP=2110TPN");
    let mut cinst = Vec::new();
    for burst in [0usize, 4] {
        let auth = w.auth(&w.publisher, 0);
        let r = open(&w, &auth, &json!({"sdp": text})).await;
        assert_eq!(r.status, 201, "{}", r.text());
        let s: Value = serde_json::from_slice(&r.body).unwrap();
        let id = s["session"].as_str().unwrap().to_string();
        assert!(contrib_live(&w, &id).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 f2 = f.clone();
        let sent = tokio::task::spawn_blocking(move || {
            let mut out = Out::new(96, 0x6e61_0001, 0);
            joule_moq_ingest::st2110::pace::Pacer::new(sock, burst)
                .run(12, (25, 1), std::time::Instant::now() + Duration::from_millis(100), |n, data, lens| {
                    for p in video_packets(&f2, &send::picture(&f2, n), 1_000 + n * 3_600, &mut out, 500) {
                        data.extend_from_slice(&p);
                        lens.push(p.len());
                    }
                })
                .unwrap()
        })
        .await
        .unwrap();
        tokio::time::sleep(Duration::from_millis(300)).await;
        assert_eq!(auth.send("DELETE", &format!("{}/st2110/s/{id}", w.base()), &[], b"").await.status, 200);
        assert!(until(Duration::from_secs(60), || w.gateway.gateway.records().iter().any(|r| r.record.session == id)).await);
        let rec = w.gateway.gateway.records().into_iter().find(|r| r.record.session == id).unwrap();
        rec.verify().unwrap();
        let sv = &rec.record.detail["streams"][0];
        let at = &sv["arrival_timing"];
        println!("burst {burst}: sent {sent:?}; arrivals {at}; receive {}", sv["receive"]);
        assert_eq!((sv["read"]["frames"].as_u64(), sv["read"]["frames_incomplete"].as_u64(), sv["lost"].as_u64()), (Some(12), Some(0), Some(0)), "every frame read whole: {sv}");
        assert_eq!((at["sender_type"].as_str(), at["cmax"].as_u64(), at["frames_read"].as_u64()), (Some("2110TPN"), Some(4), Some(11)), "{at}");
        assert_eq!(sv["receive"]["datagrams"].as_u64(), Some(sent.packets), "the receive thread read every datagram");
        if burst == 0 {
            assert_eq!(at["frames_outside"].as_u64(), Some(11), "every frame of a whole-frame burst is outside a narrow sender's model: {at}");
        }
        cinst.push(at["cinst_max"].as_u64().unwrap());
    }
    // A frame is about fifty packets: the burst's bucket held most of one.
    assert!(cinst[0] >= 30 && cinst[1] * 3 <= cinst[0], "C_INST at most: {cinst:?}");
    w.stop().await;
}