open-standards/joule-code/joule-code-rs/crates/joule-moq-room/tests/participants.rs

//! Synthetic participants (`joule_moq_room::participant`) against a relay and a room service on
//! loopback, each reached as `wai dev`'s are: the service over HTTP, the relay over its WebSocket
//! binding or its QUIC binding through an impaired path.
//!
//! - a participant that publishes with its receipts is served by a relay that holds each
//!   publisher's tracks to its own key; one that publishes without them is dropped there, its
//!   lifecycle still signed by the clock. The same receipt-less participant on a relay with no door
//!   is served: the drop is the door's;
//! - impairment changes the loss QUIC measures on the path: none on loopback without it, at least
//!   a twentieth of the packets with a fifth of the datagrams dropped, every object still arriving,
//!   later;
//! - a load of viewers on one publisher is delivered in full, and the session the participants read
//!   names the room the service serves.

use ed25519_dalek::SigningKey;
use joule_moq_room::dev::{DevRoom, Fixture};
use joule_moq_room::participant::{Impairment, Participants};
use serde_json::{json, Value};
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::time::timeout;

async fn grant(room: &DevRoom, key: &SigningKey, role: &str) -> Vec<u8> {
    let mut s = tokio::net::TcpStream::connect(room.http).await.unwrap();
    let b = json!({"principal": hex::encode(key.verifying_key().to_bytes()), "role": role}).to_string();
    let req = format!("POST /grant HTTP/1.1\r\nHost: x\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{b}", b.len());
    s.write_all(req.as_bytes()).await.unwrap();
    let mut out = Vec::new();
    s.read_to_end(&mut out).await.unwrap();
    let text = String::from_utf8(out).unwrap();
    let v: Value = serde_json::from_str(text.split_once("\r\n\r\n").unwrap().1).unwrap();
    hex::decode(v["grant"].as_str().unwrap_or_else(|| panic!("a grant: {v}"))).unwrap()
}

