//! 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 { 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"); }