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

//! How long a client's request waits is the client's to state (`ClientOptions::request_wait`).
//! A relay holds a SUBSCRIBE of a track nobody publishes for the RENDEZVOUS_TIMEOUT it carries
//! before it answers; a session that waits less gives up with the answer still to come, and one
//! that waits longer has the relay's answer. The default is the client's ten seconds.
//!
//! This is the wait a session over a path that drops packets stretches
//! (`joule-moq-room`'s synthetic participants): there the answer is late because the request
//! or its response was dropped and QUIC is recovering it.
//!
//! And the relay's half: a request stream whose first message comes late is still answered.
//! QUIC opens every stream below the highest one a packet has named (RFC 9000 §3.2), so a
//! request whose first packet was lost is open at the relay, empty, from the moment a later
//! request arrives. The relay gave such a stream ten seconds and dropped it. Here the delay
//! is the client's own: it opens two request streams, writes the second, and writes the
//! first twelve seconds later.

mod common;

use common::*;
use joule_moq::draft22::{ControlMessage, NamespaceRequest, Parameter, Parameters};
use joule_moq_relay::wire::MessageReader;
use joule_moq_relay::client::{self, ClientOptions};
use std::time::{Duration, Instant};
use tokio::time::timeout;

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_request_waits_as_long_as_its_session_states() {
    timeout(Duration::from_secs(60), async {
        assert_eq!(ClientOptions::default().request_wait, client::STEP);
        let server = relay(|_| {}).await;
        // The relay answers after 1.5 s, that nobody publishes the track.
        let late = || Parameters::new().with(Parameter::RendezvousTimeout(1500));
        let (brief, _e1) = connect(&server, Via::Quic, ClientOptions { request_wait: Duration::from_millis(300), ..ClientOptions::default() }).await;
        let began = Instant::now();
        let Err(gave_up) = brief.subscribe(ns(&["nobody"]), name("here"), late()).await else { panic!("no answer within the session's wait") };
        assert_eq!(gave_up, "no response in time");
        assert!(began.elapsed() >= Duration::from_millis(300) && began.elapsed() < Duration::from_millis(1400), "it waited what it stated: {:?}", began.elapsed());

        let (patient, _e2) = connect(&server, Via::Quic, ClientOptions { request_wait: Duration::from_secs(20), ..ClientOptions::default() }).await;
        let began = Instant::now();
        let Err(answered) = patient.subscribe(ns(&["nobody"]), name("here"), late()).await else { panic!("the relay refuses a track nobody publishes") };
        assert!(answered.starts_with("REQUEST_ERROR"), "the relay's own answer: {answered}");
        assert!(began.elapsed() >= Duration::from_millis(1400), "which came after the rendezvous wait: {:?}", began.elapsed());

        for c in [&brief, &patient] {
            c.close(0, "done");
        }
        server.shutdown().await;
    })
    .await
    .expect("the test finishes in time");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_request_whose_first_message_comes_late_is_answered() {
    timeout(Duration::from_secs(120), async {
        let server = relay(|_| {}).await;
        let (c, _e) = connect(&server, Via::Quic, ClientOptions::default()).await;
        let request = |prefix: &str| ControlMessage::PublishNamespace(NamespaceRequest { request_id: c.next_request_id(), prefix: ns(&[prefix]), params: Parameters::new() }).encode().expect("a request");
        // Two request streams, in the order of their IDs. Nothing of the first is sent yet;
        // the second's bytes open both at the relay.
        let (mut first_tx, first_rx) = c.conn.open_bi().await.expect("a stream");
        let (mut second_tx, second_rx) = c.conn.open_bi().await.expect("a stream");
        second_tx.write_all(&request("early")).await.expect("written");
        let mut second = MessageReader::new(second_rx);
        assert!(matches!(timeout(STEP, second.next()).await, Ok(Ok(Some(ControlMessage::RequestOk(_))))), "the second request is answered");
        // Longer than the ten seconds the relay waited.
        tokio::time::sleep(Duration::from_secs(12)).await;
        first_tx.write_all(&request("late")).await.expect("the relay has not dropped the stream");
        let mut first = MessageReader::new(first_rx);
        match timeout(STEP, first.next()).await {
            Ok(Ok(Some(ControlMessage::RequestOk(_)))) => {}
            other => panic!("the late request is answered as any: {other:?}"),
        }
        c.close(0, "done");
        server.shutdown().await;
    })
    .await
    .expect("the test finishes in time");
}