//! Sessions at `draft-ietf-moq-transport-23` through the relay, beside -22 ones: the relay
//! chooses -23 by ALPN, `WT-Available-Protocols` or the WebSocket subprotocol and bridges it
//! to its -22 model (`transport::bridge`). WAI objects and their receipts cross -23 and -22
//! sessions both ways and verify at each subscriber; a FETCH from the cache returns them at
//! -23, its FETCH_OK a REQUEST_OK; a tampered object is refused at the relay's door. Raw
//! sessions written byte by byte from -23's own messages, not through this crate's client,
//! check the relay's -23 wire, and each planted fault closes the session with -23's code
//! for it.
//!
//! A -23 session's subgroup streams are read by -23's object rules, where they differ from
//! -22's (a supported property outside its scope or repeated makes the track malformed; a
//! grease value in the Mandatory Track Property range is no Mandatory Track Property), each
//! held against the same object from a -22 session, which keeps -22's reading. And the relay's
//! LARGEST_OBJECT is held to -23 §7.5 by a vector of what a publisher states and sends.
//!
//! A raw -23 SUBSCRIBE_TRACKS: the subscription parameters it carries that -22's has no place for
//! (both delivery timeouts, SUBSCRIBER_PRIORITY, LOCATION_FILTER) are stated in each PUBLISH it
//! brings and govern the subscription; its FILL_PARAMETERS are carried in each PUBLISH, which
//! opens a fill fetch stream named by the PUBLISH's Request ID, and none for a paused one. A
//! fill, a PUBLISH's or a SUBSCRIBE's, ends at the Largest Object the PUBLISH or SUBSCRIBE_OK
//! stated, so an object published while it is in flight comes on the subscription alone.

mod common;

use common::*;
use ed25519_dalek::SigningKey;
use joule_moq::carriage::{object_properties, ObjectExtras, ReceiptBundle};
use joule_moq::draft::Draft;
use joule_moq::draft22::data::{FetchEntry, ObjectStatus as WireStatus, SubgroupHeader, SubgroupId, SubgroupObject};
use joule_moq::draft22::{Location, LocationFilter, Parameter, Parameters};
use joule_moq::draft23 as m23;
use joule_moq::messages::ObjectHeader;
use joule_moq::receipt::ReceiptVerifier;
use joule_moq::{GroupId, ObjectId, ObjectStatus, TrackAlias};
use joule_moq_relay::client::{Client, ClientOptions};
use joule_moq_relay::transport::{quic, BoxRecv, SharedConnection, TransportError};
use std::sync::Arc;
use std::time::Duration;
use tokio::time::timeout;

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

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn every_session_at_23_over_quic() {
    timeout(LIMIT, two_publishers_three_subscribers(Via::QuicDraft(Draft::D23))).await.expect("in time");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn every_session_at_23_over_webtransport() {
    timeout(LIMIT, two_publishers_three_subscribers(Via::WtDraft(Draft::D23))).await.expect("in time");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn every_session_at_23_over_websocket() {
    timeout(LIMIT, two_publishers_three_subscribers(Via::WsDraft(Draft::D23))).await.expect("in time");
}

/// A -23 publisher and a -22 one; subscribers at -22 (the namespace subscriber), -23 over
/// WebTransport and -23 over the WebSocket binding: every object of both publishers reaches
/// every subscriber and verifies against its publisher's receipts.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_23_and_a_22_client_through_one_relay() {
    let vias = [Via::QuicDraft(Draft::D23), Via::Quic, Via::Quic, Via::WtDraft(Draft::D23), Via::WsDraft(Draft::D23)];
    timeout(LIMIT, two_publishers_three_subscribers_on(vias)).await.expect("in time");
}

/// The reverse: a -22 publisher by namespace, a -23 one by PUBLISH, and the namespace
/// subscriber at -23.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_22_publisher_and_23_subscribers() {
    let vias = [Via::Quic, Via::QuicDraft(Draft::D23), Via::QuicDraft(Draft::D23), Via::Quic, Via::WtDraft(Draft::D22)];
    timeout(LIMIT, two_publishers_three_subscribers_on(vias)).await.expect("in time");
}

