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

//! Playout timing through the relay (`draft-nandakumar-moq-tempo-00` §2.3, §2.4, §3.5): the
//! relay declares that it reads the Sync Support parameter, and rewrites the HopTime of an
//! object it forwards with its own clock for a subscription that carried the parameter, at
//! -23 and at -22, leaving the object's Immutable Properties (its CaptureTimestamp and
//! PlayoutDelay) and its payload as they came.
//!
//! - a subscriber that asked receives a HopTime of the relay's clock, not the publisher's; one
//!   that did not, on the same track, receives the publisher's;
//! - a short HopTime (the draft's §4.1) is rewritten short;
//! - an object with no HopTime is given none;
//! - with `stamp = "always"` a subscriber at -18, which carries no such parameter, is stamped;
//! - with `stamp = "off"` the relay declares nothing, forwards HopTime as it came, and a
//!   SUBSCRIBE that carries the parameter all the same closes its session (an unknown
//!   parameter: INVALID_PARAMETER at -23);
//! - with another `hop_time_type` configured the default type's property is not touched.

mod common;

use common::*;
use joule_moq::draft::Draft;
use joule_moq::draft22::data::{SubgroupHeader, SubgroupId, SubgroupObject};
use joule_moq::draft22::kvp::{KeyValuePair, Properties};
use joule_moq::draft22::{varint, ObjectStatus, Parameter, Parameters};
use joule_moq::tempo::{TempoTypes, CAPTURE_TIMESTAMP, HOP_TIME, PLAYOUT_DELAY, TEMPO_OPTION};
use joule_moq_relay::client::{Client, ClientOptions};
use joule_moq_relay::config::HopStamp;
use joule_moq_relay::transport::TransportError;
use std::time::Duration;
use tokio::time::timeout;

/// The publisher's HopTime: a clock a day behind, so the relay's is told from it.
fn stale() -> u64 {
    joule_moq_relay::relay::now_unix_ms() - 86_400_000
}

fn vi(v: u64) -> Vec<u8> {
    let mut b = Vec::new();
    varint::put(&mut b, v);
    b
}

fn timed(hop: Option<u64>) -> Properties {
    let mut inner = Properties::new();
    inner.push(KeyValuePair::bytes(PLAYOUT_DELAY, 200u32.to_be_bytes().to_vec()));
    inner.push(KeyValuePair::bytes(CAPTURE_TIMESTAMP, vi(1_700_000_000_000)));
    let mut p = Properties::new();
    if let Some(h) = hop {
        p.push(KeyValuePair::bytes(HOP_TIME, vi(h)));
    }
    p.set_immutable(&inner).unwrap();
    p
}

const ALIAS: u64 = 7;

async fn send(p: &Client, g: u64, props: Properties) {
    let h = SubgroupHeader { track_alias: ALIAS, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(128), properties: true, end_of_group: true, first_object: true };
    p.send_subgroup(h, &[SubgroupObject { object_id: 0, properties: props, status: ObjectStatus::Normal, payload: format!("frame {g}").into_bytes() }]).await.expect("the object");
}

/// Groups 0..n as one subscriber received them: each object's properties.
async fn received(c: &Client, n: usize) -> Vec<Properties> {
    let mut got = collect(c, STEP, |g| g.len() >= n).await;
    assert_eq!(got.len(), n, "every group arrived");
    got.sort_by_key(|g| g.header.group_id);
    got.iter()
        .enumerate()
        .map(|(g, s)| {
            assert_eq!(s.objects[0].payload, format!("frame {g}").into_bytes(), "the payload as sent");
            s.objects[0].properties.clone()
        })
        .collect()
}

fn asking() -> Parameters {
    Parameters::new().with(Parameter::SyncSupport)
}

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

