open-standards/joule-code/joule-code-rs/crates/joule-moq-relay/tests/fetch_unanswered.rs

//! A FETCH no upstream answers, of a track the relay holds part of (MOQT -23 §3.2.3: "If an
//! upstream FETCH fails, the relay indicates the range of Unknown Objects [...] and continues
//! serving other known Objects"). A track that starts at {5, 1}, fetched from {0, 0}: the
//! relay cannot say the Locations before its first object do not exist, so it asks the
//! publisher, which published by PUBLISH and answers no FETCH. The request was refused
//! `TIMEOUT` `no_upstream`, and a subscriber had to know the track's first Location to
//! fetch it at all.

mod common;

use common::*;
use joule_moq::draft22::data::{FetchEntry, RangeEnd, SubgroupHeader, SubgroupId, SubgroupObject};
use joule_moq::draft22::{GroupOrder, Location, LocationFilter, Parameter, Parameters};
use joule_moq_relay::client::ClientOptions;
use std::time::Duration;
use tokio::time::timeout;

fn header(alias: u64, g: u64) -> SubgroupHeader {
    SubgroupHeader { track_alias: alias, group_id: g, subgroup_id: SubgroupId::Explicit(0), publisher_priority: Some(64), properties: false, end_of_group: false, first_object: true }
}

/// What a fetch stream said, entry by entry: an object's Location and payload, or the end
/// of a range that is not known.
fn said(entries: &[FetchEntry]) -> Vec<String> {
    entries
        .iter()
        .map(|e| match e {
            FetchEntry::Object(o) => format!("{}/{} {}", o.group_id, o.object_id, String::from_utf8_lossy(&o.payload)),
            FetchEntry::EndOfRange { kind: RangeEnd::Unknown, location } if location.object == u64::MAX => format!("unknown through group {}", location.group),
            FetchEntry::EndOfRange { kind: RangeEnd::Unknown, location } => format!("unknown through {}/{}", location.group, location.object),
            FetchEntry::EndOfRange { kind, location } => format!("{kind:?} through {}/{}", location.group, location.object),
        })
        .collect()
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_fetch_no_upstream_answers_is_served_what_the_relay_holds() {
    timeout(Duration::from_secs(120), async {
        let server = relay(|c| c.limits.upstream_timeout_ms = 300).await;
        let (space, track) = (ns(&["later"]), name("start"));
        let key = (space.clone(), track.clone());
        let (p, _ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
        // By PUBLISH: the publisher reads no request of the relay's, so a FETCH sent to it
        // is never answered.
        let _held = p.publish(space.clone(), track.clone(), 9, Parameters::new(), Default::default()).await.expect("PUBLISH");
        p.send_subgroup(header(9, 5), &[SubgroupObject::normal(1, "five-one"), SubgroupObject::normal(2, "five-two")]).await.expect("group 5");
        p.send_subgroup(header(9, 7), &[SubgroupObject::normal(0, "seven-zero"), SubgroupObject::normal(1, "seven-one")]).await.expect("group 7");
        let relay = server.relay().clone();
        assert!(until(STEP, || relay.largest(&key) == Some(Location::new(7, 1)) && relay.cache_covers(&key, Location::new(7, 0))).await, "the relay holds the track");

        let (s, _es) = connect(&server, Via::Quic, ClientOptions::default()).await;
        // From {0, 0}, the default: what the relay does not hold is not known; what it holds
        // is served. Group 5 is held from its object 1, and nothing says object 0 does not exist.
        let (ok, entries) = s.fetch(space.clone(), track.clone(), Parameters::new()).await.expect("FETCH from 0/0 is answered");
        assert_eq!(ok.end_location, Location::new(7, 1));
        assert_eq!(said(&entries), ["unknown through group 4", "unknown through 5/0", "5/1 five-one", "5/2 five-two", "unknown through group 6", "7/0 seven-zero", "7/1 seven-one"]);
        // From the track's first Location, which a subscriber used to have to know.
        let from = Parameters::new().with(Parameter::LocationFilter(LocationFilter::AbsoluteStart { start: Location::new(5, 1) }));
        let (_, entries) = s.fetch(space.clone(), track.clone(), from).await.expect("FETCH from 5/1");
        assert_eq!(said(&entries), ["5/1 five-one", "5/2 five-two", "unknown through group 6", "7/0 seven-zero", "7/1 seven-one"]);
        // Descending: the groups in that order, each group's objects ascending.
        let down = Parameters::new().with(Parameter::GroupOrder(GroupOrder::Descending));
        let (_, entries) = s.fetch(space.clone(), track.clone(), down).await.expect("FETCH, descending");
        assert_eq!(said(&entries), ["7/0 seven-zero", "7/1 seven-one", "unknown through group 6", "unknown through 5/0", "5/1 five-one", "5/2 five-two", "unknown through group 0"]);
        let stats = relay.stats();
        assert_eq!((stats.fetch_served_unanswered, stats.no_upstream), (3, 3), "each counted as unanswered, and as served from what the relay held");

        // A range that starts after the Largest Object is still INVALID_RANGE, and a track the
        // relay holds nothing of, with no one to ask, still does not exist.
        let past = Parameters::new().with(Parameter::LocationFilter(LocationFilter::AbsoluteStart { start: Location::new(8, 0) }));
        let e = s.fetch(space.clone(), track.clone(), past).await.expect_err("past the Largest Object");
        assert!(e.contains("INVALID_RANGE"), "{e}");
        let e = s.fetch(ns(&["nowhere"]), track.clone(), Parameters::new()).await.expect_err("no such track");
        assert!(e.contains("DOES_NOT_EXIST"), "{e}");
        for c in [&p, &s] {
            c.close(0, "done");
        }
        server.shutdown().await;
    })
    .await
    .expect("in time");
}