/// A late joiner at -23 FETCHes a track and its receipts, published at -22, from the
/// relay's cache: -23's FETCH_OK is a REQUEST_OK whose LARGEST_OBJECT gives the end, and the
/// fetch stream's Object ID deltas and indicators are -23's. The same objects come back and
/// verify; a range comes back whole; a relative filter is refused (INVALID_RANGE, §3.2).
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_fetch_from_the_cache_at_23() {
    timeout(LIMIT, async {
        const GROUPS: u64 = 6;
        let server = relay(|_| {}).await;
        let key = SigningKey::from_bytes(&[0x2c; 32]);
        let t = Tracks::new(ns(&["vod", "d23"]), "ep", 10, 5, 100);
        let (publisher, _ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
        let _held = publish_tracks(&publisher, &t, None).await;
        for g in 0..GROUPS {
            publish_group(&publisher, &key, &t, g, &payload("ep", g, 1500)).await;
        }
        let relay = server.relay().clone();
        let (mk, rk) = ((t.ns.clone(), t.media.clone()), (t.ns.clone(), t.receipts.clone()));
        assert!(
            until(STEP, || relay.largest(&mk) == Some(Location::new(GROUPS - 1, 0)) && relay.cache_covers(&mk, Location::new(0, 0)) && relay.cache_covers(&rk, Location::new(0, 0))).await,
            "the cache holds the track"
        );
        let verifier = ReceiptVerifier::with_anchors(vec![key.verifying_key()]);
        for via in [Via::QuicDraft(Draft::D23), Via::WtDraft(Draft::D23)] {
            let (late, _ep) = connect(&server, via, ClientOptions::default()).await;
            let (ok, media) = late.fetch(t.ns.clone(), t.media.clone(), Parameters::new()).await.unwrap_or_else(|e| panic!("FETCH at -23: {e}"));
            assert_eq!(ok.end_location, Location::new(GROUPS - 1, 0), "the end, from -23's LARGEST_OBJECT");
            let (_, rec) = late.fetch(t.ns.clone(), t.receipts.clone(), Parameters::new()).await.expect("FETCH the receipts");
            let objs = |v: &[FetchEntry]| v.iter().filter_map(|e| if let FetchEntry::Object(o) = e { Some(o.clone()) } else { None }).collect::<Vec<_>>();
            let (media, rec) = (objs(&media), objs(&rec));
            assert_eq!((media.len(), rec.len()), (GROUPS as usize, GROUPS as usize));
            for (m, r) in media.iter().zip(&rec) {
                let b = ReceiptBundle::from_payload(&r.payload).unwrap();
                let vg = verifier.verify_group(&b.group).expect("a receipt verifies");
                verifier.verify_payload(&m.payload, &b.objects[0], &vg).expect("an object verifies");
                assert_eq!(m.payload, payload("ep", m.group_id, 1500));
            }
            let range = Parameters::new().with(Parameter::LocationFilter(LocationFilter::AbsoluteStartGroupEnd { start: Location::new(2, 0), end_group_delta: 1 }));
            let (ok, part) = late.fetch(t.ns.clone(), t.media.clone(), range).await.expect("FETCH a range");
            assert_eq!(objs(&part).iter().map(|o| o.group_id).collect::<Vec<_>>(), vec![2, 3]);
            assert_eq!(ok.end_location.group, 3);
            let rel = Parameters::new().with(Parameter::LocationFilter(LocationFilter::NextObject));
            let e = late.fetch(t.ns.clone(), t.media.clone(), rel).await.expect_err("a relative filter in FETCH");
            assert!(e.contains("INVALID_RANGE"), "{e}");
            assert!(late.conn.notes().iter().any(|n| n.contains("relative Location filter")), "{:?}", late.conn.notes());
        }
        server.shutdown().await;
    })
    .await
    .expect("in time");
}

/// Group `g` of `t` signed over `signed`, its object carrying `sent`.
async fn publish_group_as(c: &Client, key: &SigningKey, t: &Tracks, g: u64, signed: &[u8], sent: &[u8]) {
    let (receipt, bundle) = signed_group(key, t.receipt_alias, g, signed);
    let rh = SubgroupHeader { track_alias: t.receipts_alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(t.priority), properties: false, end_of_group: true, first_object: true };
    c.send_subgroup(rh, &[SubgroupObject::normal(0, bundle.to_payload().unwrap())]).await.unwrap();
    let header = ObjectHeader { track_alias: TrackAlias(t.media_alias), group_id: GroupId(g), object_id: ObjectId(0), publisher_priority: t.priority, status: ObjectStatus::Normal, payload_len: sent.len() as u64, mrl_layer: 0, mrl_total_layers: 0, wai_envelope: false };
    let props = object_properties(&header, &ObjectExtras { receipt: Some(receipt), wai_receipt: None, delivery_timeout_ms: None }).unwrap();
    let mh = SubgroupHeader { track_alias: t.media_alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(t.priority), properties: true, end_of_group: true, first_object: true };
    c.send_subgroup(mh, &[SubgroupObject { object_id: 0, properties: props, status: WireStatus::Normal, payload: sent.to_vec() }]).await.unwrap();
}

/// A -23 publisher sends one group whose object is not the one its receipt signs: the
/// relay's door refuses that object alone, and a -23 subscriber gets every other group,
/// each verifying.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_tampered_object_at_23_is_refused_at_the_door() {
    timeout(LIMIT, async {
        const G: u64 = 5;
        const BAD: u64 = 2;
        let key = SigningKey::from_bytes(&[0xd3; 32]);
        let server = relay(|c| {
            c.receipts.door = vec![joule_moq_relay::config::DoorRule { prefix: vec!["live".into()], trust: vec![hex::encode(key.verifying_key().to_bytes())], trust_field: None, hold_ms: 5_000 }];
        })
        .await;
        let t = Tracks::new(ns(&["live", "t23"]), "cam", 30, 7, 80);
        let (p, _ep) = connect(&server, Via::QuicDraft(Draft::D23), ClientOptions::default()).await;
        let _held = publish_tracks(&p, &t, None).await;
        let (s, _es) = connect(&server, Via::QuicDraft(Draft::D23), ClientOptions::default()).await;
        let view = subscribe_both(&s, &t, Parameters::new()).await;
        for g in 0..G {
            let good = payload("t23", g, 700);
            if g == BAD {
                let mut bad = good.clone();
                bad[0] ^= 0xff;
                publish_group_as(&p, &key, &t, g, &good, &bad).await;
            } else {
                publish_group(&p, &key, &t, g, &good).await;
            }
        }
        assert!(until(STEP, || server.relay().stats().refused_at_door >= 1).await, "the door refused the tampered object");
        // Every media group but the tampered one, and the receipts of each: the two tracks
        // ride streams of their own, so a group's object may arrive before its receipts.
        let got = collect(&s, STEP, |got| {
            let media: Vec<u64> = got.iter().filter(|x| x.header.track_alias == view.media_alias).map(|x| x.header.group_id).collect();
            media.len() as u64 >= G - 1 && media.iter().all(|g| got.iter().any(|x| x.header.track_alias == view.receipts_alias && x.header.group_id == *g))
        })
        .await;
        let (objects, bundles, _, _) = verify(&got, &view, &key);
        assert_eq!(objects, (G - 1) as usize, "every object but the tampered one, each verifying");
        assert!(bundles >= (G - 1) as usize);
        assert!(got.iter().filter(|x| x.header.track_alias == view.media_alias).all(|x| x.header.group_id != BAD), "the tampered object is not forwarded");
        assert_eq!(server.relay().stats().refused_at_door, 1);
        server.shutdown().await;
    })
    .await
    .expect("in time");
}

// ── Raw -23 sessions ────────────────────────────────────────────────────────

async fn raw_session(server: &joule_moq_relay::RelayServer) -> (SharedConnection, quinn::Endpoint) {
    let ep = quic::client_endpoint(loopback(), vec![server.identity.chain[0].clone()], vec![m23::ALPN.to_vec()], None).unwrap();
    let (conn, got) = timeout(STEP, quic::connect_any(&ep, server.quic_addr.unwrap(), "localhost")).await.expect("in time").expect("a QUIC session");
    assert_eq!(got, Draft::D23);
    (Arc::new(conn), ep)
}