async fn run() {
    let types = TempoTypes::default();
    let space = ns(&["live", "timed"]);
    let track = name("video");

    // ── stamp = "asked" (the default) ──
    let server = relay(|_| {}).await;
    let (p, _pe) = connect(&server, Via::Quic, ClientOptions::default()).await;
    let _pub = p.publish(space.clone(), track.clone(), ALIAS, Parameters::new(), track_props(128, None)).await.unwrap();
    let (a23, _e1) = connect(&server, Via::QuicDraft(Draft::D23), ClientOptions::default()).await;
    let (a22, _e2) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
    let (plain, _e3) = connect(&server, Via::Quic, ClientOptions::default()).await;
    for c in [&a23, &a22, &plain] {
        assert!(c.peer_setup.options.extension(TEMPO_OPTION).is_some(), "the relay declares that it reads Sync Support");
    }
    let _s1 = a23.subscribe(space.clone(), track.clone(), asking()).await.expect("Sync Support at -23");
    let _s2 = a22.subscribe(space.clone(), track.clone(), asking()).await.expect("Sync Support at -22");
    let _s3 = plain.subscribe(space.clone(), track.clone(), Parameters::new()).await.unwrap();
    let old = stale();
    let before = joule_moq_relay::relay::now_unix_ms();
    send(&p, 0, timed(Some(old))).await;
    send(&p, 1, timed(Some(old & 16_383))).await; // a short timestamp
    send(&p, 2, timed(None)).await;
    for c in [&a23, &a22] {
        let got = received(c, 3).await;
        let after = joule_moq_relay::relay::now_unix_ms();
        let hop = types.hop_time(&got[0]).expect("a HopTime");
        assert!(hop >= before && hop <= after, "the relay's clock as it forwarded: {before} <= {hop} <= {after}");
        let short = types.hop_time(&got[1]).expect("a HopTime");
        assert!(short < 16_384, "a short timestamp stays short");
        assert!((before..=after).any(|t| t & 16_383 == short), "and is the relay's clock's low 14 bits");
        assert_eq!(types.hop_time(&got[2]), None, "the relay adds no HopTime of its own");
        for g in &got {
            assert_eq!(g.immutable().unwrap(), timed(None).immutable().unwrap(), "Immutable Properties as the publisher sent them");
        }
    }
    let got = received(&plain, 3).await;
    assert_eq!(types.hop_time(&got[0]), Some(old), "a subscriber that did not ask receives the publisher's HopTime");
    assert_eq!(got[0].immutable().unwrap(), timed(None).immutable().unwrap());
    for c in [&a23, &a22, &plain, &p] {
        c.close(0, "done");
    }
    server.shutdown().await;

    // ── stamp = "always": a subscriber at -18 has no parameter to ask with ──
    let server = relay(|c| c.tempo.stamp = HopStamp::Always).await;
    let (p, _pe) = connect(&server, Via::Quic, ClientOptions::default()).await;
    let _pub = p.publish(space.clone(), track.clone(), ALIAS, Parameters::new(), track_props(128, None)).await.unwrap();
    let (old18, _e) = connect(&server, Via::QuicDraft(Draft::D18), ClientOptions::default()).await;
    let _s = old18.subscribe(space.clone(), track.clone(), Parameters::new()).await.unwrap();
    let before = joule_moq_relay::relay::now_unix_ms();
    send(&p, 0, timed(Some(old))).await;
    let got = received(&old18, 1).await;
    assert!(types.hop_time(&got[0]).expect("a HopTime") >= before, "stamped without asking");
    old18.close(0, "done");
    p.close(0, "done");
    server.shutdown().await;

    // ── stamp = "off", and another type configured ──
    for other_type in [false, true] {
        let server = relay(|c| {
            if other_type {
                c.tempo.stamp = HopStamp::Always;
                c.tempo.hop_time_type = 0x3a7b;
            } else {
                c.tempo.stamp = HopStamp::Off;
            }
        })
        .await;
        let (p, _pe) = connect(&server, Via::Quic, ClientOptions::default()).await;
        let _pub = p.publish(space.clone(), track.clone(), ALIAS, Parameters::new(), track_props(128, None)).await.unwrap();
        let (c, _e) = connect(&server, Via::QuicDraft(Draft::D23), ClientOptions::default()).await;
        assert_eq!(c.peer_setup.options.extension(TEMPO_OPTION).is_none(), !other_type);
        let _s = c.subscribe(space.clone(), track.clone(), Parameters::new()).await.unwrap();
        send(&p, 0, timed(Some(old))).await;
        let got = received(&c, 1).await;
        assert_eq!(types.hop_time(&got[0]), Some(old), "HopTime forwarded as it came");
        if !other_type {
            // The parameter was not negotiated: an unknown one.
            let (bad, _e) = connect(&server, Via::QuicDraft(Draft::D23), ClientOptions::default()).await;
            let _ = bad.request(&joule_moq::draft22::ControlMessage::Subscribe(joule_moq::draft22::TrackRequest { request_id: bad.next_request_id(), namespace: space.clone(), name: track.clone(), params: asking() })).await;
            match timeout(STEP, bad.conn.closed()).await.expect("closed in time") {
                TransportError::Closed { code, .. } => assert_eq!(code, joule_moq::draft23::codes::SessionError::INVALID_PARAMETER.0),
                other => panic!("{other:?}"),
            }
        }
        c.close(0, "done");
        p.close(0, "done");
        server.shutdown().await;
    }
}