//! The relay's own monitoring (`src/monitoring.rs`; WAI moq-streaming-format §4.8,
//! delivery-telemetry §11), on loopback:
//!
//! - **the metrics track** is `GET /metrics` at one instant: a subscriber reads the catalog
//!   and data groups off the relay, and each group is held, number for number, to the text
//!   the control listener answers before the next capture. Planted: a counter one above its
//!   exposition's (`value_disagrees`), and a series the exposition does not have
//!   (`series_disagree`);
//! - **the log track** carries what the relay logged, a group a second, with no address in
//!   it. Planted: a message with a peer's address in an attribute, refused
//!   `address_in_attribute` and never published;
//! - **qlog** (feature `qlog`): each session's file reads as JSON Text Sequences, holds to
//!   the draft's event definitions and to its order, and holds no payload byte. Planted: an
//!   object's record moved before its stream's header (`object_before_header`,
//!   `time_regress`). Without the feature a relay asked for a qlog does not start;
//! - **a hop Feedback Track** is consumed: its stalls are written onward as observations in
//!   a session of the relay's own, the tally stays with the relay. Planted: an Optional
//!   Metric with no mapping (`metric_unmapped`, nothing written from that report), a tally
//!   above what the relay sent (`more_than_sent`, the track ended), a PUBLISH from a session
//!   that declared nothing (`not_negotiated`), one for a track the session does not
//!   subscribe to (`feedback_not_subscribed`), and one whose property contradicts its name
//!   (`type_mismatch`).
//!
//! With `[monitoring]` absent the relay serves none of it.

mod common;

use common::*;
use joule_moq::draft22::data::{SubgroupHeader, SubgroupId, SubgroupObject};
use joule_moq::draft22::kvp::Properties;
use joule_moq::draft22::{ControlMessage, Parameters, Publish, RequestErrorCode, TrackRequest};
use joule_moq::feedback::{self, Capabilities, FeedbackType, ObjectEntry, Report, Status, Summary};
use joule_moq::log_track;
use joule_moq::metrics_track::{self, Catalog, Level, Number, TrackObject};
use joule_moq::{TrackName, TrackNamespace};
use joule_moq_relay::client::{Client, ClientOptions};
use joule_moq_relay::monitoring;
use joule_moq_relay::RelayServer;
use std::collections::BTreeMap;
use std::net::SocketAddr;
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::time::timeout;

const LIMIT: Duration = Duration::from_secs(120);

fn resource(server: &RelayServer) -> u64 {
    metrics_track::resource_id(&server.relay().relay_id())
}

/// A request's first answer, where it is a refusal: its code and reason.
async fn refusal(c: &Client, m: ControlMessage) -> (RequestErrorCode, String) {
    let (_tx, mut rx) = c.request(&m).await.expect("the request stream");
    match timeout(STEP, Client::response(&mut rx)).await.expect("an answer in time").expect("a response") {
        ControlMessage::RequestError(e) => (e.code, String::from_utf8_lossy(&e.reason).into_owned()),
        other => panic!("not refused: {}", other.name()),
    }
}

fn subscribe(c: &Client, ns: TrackNamespace, name: &str) -> ControlMessage {
    ControlMessage::Subscribe(TrackRequest { request_id: c.next_request_id(), namespace: ns, name: TrackName::from_str(name), params: Parameters::new() })
}

fn objects(g: &Got) -> Vec<TrackObject> {
    g.objects.iter().map(TrackObject::from_subgroup_object).collect()
}

/// `GET /metrics` on the control listener: the text.
async fn scrape(addr: SocketAddr) -> String {
    let mut s = tokio::net::TcpStream::connect(addr).await.expect("the control listener");
    s.write_all(b"GET /metrics HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\n\r\n").await.unwrap();
    let mut out = Vec::new();
    timeout(STEP, s.read_to_end(&mut out)).await.expect("the connection ends").expect("the answer");
    let said = String::from_utf8(out).expect("UTF-8");
    let (head, body) = said.split_once("\r\n\r\n").expect("an HTTP answer");
    assert!(head.starts_with("HTTP/1.1 200 OK\r\n"), "{head}");
    body.to_string()
}

const CAPTURES: &str = "wai_relay_monitoring_captures_total";