/// Read one framed -23 message (Type, a varint Length, the body).
async fn read23(r: &mut BoxRecv, buf: &mut Vec<u8>) -> m23::ControlMessage {
    loop {
        let mut rd = joule_moq::draft22::io::Reader::new(buf);
        if let Ok(m) = m23::ControlMessage::decode(&mut rd) {
            let n = rd.position();
            buf.drain(..n);
            return m;
        }
        match timeout(STEP, r.read_chunk()).await.expect("in time").expect("a chunk") {
            Some(b) => buf.extend_from_slice(&b),
            None => panic!("the stream ended"),
        }
    }
}

/// A -23 session with its SETUP sent and the relay's read; the control stream is returned
/// to be held open.
async fn setup23(server: &joule_moq_relay::RelayServer) -> (SharedConnection, quinn::Endpoint, m23::Setup, joule_moq_relay::transport::BoxSend) {
    let (conn, ep) = raw_session(server).await;
    let mut ctl = conn.open_uni().await.unwrap();
    let mut o = joule_moq::draft22::kvp::SetupOptions::new();
    o.push(joule_moq::draft22::kvp::KeyValuePair::bytes(joule_moq::draft22::kvp::opt::MOQT_IMPLEMENTATION, b"raw-23".to_vec()));
    ctl.write_all(&m23::ControlMessage::Setup(m23::Setup { options: o }).encode().unwrap()).await.unwrap();
    let mut theirs = timeout(STEP, conn.accept_uni()).await.expect("in time").expect("a stream");
    let mut d = m23::UniStreamDecoder::new(joule_moq::draft22::GroupOrder::Ascending, Default::default());
    let setup = loop {
        match d.next().unwrap() {
            Some(m23::UniEvent::Control(m)) => break m,
            Some(_) => continue,
            None => d.push(&timeout(STEP, theirs.read_chunk()).await.unwrap().unwrap().unwrap()),
        }
    };
    let m23::ControlMessage::Setup(s) = setup else { panic!("{setup:?}") };
    (conn, ep, s, ctl)
}

fn closed_with(e: TransportError) -> u64 {
    match e {
        TransportError::Closed { code, .. } => code,
        other => panic!("{other:?}"),
    }
}

/// A -23 session written from -23's messages: SETUP each way, with a varint Length; a
/// PUBLISH_NAMESPACE answered with REQUEST_OK; two SUBSCRIBE_NAMESPACEs whose prefixes
/// overlap, both accepted (-23 §4.2: -22 refused the second with PREFIX_OVERLAP).
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_raw_23_session() {
    timeout(LIMIT, async {
        let server = relay(|_| {}).await;
        let (conn, _ep, s, _ctl) = setup23(&server).await;
        assert!(s.options.implementation().is_some());
        let mut held = Vec::new();
        for (id, req) in [
            (0, m23::ControlMessage::PublishNamespace(m23::NamespaceRequest { request_id: 0, prefix: ns(&["raw23"]), params: m23::Parameters::new() })),
            (2, m23::ControlMessage::SubscribeNamespace(m23::NamespaceRequest { request_id: 2, prefix: ns(&["over"]), params: m23::Parameters::new() })),
            (4, m23::ControlMessage::SubscribeNamespace(m23::NamespaceRequest { request_id: 4, prefix: ns(&["over", "lap"]), params: m23::Parameters::new() })),
        ] {
            let (mut tx, mut rx) = conn.open_bi().await.unwrap();
            tx.write_all(&req.encode().unwrap()).await.unwrap();
            let mut buf = Vec::new();
            let m = read23(&mut rx, &mut buf).await;
            assert!(matches!(m, m23::ControlMessage::RequestOk(_)), "request {id}: {m:?}");
            held.push((tx, rx));
        }
        server.shutdown().await;
    })
    .await
    .expect("in time");
}

/// Open a -23 session, write `bytes` on a request stream of its own (after `first`, when
/// given, and its answer), and return the code the relay closes the session with.
async fn fault(server: &joule_moq_relay::RelayServer, first: Option<m23::ControlMessage>, bytes: Vec<u8>) -> u64 {
    let (conn, _ep, _, _ctl) = setup23(server).await;
    let (mut tx, mut rx) = conn.open_bi().await.unwrap();
    if let Some(m) = first {
        tx.write_all(&m.encode().unwrap()).await.unwrap();
        let mut buf = Vec::new();
        let a = read23(&mut rx, &mut buf).await;
        assert!(matches!(a, m23::ControlMessage::RequestOk(_)), "{a:?}");
    }
    tx.write_all(&bytes).await.unwrap();
    closed_with(timeout(STEP, conn.closed()).await.expect("the relay closes the session"))
}

/// Planted faults, each on a session of its own, each closed with -23's code for it:
///
/// - a -22-only message, FETCH_OK (`0x18`, reserved at -23): UNEXPECTED_MESSAGE;
/// - a -22-only parameter, FORWARD (`0x10`), in a SUBSCRIBE: INVALID_PARAMETER;
/// - a message framed as -22 frames it (a 16-bit Length): MALFORMED_MESSAGE;
/// - a missing -23 field, PUBLISH_NAMESPACE without its Number of Parameters: MALFORMED_MESSAGE;
/// - a GOAWAY from the requester on its own request stream (-22 allowed it, -23 §9.2 does
///   not): PROTOCOL_VIOLATION;
/// - an empty Track Name (-23 §2.4.1): PROTOCOL_VIOLATION.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn each_planted_fault_closes_the_session_with_its_code() {
    timeout(LIMIT, async {
        let server = relay(|_| {}).await;
        let v = |x: u64| joule_moq::draft22::varint::to_vec(x);
        let frame = |ty: u64, body: &[u8]| [v(ty), v(body.len() as u64), body.to_vec()].concat();
        // -22's FETCH_OK: End Of Track 0, End Location {1, 0}, no parameters.
        let fetch_ok = frame(0x18, &[0x00, 0x01, 0x00, 0x00]);
        // A SUBSCRIBE (request 0, namespace ["a"], name "t") with FORWARD = 1.
        let forward = frame(0x03, &[0x00, 0x01, 0x01, b'a', 0x01, b't', 0x01, 0x10, 0x01]);
        // PUBLISH_NAMESPACE with a 16-bit Length, as -22 frames it.
        let body22 = [0x00u8, 0x01, 0x01, b'p', 0x00];
        let framed22 = [vec![0x06, 0x00, body22.len() as u8], body22.to_vec()].concat();
        // PUBLISH_NAMESPACE (request 0, prefix ["p"]) and no Number of Parameters.
        let missing = frame(0x06, &[0x00, 0x01, 0x01, b'p']);
        // GOAWAY with no URI and no timeout, on the stream of an accepted PUBLISH_NAMESPACE.
        let goaway = frame(0x10, &[0x00, 0x00]);
        let pn = m23::ControlMessage::PublishNamespace(m23::NamespaceRequest { request_id: 0, prefix: ns(&["held"]), params: m23::Parameters::new() });
        // SUBSCRIBE with an empty Track Name.
        let empty = frame(0x03, &[0x00, 0x01, 0x01, b'a', 0x00, 0x00]);
        let cases: Vec<(&str, Option<m23::ControlMessage>, Vec<u8>, u64)> = vec![
            ("FETCH_OK, a -22-only message", None, fetch_ok, 0x303),
            ("FORWARD, a -22-only parameter", None, forward, 0x305),
            ("a 16-bit Length", None, framed22, 0x300),
            ("no Number of Parameters", None, missing, 0x300),
            ("the requester's GOAWAY on its request stream", Some(pn), goaway, 0x3),
            ("an empty Track Name", None, empty, 0x3),
        ];
        let mut caught = 0;
        for (what, first, bytes, want) in cases {
            let got = fault(&server, first, bytes).await;
            assert_eq!(got, want, "{what}: closed with {got:#x}, not {want:#x}");
            caught += 1;
        }
        assert_eq!(caught, 6);
        server.shutdown().await;
    })
    .await
    .expect("in time");
}