async fn joins(room: &DevRoom, p: &Participants, seed: u8, receipts: bool) -> Value {
    let key = SigningKey::from_bytes(&[seed; 32]);
    let g = grant(room, &key, "publisher").await;
    p.join([seed; 32], g, 2, receipts).await.unwrap_or_else(|e| panic!("participant {seed} joins: {e}"))
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_participant_without_receipts_is_dropped_by_the_relay() {
    timeout(Duration::from_secs(240), async {
        let room = DevRoom::start_with(&Fixture { doors: true, ..Fixture::default() }).await.unwrap();
        let p = Participants::new(&format!("http://{}", room.http), room.relay_path(None)).unwrap();
        let alice = joins(&room, &p, 0x41, true).await;
        assert_eq!(alice["events"], json!(["join", "publish", "publish"]), "{alice}");
        assert_eq!(alice["receipts"], true);
        assert!(alice["objects_served"].as_u64().unwrap() >= 2, "a receipted participant is served: {alice}");
        let bob = joins(&room, &p, 0x42, false).await;
        assert_eq!(bob["events"], json!(["join", "publish", "publish"]), "the clock still signs its lifecycle: {bob}");
        assert_eq!(bob["objects_served"], 0, "the relay dropped what it published without receipts: {bob}");

        // The session names the room the service serves, both present.
        let s = p.session().await.unwrap();
        assert_eq!(s["room"], alice["room"]);
        assert!(s["chains"][0].as_array().unwrap().len() >= 6, "{s}");
        let left = p.leave(<[u8; 32]>::try_from(hex::decode(bob["principal"].as_str().unwrap()).unwrap()).unwrap()).await.unwrap();
        assert!(left["events"].as_array().unwrap().iter().any(|e| e == "leave"), "{left}");
        p.stop().await;
        room.stop().await;

        // The positive control: the same participant on a relay with no door is served.
        let open = DevRoom::start_with(&Fixture::default()).await.unwrap();
        let p = Participants::new(&format!("http://{}", open.http), open.relay_path(None)).unwrap();
        let bob = joins(&open, &p, 0x42, false).await;
        assert!(bob["objects_served"].as_u64().unwrap() >= 2, "with no door, nothing drops it: {bob}");
        p.stop().await;
        open.stop().await;
    })
    .await
    .expect("the test finishes in time");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn impairment_changes_the_measured_loss() {
    timeout(Duration::from_secs(300), async {
        let room = DevRoom::start_with(&Fixture { doors: true, quic: true, ..Fixture::default() }).await.unwrap();
        let service = format!("http://{}", room.http);
        let run = |impair: Option<Impairment>| {
            let p = Participants::new(&service, room.relay_path(impair)).unwrap();
            async move {
                let r = p.test(120, 5).await.unwrap_or_else(|e| panic!("a test over {impair:?}: {e}"));
                p.stop().await;
                r
            }
        };
        // Over the WebSocket binding: every object, and no QUIC path to read.
        let ws = run(None).await;
        assert_eq!((ws["transport"].as_str(), ws["objects_received"].as_u64(), &ws["path"]), (Some("websocket"), Some(120), &Value::Null), "{ws}");

        // Over QUIC with nothing impaired: every object, and on loopback no loss to speak of.
        let clear = run(Some(Impairment::default())).await;
        assert_eq!((clear["transport"].as_str(), clear["objects_received"].as_u64()), (Some("quic"), Some(120)), "{clear}");
        let (sent, lost) = (clear["path"]["packets_sent"].as_u64().unwrap(), clear["path"]["packets_lost"].as_u64().unwrap());
        assert!(sent > 120 && lost * 100 <= sent, "under 1% lost unimpaired: {clear}");
        assert_eq!(clear["impaired"]["datagrams_dropped"], 0, "{clear}");

        // A fifth of the datagrams dropped each way, and 20 ms of delay: QUIC sees the loss, and
        // retransmits until every object has arrived.
        let lossy = run(Some(Impairment { loss_ppm: 200_000, delay_ms: 10, jitter_ms: 0, seed: 7 })).await;
        // An object that does not arrive is accounted for: where it stands at the viewer
        // (`undelivered`), and what the relay did with what it was sent (its door refuses an
        // object whose receipts are late, and forwards it to nobody).
        let relay = room.relay.relay().stats();
        assert_eq!(lossy["objects_received"].as_u64(), Some(120), "every object arrives: {lossy}\nthe relay: {relay:?}");
        assert_eq!((relay.refused_at_door, &lossy["undelivered"]), (0, &Value::Null), "the door refused nothing: {lossy}");
        assert!(lossy["impaired"]["datagrams_dropped"].as_u64().unwrap() > 0, "{lossy}");
        let loss = lossy["path"]["loss"].as_f64().unwrap();
        assert!(loss >= 0.05, "the measured loss follows the impairment: {lossy}");
        assert!(loss > clear["path"]["loss"].as_f64().unwrap(), "{clear} {lossy}");
        assert!(lossy["path"]["rtt_ms"].as_f64().unwrap() >= 15.0, "the round trip carries the delay: {lossy}");
        room.stop().await;
    })
    .await
    .expect("the test finishes in time");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_load_of_viewers_on_one_publisher_is_delivered() {
    timeout(Duration::from_secs(240), async {
        let room = DevRoom::start_with(&Fixture { doors: true, ..Fixture::default() }).await.unwrap();
        let p = Participants::new(&format!("http://{}/", room.http), room.relay_path(None)).unwrap();
        let l = p.load(6, 30, 10).await.unwrap();
        assert_eq!((l["viewers"].as_u64(), l["deliveries_owed"].as_u64(), l["deliveries_made"].as_u64()), (Some(6), Some(180), Some(180)), "{l}");
        assert!(l["latency_ms"]["p99"].as_f64().is_some(), "{l}");
        p.stop().await;
        room.stop().await;
    })
    .await
    .expect("the test finishes in time");
}