fn captures_of(samples: &[metrics_track::Sample]) -> Number {
    samples.iter().find(|s| s.name == CAPTURES).expect("the captures counter").value
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn the_metrics_track_is_metrics_at_one_instant() {
    timeout(LIMIT, async {
        let server = relay(|c| {
            c.control.listen = Some(loopback());
            c.metering.energy = joule_moq_relay::config::EnergyMode::Off;
            c.monitoring.metrics_track = true;
            c.monitoring.metrics_interval_ms = 400;
        })
        .await;
        let control = server.control_addr.expect("the loopback listener");
        let ns = metrics_track::namespace(resource(&server));
        let (c, _ep) = connect(&server, Via::Quic, ClientOptions::default()).await;

        // What the relay does not have, or does not take, under its own namespace.
        assert_eq!(refusal(&c, subscribe(&c, ns.clone(), "3")).await.0, RequestErrorCode::DOES_NOT_EXIST, "a level the relay publishes nothing at");
        assert_eq!(refusal(&c, ControlMessage::Fetch(TrackRequest { request_id: c.next_request_id(), namespace: ns.clone(), name: TrackName::from_str("6"), params: Parameters::new() })).await.0, RequestErrorCode::NOT_SUPPORTED);
        let planted = ControlMessage::Publish(Publish { request_id: c.next_request_id(), namespace: ns.clone(), name: TrackName::from_str("6"), track_alias: 9, params: Parameters::new(), track_properties: Properties::new() });
        assert_eq!(refusal(&c, planted).await, (RequestErrorCode::UNAUTHORIZED, monitoring::OWN_NAMESPACE.to_string()), "nobody else publishes as the relay");

        let (cat_ok, _ctx, _crx) = c.subscribe(ns.clone(), TrackName::from_str(metrics_track::CATALOG_TRACK), Parameters::new()).await.expect("the catalog track");
        let (data_ok, _dtx, _drx) = c.subscribe(ns.clone(), Level::INFORMATIONAL.track_name(), Parameters::new()).await.expect("the data track");

        let mut catalogs: BTreeMap<u64, Catalog> = BTreeMap::new();
        let mut waiting: Vec<Got> = Vec::new();
        let mut held = 0;
        let mut last = None;
        let until = tokio::time::Instant::now() + Duration::from_secs(60);
        while held < 3 {
            assert!(tokio::time::Instant::now() < until, "only {held} groups were held to an exposition");
            let r = c.next_data(STEP).await.expect("a group in time");
            let g = read_stream(r).await.expect("a subgroup stream");
            assert_eq!(g.reset, None);
            if g.header.track_alias == cat_ok.track_alias {
                let cat = Catalog::read(g.header.group_id, &objects(&g)).unwrap_or_else(|e| panic!("the catalog group: {e:?}"));
                assert_eq!(cat.attributes, vec![("service".to_string(), "wai-relay".to_string())]);
                catalogs.insert(cat.version, cat);
            } else {
                assert_eq!(g.header.track_alias, data_ok.track_alias);
                assert_eq!(g.header.publisher_priority, Some(Level::INFORMATIONAL.default_priority()));
                waiting.push(g);
            }
            // The text the control listener answers now, before the next capture.
            let now = scrape(control).await;
            let theirs = metrics_track::parse_exposition(&now).expect("the exposition");
            for g in std::mem::take(&mut waiting) {
                // A data group names its catalog version; one whose catalog is still on its
                // way waits for it.
                let reading = catalogs.values().rev().find_map(|cat| metrics_track::read_group(cat, g.header.group_id, &objects(&g)).ok());
                let Some(reading) = reading else {
                    waiting.push(g);
                    continue;
                };
                assert!(last.is_none_or(|l| g.header.group_id > l), "a group a capture, each later than the one before");
                last = Some(g.header.group_id);
                assert_eq!(reading.unix_ns / 1_000_000, g.header.group_id, "the Group ID is the capture time");
                // Only a group that is the latest capture can be held to this text.
                if captures_of(&reading.samples) != captures_of(&theirs) {
                    continue;
                }
                metrics_track::equal_to_exposition(&reading.samples, &now).unwrap_or_else(|e| panic!("the track and /metrics differ: {e:?}"));
                held += 1;

                // Planted: one counter above its exposition's.
                let mut above = reading.samples.clone();
                let s = above.iter_mut().find(|s| s.name == "wai_relay_sessions_opened_total").expect("the sessions counter");
                let Number::Unsigned(n) = s.value else { panic!("a counter that is no unsigned integer") };
                assert!(n >= 1, "this test's own session is counted");
                s.value = Number::Unsigned(n + 1);
                assert_eq!(metrics_track::equal_to_exposition(&above, &now).unwrap_err().code, "value_disagrees");
                // Planted: a series the exposition does not have.
                let mut more = reading.samples.clone();
                more.push(metrics_track::Sample { name: "wai_relay_planted".into(), labels: vec![], kind: metrics_track::Kind::Gauge, value: Number::Unsigned(1) });
                assert_eq!(metrics_track::equal_to_exposition(&more, &now).unwrap_err().code, "series_disagree");
            }
        }
        // Nothing of the track is a track of the relay's: none cached, none to fetch.
        assert_eq!(server.relay().largest(&(ns, Level::INFORMATIONAL.track_name())), None);
        c.close(0, "done");
        server.shutdown().await;

        // A relay whose operator turned nothing on serves none of it.
        let server = relay(|_| {}).await;
        let (c, _ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
        let (code, reason) = refusal(&c, subscribe(&c, metrics_track::namespace(resource(&server)), "6")).await;
        assert_ne!(reason, "the relay has no such monitoring track", "{code:?}: answered as any track nobody publishes");
        c.close(0, "done");
        server.shutdown().await;
    })
    .await
    .expect("in time");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn the_log_track_carries_what_the_relay_logged_and_no_address() {
    timeout(LIMIT, async {
        let server = relay(|c| c.monitoring.log_track = true).await;
        let ns = log_track::namespace(resource(&server));
        let (c, _ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
        let (ok, _tx, _rx) = c.subscribe(ns, Level::INFORMATIONAL.track_name(), Parameters::new()).await.expect("the log track");
        // The subscription begins with the next second's group.
        tokio::time::sleep(Duration::from_millis(1100)).await;

        // Planted: a message with a peer's address in an attribute.
        let monitor = server.relay().monitor();
        for address in ["192.0.2.7", "192.0.2.7:4443", "[2001:db8::1]:443", "2001:db8::1"] {
            let refused = monitor.log(Level::INFORMATIONAL, "planted", "a planted message", vec![("peer".into(), address.into())]);
            assert_eq!(refused, Err("address_in_attribute"), "{address}");
        }
        assert_eq!(monitor.log_refused("address_in_attribute"), 4);
        // Another session comes and goes: the relay logs it opened and closed.
        let (other, _eo) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
        other.close(0, "done");
        assert!(until(STEP, || server.relay().stats().sessions_open == 1).await, "the other session ended");
        monitor.log(Level::INFORMATIONAL, "after", "a message after", vec![("n".into(), "1".into())]).expect("a message the rules take");

        let mut entries = Vec::new();
        let until_all = tokio::time::Instant::now() + Duration::from_secs(30);
        while !entries.iter().any(|(_, e): &(u64, log_track::LogEntry)| e.msg_id == "after") || !entries.iter().any(|(_, e)| e.msg_id == "session_closed") {
            assert!(tokio::time::Instant::now() < until_all, "the messages did not all come: {entries:?}");
            let g = read_stream(c.next_data(STEP).await.expect("a group in time")).await.expect("a subgroup stream");
            assert_eq!((g.header.track_alias, g.reset), (ok.track_alias, None));
            let read = log_track::read_group(g.header.group_id, &objects(&g)).unwrap_or_else(|e| panic!("a log group: {e:?}"));
            for (unix_ns, _) in &read {
                assert_eq!(unix_ns / 1_000_000_000, g.header.group_id, "a group a second");
            }
            entries.extend(read);
        }
        let ids: Vec<&str> = entries.iter().map(|(_, e)| e.msg_id.as_str()).collect();
        assert!(!ids.contains(&"planted"), "{ids:?}");
        let opened = entries.iter().find(|(_, e)| e.msg_id == "session_opened").expect("the other session's opening").1.clone();
        assert_eq!((opened.app_name.as_str(), opened.hostname.as_str(), opened.proc_id.as_str()), ("wai-relay", "", ""), "no name of the machine");
        assert!(opened.attributes.iter().any(|(k, v)| k == "transport" && v == "websocket"), "{opened:?}");
        for (_, e) in &entries {
            for (k, v) in &e.attributes {
                assert!(v.parse::<std::net::IpAddr>().is_err() && v.parse::<SocketAddr>().is_err() && !v.contains("127.0.0.1"), "{k}={v}");
            }
            assert!(!e.message.contains("127.0.0.1"), "{}", e.message);
        }
        c.close(0, "done");
        server.shutdown().await;
    })
    .await
    .expect("in time");
}

/// A publisher of one track and a subscriber of it, with `groups` groups sent and received.
async fn a_track(server: &RelayServer, sub_opts: ClientOptions, groups: u64) -> (Client, Client, TrackNamespace, TrackName, Vec<(joule_moq_relay::wire::ReqTx, joule_moq_relay::wire::ReqRx)>) {
    let (ns_, name_) = (ns(&["monitoring", "room", "video"]), name("0"));
    let (p, _) = connect(server, Via::WebSocket, ClientOptions::default()).await;
    let mut held = vec![p.publish(ns_.clone(), name_.clone(), 5, Parameters::new(), track_props(64, None)).await.expect("PUBLISH")];
    let (s, _) = connect(server, Via::WebSocket, sub_opts).await;
    let (_, tx, rx) = s.subscribe(ns_.clone(), name_.clone(), Parameters::new()).await.expect("SUBSCRIBE");
    held.push((tx, rx));
    for g in 0..groups {
        let h = SubgroupHeader { track_alias: 5, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(64), properties: false, end_of_group: true, first_object: true };
        p.send_subgroup(h, &[SubgroupObject::normal(0, format!("payload-not-for-a-log-{g}").into_bytes())]).await.expect("a group");
    }
    let got = collect(&s, STEP, |g| g.len() as u64 >= groups).await;
    assert_eq!(got.len() as u64, groups);
    (p, s, ns_, name_, held)
}

#[cfg(feature = "qlog")]
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn each_session_has_a_qlog_that_holds_to_its_order() {
    use joule_moq::qlog;
    timeout(LIMIT, async {
        let dir = store_dir("monitoring-qlog");
        let _ = std::fs::remove_dir_all(&dir);
        let server = relay(|c| c.monitoring.qlog_dir = Some(dir.clone())).await;
        let (p, s, _, _, held) = a_track(&server, ClientOptions::default(), 3).await;
        drop(held);
        p.close(0, "done");
        s.close(0, "done");
        server.shutdown().await;

        let mut names = BTreeMap::new();
        let mut files = 0;
        for f in std::fs::read_dir(&dir).expect("the directory") {
            let path = f.unwrap().path();
            assert_eq!(path.extension().and_then(|e| e.to_str()), Some("sqlog"));
            let bytes = std::fs::read(&path).expect("a log");
            assert!(!String::from_utf8_lossy(&bytes).contains("payload-not-for-a-log"), "a qlog states lengths, never bytes");
            let records = qlog::parse_seq(&bytes).expect("JSON Text Sequences");
            assert_eq!(records[0], qlog::header(qlog::Vantage::Server, None), "the header record");
            let events = &records[1..];
            let found = qlog::check(events);
            assert!(found.is_empty(), "{}: {found:?}", path.display());
            for e in events {
                *names.entry(e["name"].as_str().unwrap().to_string()).or_insert(0u64) += 1;
            }
            files += 1;

            // Planted: an object's record moved before its stream's header.
            if let Some(at) = events.iter().position(|e| e["name"].as_str().is_some_and(|n| n.starts_with("moqt:subgroup_object"))) {
                let stream = &events[at]["data"]["stream_id"];
                let header = events.iter().position(|e| e["name"].as_str().is_some_and(|n| n.starts_with("moqt:subgroup_header")) && &e["data"]["stream_id"] == stream).expect("its header");
                assert!(header < at);
                let mut moved = events.to_vec();
                let o = moved.remove(at);
                moved.insert(header, o);
                let codes: Vec<&str> = qlog::check(&moved).iter().map(|f| f.code).collect();
                assert!(codes.contains(&"object_before_header") && codes.contains(&"time_regress"), "{codes:?}");
            }
        }
        assert_eq!(files, 2, "a file a session");
        // The publisher's log has what the relay read of it, the subscriber's what it wrote.
        for want in ["moqt:control_message_parsed", "moqt:control_message_created", "moqt:stream_type_set", "moqt:subgroup_header_parsed", "moqt:subgroup_object_parsed", "moqt:subgroup_header_created", "moqt:subgroup_object_created"] {
            assert!(names.get(want).is_some_and(|n| *n >= 1), "no {want}: {names:?}");
        }
        assert_eq!((names["moqt:subgroup_object_parsed"], names["moqt:subgroup_object_created"]), (3, 3), "{names:?}");
        let _ = std::fs::remove_dir_all(&dir);
    })
    .await
    .expect("in time");
}

#[cfg(not(feature = "qlog"))]
#[tokio::test]
async fn a_relay_built_without_qlog_does_not_start_with_one_asked_for() {
    let mut c = joule_moq_relay::RelayConfig::loopback();
    c.monitoring.qlog_dir = Some(store_dir("monitoring-qlog-off"));
    let refused = RelayServer::start(c).await.err().expect("a relay with no qlog code");
    assert!(refused.contains("qlog"), "{refused}");
}

fn report(sequence: u64, entries: Vec<ObjectEntry>, metrics: Vec<(u64, u64)>) -> Vec<u8> {
    let received = entries.iter().filter(|e| e.status == Status::Received).count() as u64;
    let lost = entries.len() as u64 - received;
    let r = Report { timestamp_us: 1_000_000 * (sequence + 1), sequence, entries, summary: Summary { interval_us: 1_000_000, evaluated: received + lost, received, received_late: 0, lost, avg_inter_arrival_delta_us: 0 }, metrics };
    r.encode(true).expect("a report")
}

fn got(g: u64, at_us: u64) -> ObjectEntry {
    ObjectEntry { group_id: g, object_id: 0, status: Status::Received, arrival_us: Some(at_us) }
}

async fn send_report(c: &Client, alias: u64, group: u64, payload: Vec<u8>) {
    let h = SubgroupHeader { track_alias: alias, group_id: group, subgroup_id: SubgroupId::Zero, publisher_priority: Some(64), properties: false, end_of_group: true, first_object: true };
    c.send_subgroup(h, &[SubgroupObject::normal(0, payload)]).await.expect("a report");
}

fn feedback_publish(c: &Client, ns: &TrackNamespace, name: &TrackName, alias: u64, labelled: FeedbackType, property: FeedbackType) -> ControlMessage {
    let (fns, fname) = feedback::feedback_track(ns, name, labelled, b"viewer-device-17").unwrap();
    let mut props = Properties::new();
    props.push(feedback::type_property(property));
    ControlMessage::Publish(Publish { request_id: c.next_request_id(), namespace: fns, name: fname, track_alias: alias, params: Parameters::new(), track_properties: props })
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_hop_feedback_track_is_consumed_as_the_mapping_allows() {
    timeout(LIMIT, async {
        let server = relay(|c| c.monitoring.feedback = true).await;
        let monitor = server.relay().monitor();
        let declares = ClientOptions { extra_setup: vec![Capabilities { publish: true, optional_metrics: true, subscribe: false }.option()], ..ClientOptions::default() };
        let (_p, s, ns_, name_, _held) = a_track(&server, declares.clone(), 3).await;
        assert_eq!(Capabilities::of_setup(&s.peer_setup.options).unwrap(), monitoring::capabilities(), "the relay's SETUP declares what it consumes");

        // Refused, each for its own reason.
        let (quiet, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
        assert_eq!(refusal(&quiet, feedback_publish(&quiet, &ns_, &name_, 70, FeedbackType::Hop, FeedbackType::Hop)).await, (RequestErrorCode::NOT_SUPPORTED, "not_negotiated".into()));
        let (elsewhere, _) = connect(&server, Via::WebSocket, declares).await;
        assert_eq!(refusal(&elsewhere, feedback_publish(&elsewhere, &ns_, &name_, 70, FeedbackType::Hop, FeedbackType::Hop)).await, (RequestErrorCode::UNAUTHORIZED, monitoring::FEEDBACK_NOT_SUBSCRIBED.into()));
        assert_eq!(refusal(&s, feedback_publish(&s, &ns_, &name_, 70, FeedbackType::Hop, FeedbackType::Origin)).await, (RequestErrorCode::NOT_SUPPORTED, "type_mismatch".into()));

        // Accepted from the track's own subscriber.
        let (_ftx, mut frx) = s.request(&feedback_publish(&s, &ns_, &name_, 77, FeedbackType::Hop, FeedbackType::Hop)).await.expect("the request stream");
        assert!(matches!(Client::response(&mut frx).await.expect("an answer"), ControlMessage::RequestOk(_)));

        // A stall begins: PLAYOUT_AHEAD_MS at 0, with a bandwidth estimate, which is withheld.
        send_report(&s, 77, 0, report(0, vec![got(0, 987_654_321)], vec![(feedback::metric::PLAYOUT_AHEAD_MS, 0), (feedback::metric::ESTIMATED_BANDWIDTH_KBPS, 7_777_777)])).await;
        assert!(until(STEP, || monitor.observations().len() == 1).await, "{:?}", monitor.observations());
        // Planted: an Optional Metric with no mapping, beside a value that would end the stall.
        send_report(&s, 77, 1, report(1, vec![got(1, 1_900_000)], vec![(feedback::metric::PLAYOUT_AHEAD_MS, 150), (0x3e, 7)])).await;
        assert!(until(STEP, || monitor.feedback_refused("metric_unmapped") == 1).await);
        assert_eq!(monitor.observations().len(), 1, "nothing is written from a refused report");
        // The stall ends.
        send_report(&s, 77, 2, report(2, vec![got(1, 2_900_000), got(2, 2_950_000)], vec![(feedback::metric::PLAYOUT_AHEAD_MS, 150)])).await;
        assert!(until(STEP, || monitor.observations().len() == 2).await, "{:?}", monitor.observations());

        let obs = monitor.observations();
        assert_eq!((obs[0]["event"].as_str(), obs[1]["event"].as_str()), (Some("stall-start"), Some("stall-end")));
        assert_eq!((obs[1]["metrics"]["bsa"].as_u64(), obs[0]["sn"].as_u64(), obs[1]["sn"].as_u64()), (Some(1), Some(0), Some(1)));
        let sid = obs[0]["sid"].as_str().expect("a session id");
        assert!(sid.len() == 32 && obs[1]["sid"] == obs[0]["sid"], "a session of the relay's own");
        let written = serde_json::to_string(&obs).unwrap();
        // The feedback-source-id is in nothing written, nor a figure with no member, nor a
        // time of the publisher's clock.
        for absent in ["viewer-device-17", &hex::encode("viewer-device-17"), "7777777", "987654321"] {
            assert!(!written.contains(absent), "{absent} in {written}");
        }
        let s_ = server.relay().stats();
        assert_eq!(s_.sessions_open, 4);
        let text = joule_moq_relay::control::exposition(server.relay());
        let f = wai_metrics::read::parse(&text).expect("the exposition");
        let at = |n: &str, l: &[(&str, &str)]| wai_metrics::read::value(&f, n, l);
        assert_eq!((at("wai_relay_feedback_reports_total", &[]), at("wai_relay_feedback_observations_total", &[])), (Some(2.0), Some(2.0)), "{text}");
        assert_eq!((at("wai_relay_feedback_refused_total", &[("why", "metric_unmapped")]), at("wai_relay_feedback_withheld_total", &[("why", "no_member")]), at("wai_relay_feedback_withheld_total", &[("why", "local_clock")])), (Some(1.0), Some(1.0), Some(2.0)), "{text}");

        // Planted: a tally above what the relay sent this session on the track (3 objects).
        send_report(&s, 77, 3, report(3, (0..9).map(|g| got(g, 3_000_000 + g)).collect(), vec![])).await;
        assert!(until(STEP, || monitor.feedback_refused("more_than_sent") == 1).await);
        assert_eq!(monitor.observations().len(), 2);
        // The relay ends the Feedback Track; the session stands.
        let ended = timeout(STEP, frx.next()).await.expect("the track's end in time");
        assert!(ended.is_err(), "the request stream is reset: {ended:?}");
        let (_, _tx2, _rx2) = s.subscribe(ns(&["monitoring", "room", "video"]), name("0"), Parameters::new()).await.expect("the session stands");

        // Nothing of the Feedback Track is a track of the relay's.
        let (fns, fname) = feedback::feedback_track(&ns_, &name_, FeedbackType::Hop, b"viewer-device-17").unwrap();
        assert_eq!(server.relay().largest(&(fns, fname)), None);
        for c in [&s, &quiet, &elsewhere] {
            c.close(0, "done");
        }
        server.shutdown().await;
    })
    .await
    .expect("in time");
}