/// What a session left: a -23 session's summary records `moqt-23`.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_23_sessions_summary_names_its_draft() {
    timeout(LIMIT, async {
        let server = relay(|_| {}).await;
        let (c, _ep) = connect(&server, Via::QuicDraft(Draft::D23), ClientOptions::default()).await;
        let e = c.subscribe(ns(&["nothing", "here"]), name("b"), Parameters::new()).await.expect_err("no publisher");
        assert!(e.contains("DOES_NOT_EXIST"), "{e}");
        c.close(0, "done");
        assert!(until(STEP, || !server.relay().summaries().is_empty()).await);
        assert_eq!(server.relay().summaries()[0].summary.draft, "moqt-23");
        tokio::time::sleep(Duration::from_millis(10)).await;
        server.shutdown().await;
    })
    .await
    .expect("in time");
}

// ── -23's object rules on a -23 session's subgroup streams ──────────────────

/// One object of group `g` on the track of alias `alias`, with `properties`, on a stream of
/// its own.
async fn send_object(c: &Client, alias: u64, g: u64, properties: joule_moq::draft22::kvp::Properties) {
    let h = SubgroupHeader { track_alias: alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(100), properties: true, end_of_group: true, first_object: true };
    // A stream the relay stops for a malformed object may be stopped before its FIN is sent.
    let _ = c.send_subgroup(h, &[SubgroupObject { object_id: 0, properties, status: WireStatus::Normal, payload: payload("o", g, 64) }]).await;
}

/// The groups of the objects `v` is delivered within `wait`, once `n` have come or it passes.
async fn groups_delivered(v: &Client, n: usize, wait: Duration) -> Vec<u64> {
    let mut g: Vec<u64> = collect(v, wait, |x| x.iter().filter(|s| !s.objects.is_empty()).count() >= n).await.iter().filter(|s| !s.objects.is_empty()).map(|s| s.header.group_id).collect();
    g.sort_unstable();
    g
}

/// The same four objects from a -23 publisher and from a -22 one, each to a viewer at its own
/// draft. What -23 changed is what differs:
///
/// - a grease value in the Mandatory Track Property range on an object: at -23 no Mandatory
///   Track Property, forwarded; at -22 one, the track malformed;
/// - a Track-only property (DEFAULT_PUBLISHER_PRIORITY) on an object: at -23 outside its
///   scope, the track malformed; at -22 forwarded;
/// - OBJECT_DELIVERY_TIMEOUT twice, once in each property list: at -23 repeated, the track
///   malformed; at -22 forwarded;
/// - an object with an Object-scoped property once: forwarded at both.
///
/// A malformed object ends its stream and no more: the session stands, and the next object
/// is forwarded.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_23_sessions_objects_are_held_to_23s_rules() {
    use joule_moq::draft22::kvp::{prop, KeyValuePair, Properties};
    timeout(LIMIT, async {
        let server = relay(|_| {}).await;
        // 0x7f * 129 + 0x9d: a grease value (§13) inside 0x4000-0x7FFF.
        let grease: u64 = 0x7f * 129 + 0x9d;
        assert!(joule_moq::draft::is_grease(grease) && prop::MANDATORY_TRACK.contains(&grease) && grease.is_multiple_of(2));
        let one = |kv: KeyValuePair| {
            let mut p = Properties::new();
            p.push(kv);
            p
        };
        let twice = {
            let mut p = one(KeyValuePair::varint(prop::OBJECT_DELIVERY_TIMEOUT, 500));
            p.set_immutable(&one(KeyValuePair::varint(prop::OBJECT_DELIVERY_TIMEOUT, 700))).unwrap();
            p
        };
        // (what, the object's properties, forwarded at -23, forwarded at -22)
        let cases: Vec<(&str, Properties, bool, bool)> = vec![
            ("an Object-scoped property, once", one(KeyValuePair::varint(prop::OBJECT_DELIVERY_TIMEOUT, 500)), true, true),
            ("a grease value in the Mandatory range", one(KeyValuePair::varint(grease, 1)), true, false),
            ("a Track-only property on an object", one(KeyValuePair::varint(prop::DEFAULT_PUBLISHER_PRIORITY, 7)), false, true),
            ("a property that allows one instance, twice", twice, false, true),
        ];
        let mut differ = 0;
        for (draft, via) in [(Draft::D23, Via::QuicDraft(Draft::D23)), (Draft::D22, Via::Quic)] {
            let nsp = ns(&["objects", draft.alpn_str()]);
            let (p, _ep) = connect(&server, via, ClientOptions::default()).await;
            let _held = p.publish(nsp.clone(), name("t"), 7, Parameters::new(), track_props(100, None)).await.expect("PUBLISH");
            let (v, _ev) = connect(&server, via, ClientOptions::default()).await;
            let _sub = v.subscribe(nsp.clone(), name("t"), Parameters::new()).await.expect("SUBSCRIBE");
            let mut expect = Vec::new();
            for (g, (what, props, at23, at22)) in cases.iter().enumerate() {
                send_object(&p, 7, g as u64, props.clone()).await;
                let forwarded = if draft == Draft::D23 { *at23 } else { *at22 };
                if forwarded {
                    expect.push(g as u64);
                }
                differ += usize::from(at23 != at22 && draft == Draft::D23);
                let _ = what;
            }
            // A last object, well formed: forwarded after every malformed one, the session open.
            send_object(&p, 7, 9, Properties::new()).await;
            expect.push(9);
            let got = groups_delivered(&v, expect.len(), Duration::from_secs(5)).await;
            assert_eq!(got, expect, "at {draft}: the objects forwarded, of {:?}", cases.iter().map(|c| c.0).collect::<Vec<_>>());
            assert!(timeout(Duration::from_millis(300), p.conn.closed()).await.is_err(), "at {draft}: a malformed object closed its publisher's session");
            p.close(0, "done");
            v.close(0, "done");
        }
        assert_eq!(differ, 3, "three of the four are read differently at the two drafts");
        server.shutdown().await;
    })
    .await
    .expect("in time");
}

// ── LARGEST_OBJECT, by -23 §7.5 ─────────────────────────────────────────────

/// What a TRACK_STATUS of `key`'s track answers for its Largest Object, asked by `c`.
async fn status_largest(c: &Client, nsp: &joule_moq::TrackNamespace) -> Option<Location> {
    use joule_moq::draft22::{ControlMessage, TrackRequest};
    let m = ControlMessage::TrackStatus(TrackRequest { request_id: c.next_request_id(), namespace: nsp.clone(), name: name("t"), params: Parameters::new() });
    let (_tx, mut rx) = c.request(&m).await.expect("TRACK_STATUS");
    match Client::response(&mut rx).await.expect("an answer") {
        ControlMessage::RequestOk(ok) => ok.params.largest_object(),
        other => panic!("{other:?}"),
    }
}

/// -23 §7.5: a relay's LARGEST_OBJECT is the largest of the LARGEST_OBJECT values its upstream
/// publisher stated (in PUBLISH and PUBLISH_STATE_NOTIFY here) and the largest Location of an
/// object received on the upstream subscription. The vector: what the publisher states and
/// sends, step by step, and the Largest Object the relay answers after each, in TRACK_STATUS_OK
/// and in the SUBSCRIBE_OK of a subscription made then. A value stated in PUBLISH_STATE_NOTIFY
/// was not taken before; with it left out, step 3 answers {7, 4}.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_relays_largest_object_is_the_largest_stated_or_received() {
    use joule_moq::draft22::ControlMessage;
    #[derive(Clone, Copy)]
    enum Step {
        Notify(u64, u64),
        Object(u64, u64),
    }
    // (what the publisher does, the Largest Object after it)
    let vector: [(Step, (u64, u64)); 6] = [
        (Step::Object(7, 4), (7, 4)),   // an object past the stated {7, 3}
        (Step::Notify(9, 0), (9, 0)),   // stated in PUBLISH_STATE_NOTIFY, no object of it yet
        (Step::Object(8, 5), (9, 0)),   // an object of an earlier group arrives: no smaller
        (Step::Notify(2, 0), (9, 0)),   // a smaller value stated: the largest stands
        (Step::Object(9, 1), (9, 1)),   // an object past what was stated
        (Step::Notify(9, 1), (9, 1)),   // stated again as it is
    ];
    timeout(LIMIT, async {
        let server = relay(|_| {}).await;
        for (draft, via) in [(Draft::D23, Via::QuicDraft(Draft::D23)), (Draft::D22, Via::Quic)] {
            let nsp = ns(&["largest", draft.alpn_str()]);
            let key = (nsp.clone(), name("t"));
            let (p, _ep) = connect(&server, via, ClientOptions::default()).await;
            let stated = Parameters::new().with(Parameter::LargestObject(Location::new(7, 3)));
            let (tx, _rx) = p.publish(nsp.clone(), name("t"), 7, stated, track_props(100, None)).await.expect("PUBLISH");
            let (v, _ev) = connect(&server, via, ClientOptions::default()).await;
            assert_eq!(status_largest(&v, &nsp).await, Some(Location::new(7, 3)), "at {draft}: what PUBLISH stated, with no object received");
            for (i, (step, (g, o))) in vector.iter().enumerate() {
                match *step {
                    Step::Notify(g, o) => tx.send(&ControlMessage::PublishStateNotify(Parameters::new().with(Parameter::LargestObject(Location::new(g, o))))).await.expect("PUBLISH_STATE_NOTIFY"),
                    Step::Object(g, o) => {
                        let h = SubgroupHeader { track_alias: 7, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(100), properties: false, end_of_group: false, first_object: false };
                        p.send_subgroup(h, &[SubgroupObject::normal(o, payload("l", g, 32))]).await.expect("the object");
                    }
                }
                let want = Some(Location::new(*g, *o));
                // A step that leaves the Largest Object where it was shows only by not moving it.
                let _ = until(Duration::from_millis(400), || server.relay().largest(&key) != vector[..i].last().map_or(Some(Location::new(7, 3)), |(_, (g, o))| Some(Location::new(*g, *o)))).await;
                assert!(until(STEP, || server.relay().largest(&key) == want).await, "at {draft}, step {}: the relay holds {:?}, not {want:?}", i + 1, server.relay().largest(&key));
                assert_eq!(status_largest(&v, &nsp).await, want, "at {draft}, step {}: TRACK_STATUS_OK", i + 1);
                let (ok, stx, _srx) = v.subscribe(nsp.clone(), name("t"), Parameters::new()).await.expect("SUBSCRIBE");
                assert_eq!(ok.params.largest_object(), want, "at {draft}, step {}: SUBSCRIBE_OK", i + 1);
                stx.reset(0x1).await;
            }
            p.close(0, "done");
            v.close(0, "done");
        }
        server.shutdown().await;
    })
    .await
    .expect("in time");
}

// ── SUBSCRIBE_TRACKS at -23, written raw ───────────────────────────────────

/// What one data stream of a raw -23 session carried: its header, its subgroup objects or
/// fetch entries, and whether it ended in a reset.
#[derive(Debug)]
struct Arrived23 {
    header: Option<m23::StreamHeader>,
    objects: Vec<SubgroupObject>,
    entries: Vec<m23::FetchEntry>,
    reset: bool,
}

/// The data streams the relay opens to a raw -23 session, each read to its end, until `done`
/// has held of what arrived for `quiet` (what comes meanwhile is kept too) or `limit` passes.
async fn data23(conn: &SharedConnection, limit: Duration, quiet: Duration, mut done: impl FnMut(&[Arrived23]) -> bool) -> Vec<Arrived23> {
    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<Arrived23>();
    let c = conn.clone();
    let acceptor = tokio::spawn(async move {
        while let Ok(mut r) = c.accept_uni().await {
            let tx = tx.clone();
            tokio::spawn(async move {
                let mut d = m23::UniStreamDecoder::new(joule_moq::draft22::GroupOrder::Ascending, Default::default());
                let mut a = Arrived23 { header: None, objects: Vec::new(), entries: Vec::new(), reset: false };
                'stream: loop {
                    loop {
                        match d.next() {
                            Ok(Some(m23::UniEvent::Header(h))) => a.header = Some(h),
                            Ok(Some(m23::UniEvent::SubgroupObject(o))) => a.objects.push(o),
                            Ok(Some(m23::UniEvent::FetchEntry(e))) => a.entries.push(e),
                            Ok(Some(_)) => {}
                            Ok(None) => break,
                            Err(_) => break 'stream,
                        }
                    }
                    match r.read_chunk().await {
                        Ok(Some(b)) => d.push(&b),
                        Ok(None) => break,
                        Err(_) => {
                            a.reset = true;
                            break;
                        }
                    }
                }
                let _ = tx.send(a);
            });
        }
    });
    let deadline = tokio::time::Instant::now() + limit;
    let mut got = Vec::new();
    let mut since: Option<tokio::time::Instant> = None;
    loop {
        if done(&got) {
            let s = *since.get_or_insert_with(tokio::time::Instant::now);
            if s.elapsed() >= quiet {
                break;
            }
        }
        let until = since.map_or(deadline, |s| deadline.min(s + quiet));
        match tokio::time::timeout_at(until, rx.recv()).await {
            Ok(Some(a)) => got.push(a),
            Ok(None) => break,
            Err(_) if tokio::time::Instant::now() >= deadline => break,
            Err(_) => {}
        }
    }
    acceptor.abort();
    got
}

/// A raw -23 SUBSCRIBE_TRACKS on a stream of its own, and the relay's answer; the stream is
/// returned to be held open.
async fn subscribe_tracks23(conn: &SharedConnection, request_id: u64, prefix: joule_moq::TrackNamespace, params: m23::Parameters) -> (joule_moq_relay::transport::BoxSend, BoxRecv, m23::ControlMessage) {
    let (mut tx, mut rx) = conn.open_bi().await.unwrap();
    tx.write_all(&m23::ControlMessage::SubscribeTracks(m23::NamespaceRequest { request_id, prefix, params }).encode().unwrap()).await.unwrap();
    let mut buf = Vec::new();
    let answer = read23(&mut rx, &mut buf).await;
    (tx, rx, answer)
}

/// The relay's next PUBLISH to a raw -23 session, within `wait`, accepted with REQUEST_OK; its
/// stream returned to be held open. `None` if none comes.
async fn publish23(conn: &SharedConnection, wait: Duration) -> Option<(m23::Publish, joule_moq_relay::transport::BoxSend, BoxRecv)> {
    let (mut tx, mut rx) = timeout(wait, conn.accept_bi()).await.ok()?.ok()?;
    let mut buf = Vec::new();
    let m = read23(&mut rx, &mut buf).await;
    let m23::ControlMessage::Publish(p) = m else { panic!("the relay opened {m:?} where a PUBLISH was expected") };
    tx.write_all(&m23::ControlMessage::RequestOk(m23::RequestOk { params: m23::Parameters::new(), track_properties: Default::default() }).encode().unwrap()).await.unwrap();
    Some((p, tx, rx))
}

/// One object of group `g` of the track a -22 publisher published under `alias`.
async fn object22(c: &Client, alias: u64, g: u64, payload: &[u8]) {
    let h = SubgroupHeader { track_alias: alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(128), properties: false, end_of_group: false, first_object: true };
    c.send_subgroup(h, &[SubgroupObject::normal(0, payload.to_vec())]).await.expect("the object goes");
}

/// The groups whose objects arrived on the subscription `alias`, by their payloads.
fn payloads_of(got: &[Arrived23], alias: u64) -> Vec<Vec<u8>> {
    let mut v: Vec<Vec<u8>> = got
        .iter()
        .filter(|a| matches!(&a.header, Some(m23::StreamHeader::Subgroup(h)) if h.track_alias == alias))
        .flat_map(|a| a.objects.iter().map(|o| o.payload.clone()))
        .collect();
    v.sort();
    v
}

/// -23 §3.6.2, §9.17: the subscription parameters a SUBSCRIBE_TRACKS carries that -22's has no
/// place for (both delivery timeouts, SUBSCRIBER_PRIORITY, LOCATION_FILTER), carried beside it by
/// the bridge (`Connection::carried`), are "the initial Subscription parameters when a PUBLISH is
/// sent as a result of SUBSCRIBE_TRACKS", and the relay "includes those in the PUBLISH": each is
/// in the PUBLISH, with SUBSCRIBE_TRACKS_ID, and the Location filter is where the subscription
/// starts, so groups before it are not sent. A request that carries none of them brings a
/// PUBLISH that states none.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_23_subscribe_tracks_carries_its_subscription_parameters() {
    timeout(LIMIT, async {
        let server = relay(|_| {}).await;
        let (publ, _pe) = connect(&server, Via::Quic, ClientOptions::default()).await;
        let (_p, _prx) = publ.publish(ns(&["carry", "a"]), name("t"), 61, Parameters::new(), Default::default()).await.expect("PUBLISH");
        object22(&publ, 61, 0, b"g0").await;
        let key = (ns(&["carry", "a"]), name("t"));
        assert!(until(STEP, || server.relay().largest(&key) == Some(Location::new(0, 0))).await, "the relay holds group 0");

        let (conn, _ep, _s, _ctl) = setup23(&server).await;
        let filter = LocationFilter::AbsoluteStart { start: Location::new(3, 0) };
        let asked = m23::Parameters::new()
            .with(m23::Parameter::ObjectDeliveryTimeout(60_000))
            .with(m23::Parameter::SubgroupDeliveryTimeout(90_000))
            .with(m23::Parameter::SubscriberPriority(7))
            .with(m23::Parameter::LocationFilter(filter));
        let (_ttx, _trx, answer) = subscribe_tracks23(&conn, 0, ns(&["carry"]), asked).await;
        assert!(matches!(answer, m23::ControlMessage::RequestOk(_)), "{answer:?}");
        let (p, _ptx, _prx2) = publish23(&conn, STEP).await.expect("a PUBLISH of the track");
        let has = |ty: u64| p.params.get(ty).cloned();
        use m23::params::param;
        assert_eq!(has(param::OBJECT_DELIVERY_TIMEOUT), Some(m23::Parameter::ObjectDeliveryTimeout(60_000)), "{:?}", p.params);
        assert_eq!(has(param::SUBGROUP_DELIVERY_TIMEOUT), Some(m23::Parameter::SubgroupDeliveryTimeout(90_000)), "{:?}", p.params);
        assert_eq!(has(param::SUBSCRIBER_PRIORITY), Some(m23::Parameter::SubscriberPriority(7)), "{:?}", p.params);
        assert_eq!(p.params.location_filter(), Some(filter), "{:?}", p.params);
        assert_eq!(p.params.subscribe_tracks_id(), Some(0), "the PUBLISH names its request (§9.19.21)");
        assert_eq!(p.params.largest_object(), Some(Location::new(0, 0)), "and the Largest Object");

        // Groups 1 and 2 are before the filter's start, 3 and 4 in it.
        for g in 1..=4u64 {
            object22(&publ, 61, g, format!("g{g}").as_bytes()).await;
        }
        let got = data23(&conn, STEP, Duration::from_millis(500), |a| {
            let have = payloads_of(a, p.track_alias);
            have.contains(&b"g3".to_vec()) && have.contains(&b"g4".to_vec())
        })
        .await;
        assert_eq!(payloads_of(&got, p.track_alias), vec![b"g3".to_vec(), b"g4".to_vec()], "the subscription starts where its Location filter does");
        assert!(got.iter().all(|a| !a.reset && a.entries.is_empty()), "every stream whole, and no fill, which the request did not ask for: {got:?}");

        // A request that carries none of them: its PUBLISH states none.
        let (conn2, _ep2, _s2, _ctl2) = setup23(&server).await;
        let (_ttx2, _trx2, answer) = subscribe_tracks23(&conn2, 0, ns(&["carry"]), m23::Parameters::new()).await;
        assert!(matches!(answer, m23::ControlMessage::RequestOk(_)), "{answer:?}");
        let (q, _qtx, _qrx) = publish23(&conn2, STEP).await.expect("a PUBLISH of the track");
        for ty in [param::OBJECT_DELIVERY_TIMEOUT, param::SUBGROUP_DELIVERY_TIMEOUT, param::SUBSCRIBER_PRIORITY, param::LOCATION_FILTER, param::FILL_PARAMETERS] {
            assert!(q.params.get(ty).is_none(), "{ty:#x} in a PUBLISH whose request carried none: {:?}", q.params);
        }
        publ.close(0, "done");
        server.shutdown().await;
    })
    .await
    .expect("in time");
}

/// -23 §3.4, §3.4.1, §3.6.2, §9.19.15: FILL_PARAMETERS in a SUBSCRIBE_TRACKS, which the relay
/// refused NOT_SUPPORTED before. Each PUBLISH the request brings carries them, and opens a fill
/// fetch stream whose FETCH_HEADER names the PUBLISH's Request ID, carrying the fill range (the
/// Location filter inside them, here the whole track to its Largest Object); the subscription's
/// own objects follow on its subgroups. The fill ends at the Largest Object the PUBLISH stated,
/// and the subscription starts at the Next Object after it, so an object published once the
/// PUBLISH has come, while the fill may still be begun or served, is the subscription's alone
/// ("a subscriber that wants each Object delivered exactly once uses the Next Object
/// Subscription Location Filter coupled with an open-ended fill range, which the publisher will
/// end at Largest Object", §3.4). A request that begins its subscriptions paused (PAUSED 1) has
/// its PUBLISHes carry the parameters and open no fill ("FILL_PARAMETERS carried while the
/// subscription is paused opens no fill fetch stream").
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_23_subscribe_tracks_with_fill_parameters() {
    timeout(LIMIT, async {
        let server = relay(|_| {}).await;
        let (publ, _pe) = connect(&server, Via::Quic, ClientOptions::default()).await;
        let tname = name("t");
        let (_p, _prx) = publ.publish(ns(&["fill", "a"]), tname.clone(), 62, Parameters::new(), Default::default()).await.expect("PUBLISH");
        for g in 0..3u64 {
            object22(&publ, 62, g, format!("f{g}").as_bytes()).await;
        }
        let key = (ns(&["fill", "a"]), tname);
        assert!(until(STEP, || server.relay().largest(&key) == Some(Location::new(2, 0))).await, "the relay holds groups 0 to 2");

        let (conn, _ep, _s, _ctl) = setup23(&server).await;
        let inner = m23::Parameters::new().with(m23::Parameter::LocationFilter(LocationFilter::AbsoluteStart { start: Location::new(0, 0) }));
        let asked = m23::Parameters::new().with(m23::Parameter::FillParameters(inner.clone()));
        let (_ttx, _trx, answer) = subscribe_tracks23(&conn, 0, ns(&["fill"]), asked).await;
        assert!(matches!(answer, m23::ControlMessage::RequestOk(_)), "SUBSCRIBE_TRACKS with FILL_PARAMETERS is accepted: {answer:?}");
        let (p, _ptx, _prx2) = publish23(&conn, STEP).await.expect("a PUBLISH of the track");
        assert_eq!(p.params.get(m23::params::param::FILL_PARAMETERS).cloned(), Some(m23::Parameter::FillParameters(inner.clone())), "the PUBLISH carries them: {:?}", p.params);
        assert_eq!(p.params.subscribe_tracks_id(), Some(0));
        assert_eq!(p.params.largest_object(), Some(Location::new(2, 0)), "the PUBLISH states the Largest Object its fill ends at");
        // Published after the PUBLISH, while the fill may be in flight: the subscription's.
        object22(&publ, 62, 3, b"f3").await;
        let fill_of = |a: &Arrived23| matches!(&a.header, Some(m23::StreamHeader::Fetch(h)) if h.request_id == p.request_id);
        let got = data23(&conn, STEP, Duration::from_millis(300), |a| a.iter().any(fill_of) && payloads_of(a, p.track_alias).contains(&b"f3".to_vec())).await;
        let fill = got.iter().find(|a| fill_of(a)).unwrap_or_else(|| panic!("a fill fetch stream named by the PUBLISH's Request ID: {got:?}"));
        assert!(!fill.reset, "the fill is served, not reset");
        let filled: Vec<(u64, Vec<u8>)> = fill.entries.iter().filter_map(|e| if let m23::FetchEntry::Object(o) = e { Some((o.group_id, o.payload.clone())) } else { None }).collect();
        assert_eq!(filled, vec![(0, b"f0".to_vec()), (1, b"f1".to_vec()), (2, b"f2".to_vec())], "the fill range: the track to the Largest Object its PUBLISH stated");
        assert_eq!(payloads_of(&got, p.track_alias), vec![b"f3".to_vec()], "the subscription's own objects, from the Next Object");

        // Paused: the PUBLISH carries FILL_PARAMETERS and no fill is opened.
        let (conn2, _ep2, _s2, _ctl2) = setup23(&server).await;
        let paused = m23::Parameters::new().with(m23::Parameter::Paused(true)).with(m23::Parameter::FillParameters(inner.clone()));
        let (_ttx2, _trx2, answer) = subscribe_tracks23(&conn2, 0, ns(&["fill"]), paused).await;
        assert!(matches!(answer, m23::ControlMessage::RequestOk(_)), "{answer:?}");
        let (q, _qtx, _qrx) = publish23(&conn2, STEP).await.expect("a PUBLISH of the track");
        assert_eq!(q.params.paused(), Some(true));
        assert!(q.params.get(m23::params::param::FILL_PARAMETERS).is_some(), "{:?}", q.params);
        let got = data23(&conn2, Duration::from_millis(1_500), Duration::ZERO, |_| false).await;
        assert!(got.iter().all(|a| !matches!(a.header, Some(m23::StreamHeader::Fetch(_)))), "no fill for a paused subscription: {got:?}");
        publ.close(0, "done");
        server.shutdown().await;
    })
    .await
    .expect("in time");
}

/// -23 §3.4 for a raw -23 SUBSCRIBE from the Next Object with FILL_PARAMETERS whose Location
/// filter is open-ended: the fill fetch stream, named by the SUBSCRIBE's Request ID, ends at the
/// Largest Object the SUBSCRIBE_OK stated, and an object published once SUBSCRIBE_OK has come,
/// while the fill may still be begun or served, is the subscription's alone, each Object coming
/// once ("a subscriber that wants each Object delivered exactly once uses the Next Object
/// Subscription Location Filter coupled with an open-ended fill range, which the publisher will
/// end at Largest Object").
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_23_subscribes_fill_ends_at_the_largest_object_its_subscribe_ok_states() {
    timeout(LIMIT, async {
        let server = relay(|_| {}).await;
        let (publ, _pe) = connect(&server, Via::Quic, ClientOptions::default()).await;
        let key = (ns(&["fill", "s"]), name("t"));
        let (_p, _prx) = publ.publish(key.0.clone(), key.1.clone(), 63, Parameters::new(), Default::default()).await.expect("PUBLISH");
        for g in 0..3u64 {
            object22(&publ, 63, g, format!("s{g}").as_bytes()).await;
        }
        assert!(until(STEP, || server.relay().largest(&key) == Some(Location::new(2, 0))).await, "the relay holds groups 0 to 2");

        let (conn, _ep, _s, _ctl) = setup23(&server).await;
        let inner = m23::Parameters::new().with(m23::Parameter::LocationFilter(LocationFilter::AbsoluteStart { start: Location::new(0, 0) }));
        let asked = m23::Parameters::new().with(m23::Parameter::LocationFilter(LocationFilter::NextObject)).with(m23::Parameter::FillParameters(inner));
        let (mut tx, mut rx) = conn.open_bi().await.unwrap();
        tx.write_all(&m23::ControlMessage::Subscribe(m23::TrackRequest { request_id: 0, namespace: key.0.clone(), name: key.1.clone(), params: asked }).encode().unwrap()).await.unwrap();
        let mut buf = Vec::new();
        let answer = read23(&mut rx, &mut buf).await;
        let m23::ControlMessage::SubscribeOk(ok) = answer else { panic!("SUBSCRIBE with FILL_PARAMETERS is accepted: {answer:?}") };
        assert_eq!(ok.params.largest_object(), Some(Location::new(2, 0)), "the SUBSCRIBE_OK states the Largest Object its fill ends at");
        // Published after the SUBSCRIBE_OK, while the fill may be in flight: the subscription's.
        object22(&publ, 63, 3, b"s3").await;
        let fill_of = |a: &Arrived23| matches!(&a.header, Some(m23::StreamHeader::Fetch(h)) if h.request_id == 0);
        let got = data23(&conn, STEP, Duration::from_millis(300), |a| a.iter().any(fill_of) && payloads_of(a, ok.track_alias).contains(&b"s3".to_vec())).await;
        let fill = got.iter().find(|a| fill_of(a)).unwrap_or_else(|| panic!("a fill fetch stream named by the SUBSCRIBE's Request ID: {got:?}"));
        assert!(!fill.reset, "the fill is served, not reset");
        let filled: Vec<(u64, Vec<u8>)> = fill.entries.iter().filter_map(|e| if let m23::FetchEntry::Object(o) = e { Some((o.group_id, o.payload.clone())) } else { None }).collect();
        assert_eq!(filled, vec![(0, b"s0".to_vec()), (1, b"s1".to_vec()), (2, b"s2".to_vec())], "the fill range: the track to the Largest Object its SUBSCRIBE_OK stated");
        assert_eq!(payloads_of(&got, ok.track_alias), vec![b"s3".to_vec()], "the subscription's own objects, from the Next Object");
        drop((tx, rx));
        publ.close(0, "done");
        server.shutdown().await;
    })
    .await
    .expect("in time");
}