open-standards/joule-code/joule-code-rs/crates/joule-moq-room/tests/metrics.rs
Transaction Science 6250fa5a4e
Some checks failed
joule-moq-remote host agent / joule-moq-remote · does this change touch it (push) Successful in 33s
WAI operation catalogue / does this change touch it (push) Successful in 28s
wai-app and Create / wai-app · does this change touch it (push) Successful in 29s
joule-moq-room room service / joule-moq-room · does this change touch it (push) Successful in 30s
WAI Studio / does this change touch it (push) Successful in 16s
wai-sink-swift native sink package / does this change touch it (push) Successful in 1m32s
wai-quantum studio and desktop / does this change touch it (push) Successful in 1m34s
wai-fleet service / does this change touch it (push) Successful in 1m40s
wai-audit collector and player integration / does this change touch it (push) Successful in 1m39s
wai-ledger service / does this change touch it (push) Successful in 1m37s
wai-quantum for JavaScript / does this change touch it (push) Successful in 1m34s
joule-moq-web browser client / joule-moq-web · does this change touch it (push) Successful in 1m41s
wai-sink-android native sink library / does this change touch it (push) Successful in 1m35s
joule-moq-remote host agent / joule-moq-remote, every target (linux-arm64) (push) Successful in 2m11s
repository checks / lockfiles, advisories, tests in CI, workflow hygiene, spec and section references, public text (push) Failing after 4m29s
joule-code conformance gate / joule-moq-relay, every target (push) Successful in 7m32s
joule-code conformance gate / joule-moq-ingest, every target (push) Successful in 9m9s
wai-sink-android native sink library / the JNI layer and the platform-free half on the host JVM · linux-arm64 (push) Successful in 6m1s
joule-moq-room room service / joule-moq-room, every target (push) Successful in 8m1s
WAI breaking changes and the conformance runner / WAI breaking · does this change touch WAI (push) Successful in 11s
wai-app and Create / wai-app · tests, clippy, the line language's verifier (linux) (push) Successful in 45s
WAI Studio / the catalogue's actions, the unit tests and the core · linux (push) Successful in 1m22s
joule-code conformance gate / Build + round-trip C1-C4 vectors (push) Successful in 2m54s
WAI operation catalogue / the catalogue and every door · linux (push) Failing after 2m27s
wai-audit collector and player integration / collector · linux (push) Successful in 2m23s
wai-ledger service / service · linux (push) Successful in 3m55s
wai-fleet service / service · linux (push) Successful in 4m31s
wai-quantum for JavaScript / javascript · linux (push) Successful in 6m17s
wai-quantum studio and desktop / studio · browser (push) Successful in 7m43s
wai-quantum studio and desktop / desktop app · linux (push) Successful in 10m31s
WAI Studio / Studio in WebKit · linux (push) Successful in 22m31s
WAI breaking changes and the conformance runner / WAI breaking and conform · their tests, and the independent verifiers through wai conform · linux (push) Failing after 1m31s
WAI breaking changes and the conformance runner / WAI breaking · wire, record and API against main · linux (push) Successful in 3m35s
wai release binaries build on Windows / the seven release binaries · windows (push) Successful in 18m5s
wai-quantum studio and desktop / desktop app · windows (push) Successful in 10m46s
joule-code conformance gate / joule-moq-relay, its own energy on a counter (macos) (push) Successful in 1m3s
joule-moq-remote host agent / joule-moq-remote, every target (macos) (push) Successful in 56s
wai-app and Create / wai-app · Create, browser end-to-end (macos) (push) Successful in 7m44s
WAI Studio / Studio in Chromium and Firefox · macos (push) Successful in 9m17s
wai-fleet service / service · macos (push) Successful in 1m4s
wai-quantum studio and desktop / desktop app · macos (push) Successful in 4m41s
wai-ledger service / service · macos (push) Successful in 1m7s
wai-audit collector and player integration / collector, player integration and the browser end to end · macos (push) Successful in 2m35s
wai-sink-android native sink library / every ABI, the AAR, and the instrumented tests in a headless emulator · macos (push) Successful in 7m17s
wai-sink-swift native sink package / libwai slices, desktop tests, simulator tests · macos (push) Successful in 15m0s
joule-moq-web browser client / joule-moq-web · browser end-to-end (macos) (push) Successful in 15m50s
The relay reads the client's SETUP first (Privacy Pass auth -04), and what the draft-23 package left
SETUP order. draft-ietf-moq-privacy-pass-auth-04 §3.5.1 has a relay that
authenticates SETUP process the client's SETUP before it sends its own; WAI
age-assurance §5.1 and live-rooms §10.5.2 had the relay send first. The decision
was to follow -04, and the texts, the relay, the clients and the corpora do.

The relay's SETUP used to carry the challenge the client's SETUP then proved its
grant over, which the new order cannot do in one session. It is now -04's own
flow (§3.5.6.1, Figure 4), and live-rooms §10.5.3's over HTTP: every SETUP a
relay sends carries a fresh GRANT_CHALLENGE, good for the proofs of one later
SETUP at that relay for limits.grant_challenge_ms (60 s), kept by the relay
(src/challenge.rs) and consumed by the SETUP it admits. A client that holds none
presents its grant unproved, is refused pop_missing with a challenge, and proves
on a second session; one that holds the challenge of the relay's last SETUP to
it proves on its first. A refused SETUP is answered with the relay's SETUP
carrying a new Setup Option, GRANT_REFUSED (0x3A53, the reason's code), and the
session is closed UNAUTHORIZED once the client has left or
limits.setup_refusal_wait_ms (2 s) has passed: a close carries nothing sent
before it, which -04 notes of its own challenge, so the refusal is said in the
SETUP and the client that reads it leaves at once. A consumed challenge is
pop_replayed; one never issued, another relay's, or too old, pop_wrong_challenge.

The clients: joule_moq_relay::client (Client::open, connect_proving, challenge;
its SETUP always first), the room service's clock and participants, the remote
host agent, relay-to-relay links (a link's first SETUP is refused
relay_id_unproved and it dials again at once with its proof), and the browser
(MoqSession.connect's redial, SetupRefused). The interop client and the ingest
gateway already sent first and present no grant. The order is the same at every
draft; grants are proved at -23, -22, -18 and -16 in the relay's test. Not
interoperable with the order before where a grant is presented: a client that
waits for the relay's SETUP is sent none, and relays that peer change together.
Sessions that present no grant are unaffected.

What the draft-23 package left:
- A -23 session's subgroup streams and datagrams are read by -23's object rules
  (a supported property outside its scope or repeated makes the track malformed;
  a grease value in the Mandatory Track Property range is no Mandatory Track
  Property); a -22 session keeps -22's reading.
- Relays peer at -23: RELAY_PATH in the -23 codec and its bridge, a relay id
  taken at -23, links offering -23 and -22. An unnegotiated path closes a -23
  session INVALID_PARAMETER. The -23 corpus gains 6 vectors (125, 70 negative)
  and moqt23_check.py reads the extension.
- A grant that expires ends its subscriptions, each with PUBLISH_DONE
  EXPIRED_AUTH_TOKEN, and not the session, at -23; a -23 session that publishes
  by the grant, and every session at the older drafts, is closed as before.
- LARGEST_OBJECT against -23 §7.5: confirmed but for one source, a publisher's
  PUBLISH_STATE_NOTIFY, which is taken now; held by a vector of six steps. Kept
  and stated: an object counts once whole, not from its first byte.

Tests, each against planted faults: grant_proof (the relay sends nothing before
the client's SETUP; the second session; seven refusals, each with its reason),
room_authority (expiry at -23 beside -22), draft23 (the object rules against the
same objects at -22; the LARGEST_OBJECT vector), cascade and control_sessions
(a link at -23, opened on its second dial; the path at -23), the browser
session's unit tests. Falsified: twelve of the changes reverted in turn, each
failing a test.

live-rooms §10.5.2, §10.5.4, §10.7 and §13, and age-assurance §5.1 and §5.5,
say the order and what a challenge is bound to now. The rooms corpus renames one
proof vector (acknowledged in the change note); the registry lists
GRANT_REFUSED.

Co-Authored-By: Claude <[email protected]>
2026-11-10 17:22:02 -05:00

428 lines
24 KiB
Rust

//! `GET /metrics` of the room service beside a relay on loopback: it reads strictly as the text
//! format; its counters move with the requests made (grants by role, requests by route, method
//! and status) and its gauges with the room, whatever its kind (a publisher that enters and
//! leaves, a room with co-performance's audience, a watch room's viewer that joins beside a
//! publisher counted once, a remote desktop room's controllers and viewers, a conversation's
//! members; a room that is spent and the next one kept; the clock's session at the relay as it
//! ends); it answers `HEAD` as it does `GET`, without the text; and no room, principal, grant or
//! client address of those requests is in it.
use ed25519_dalek::SigningKey;
use joule_moq::draft22::kvp::{prop, KeyValuePair, Properties};
use joule_moq::draft22::Parameters;
use joule_moq::grant_proof::{self as gp, GrantProof};
use joule_moq::{TrackName, TrackNamespace};
use joule_moq_relay::client::{Client, ClientOptions};
use joule_moq_relay::config::{Authority, DoorRule, Grant};
use joule_moq_relay::transport::mux::MuxConfig;
use joule_moq_relay::transport::websocket;
use joule_moq_relay::{RelayConfig, RelayServer};
use joule_moq_room::keys::Keys;
use joule_moq_room::{http, remote, Config, Service};
use serde_json::{json, Value};
use sha2::{Digest, Sha256};
use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::time::timeout;
use wai_metrics::read::{identifier_like, label_values, parse, value, Family};
const STEP: Duration = Duration::from_secs(20);
async fn relay(keys: &Keys) -> RelayServer {
let mut c = RelayConfig::loopback();
c.quic_listen = None;
c.auth.required = true;
c.auth.authorities = vec![Authority { key: hex::encode(keys.authority.verifying_key().to_bytes()), rooms: vec![] }];
let t = hex::encode(Sha256::digest(keys.clock_token));
let clock = hex::encode(keys.clock.verifying_key().to_bytes());
c.auth.grants = vec![
Grant { prefix: vec!["*".into(), clock, "events".into()], token_sha256: t.clone(), publish: true, subscribe: false, proof: false, unlimited: true, seek: false },
Grant { prefix: vec![], token_sha256: t, publish: false, subscribe: true, proof: false, unlimited: false, seek: false },
];
c.receipts.door = ["audio", "video"].iter().map(|m| DoorRule { prefix: vec!["*".into(), "*".into(), m.to_string()], trust: vec![], trust_field: Some(1), hold_ms: 2_000 }).collect();
RelayServer::start(c).await.expect("the relay starts")
}
/// The service beside `server`, `rest` the end of its configuration, `start` starting its clock:
/// listening, its clock's session open.
async fn service(keys: &Keys, server: &RelayServer, rest: &str, start: impl FnOnce(&Arc<Service>)) -> (Arc<Service>, std::net::SocketAddr) {
let cfg = Config::from_toml(&format!("listen = \"127.0.0.1:0\"\nkeys = \"/nonexistent\"\nrelay_url = \"ws://{}{}\"\n{rest}", server.websocket_addr.unwrap(), server.relay().config().path)).unwrap();
let svc = Service::new(Keys::from_toml(&keys.to_toml()).unwrap(), &cfg);
let listener = tokio::net::TcpListener::bind(cfg.listen).await.unwrap();
let addr = listener.local_addr().unwrap();
start(&svc);
tokio::spawn(http::serve(svc.clone(), listener));
let up = timeout(STEP, async {
while !svc.connected.load(Ordering::Acquire) {
tokio::time::sleep(Duration::from_millis(50)).await;
}
})
.await;
assert!(up.is_ok(), "the clock's session opens");
(svc, addr)
}
/// One HTTP/1.1 request to the service from `client`, through the front that names it, or at the
/// service's own address when `client` is empty: its status, its head and its body.
async fn call(addr: std::net::SocketAddr, method: &str, path: &str, body: Option<&Value>, client: &str) -> (u16, String, String) {
request(addr, method, path, &[], body, client).await
}
/// [`call`], with `headers` beside the request's own.
async fn request(addr: std::net::SocketAddr, method: &str, path: &str, headers: &[(&str, String)], body: Option<&Value>, client: &str) -> (u16, String, String) {
let mut s = tokio::net::TcpStream::connect(addr).await.unwrap();
let b = body.map(|v| v.to_string()).unwrap_or_default();
let extra: String = headers.iter().map(|(k, v)| format!("{k}: {v}\r\n")).collect();
let front = if client.is_empty() { String::new() } else { format!("CF-Connecting-IP: {client}\r\n") };
let req = format!("{method} {path} HTTP/1.1\r\nHost: room.test\r\n{front}{extra}Content-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();
timeout(STEP, s.read_to_end(&mut out)).await.expect("an answer in time").unwrap();
let text = String::from_utf8(out).unwrap();
let (head, body) = text.split_once("\r\n\r\n").expect("a response");
assert!(head.to_ascii_lowercase().contains("cache-control: no-store"), "{head}");
(head.split(' ').nth(1).unwrap().parse().unwrap(), head.to_ascii_lowercase(), body.to_string())
}
async fn scrape(addr: std::net::SocketAddr) -> (String, Vec<Family>) {
let (status, head, text) = call(addr, "GET", "/metrics", None, "").await;
assert_eq!(status, 200, "{text}");
assert!(head.contains(&format!("content-type: {}", wai_metrics::CONTENT_TYPE)), "{head}");
let f = parse(&text).unwrap_or_else(|e| panic!("{e}\n{text}"));
(text, f)
}
/// Scrape until the gauge `name` reads `want`: the service follows a session by what the relay
/// tells it, a moment after the session changes.
async fn settles(addr: std::net::SocketAddr, name: &str, want: f64) {
let mut last = None;
let reached = timeout(STEP, async {
loop {
let (_, f) = scrape(addr).await;
last = value(&f, name, &[]);
if last == Some(want) {
return;
}
tokio::time::sleep(Duration::from_millis(50)).await;
}
})
.await;
assert!(reached.is_ok(), "{name} reads {want}, not {last:?}");
}
fn pubhex(k: &SigningKey) -> String {
hex::encode(k.verifying_key().to_bytes())
}
/// `POST /grant` for `k` as `role`, from `client`: the grant's answer.
async fn grant(addr: std::net::SocketAddr, k: &SigningKey, role: &str, client: &str) -> Value {
let (st, _, g) = call(addr, "POST", "/grant", Some(&json!({"principal": pubhex(k), "role": role})), client).await;
assert_eq!(st, 200, "{g}");
serde_json::from_str(&g).unwrap()
}
/// A session at the relay under `grant`, publishing `k`'s media tracks in `room`: the clock
/// follows it in.
async fn enter(server: &RelayServer, room: &str, grant: &[u8], k: &SigningKey) -> (Client, Vec<(joule_moq_relay::wire::ReqTx, joule_moq_relay::wire::ReqRx)>) {
let url = format!("ws://{}{}", server.websocket_addr.unwrap(), server.relay().config().path);
let dial = || async { Ok::<joule_moq_relay::transport::SharedConnection, String>(Arc::new(websocket::connect(&url, MuxConfig::default()).await.map_err(|e| e.to_string())?)) };
let opts = ClientOptions { grants: vec![(grant.to_vec(), k.clone())], target: format!("//{}{}", server.websocket_addr.unwrap(), server.relay().config().path), ..ClientOptions::default() };
// The relay reads this SETUP before it sends its own, so the grant is proved over a
// challenge its SETUP gave: on a second session, where the first held none.
let c = Client::connect_proving(dial, opts).await.expect("SETUP");
let mut held = Vec::new();
let mut alias = 10;
for m in ["audio", "video"] {
let ns = TrackNamespace::from_strs([room, pubhex(k).as_str(), m]);
for name in ["0_receipts", "0"] {
let mut p = Properties::new();
p.push(KeyValuePair::varint(prop::DEFAULT_PUBLISHER_PRIORITY, 64));
held.push(c.publish(ns.clone(), TrackName::from_str(name), alias, Parameters::new(), p).await.expect("PUBLISH"));
alias += 1;
}
}
(c, held)
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn metrics_move_with_grants_and_name_no_party() {
timeout(Duration::from_secs(120), run()).await.expect("the test finishes in time");
}
async fn run() {
let keys = Keys::generate();
let server = relay(&keys).await;
let (_svc, addr) = service(&keys, &server, "[room]\nmax_publishers = 2\njoin_wait_s = 30\n[rate]\nclient_burst = 8\nclient_every_s = 600\n", |s| s.start()).await;
let (_, f) = scrape(addr).await;
assert_eq!(value(&f, "wai_room_grants_total", &[("role", "viewer")]), Some(0.0));
assert_eq!(value(&f, "wai_room_grants_total", &[("role", "host")]), None, "this room grants no host");
assert_eq!(value(&f, "wai_room_relay_connected", &[]), Some(1.0));
assert_eq!(value(&f, "wai_room_rooms_total", &[]), Some(1.0));
assert_eq!(value(&f, "wai_room_publishers", &[]), Some(0.0));
// A publisher's grant and a viewer's, each from its own address; a request that is refused;
// the room; a path the service does not have, naming a principal; the scrape by another method.
let (alice, viewer) = (SigningKey::from_bytes(&[0x41; 32]), SigningKey::from_bytes(&[0x42; 32]));
let a = grant(addr, &alice, "publisher", "203.0.113.41").await;
let v = grant(addr, &viewer, "viewer", "203.0.113.42").await;
let (st, _, _) = call(addr, "POST", "/grant", Some(&json!({"principal": "00", "role": "publisher"})), "203.0.113.43").await;
assert_eq!(st, 400);
assert_eq!(call(addr, "GET", "/room", None, "203.0.113.41").await.0, 200);
assert_eq!(call(addr, "GET", &format!("/principal/{}", pubhex(&alice)), None, "203.0.113.41").await.0, 404);
let (st, head, _) = call(addr, "PUT", "/metrics", None, "").await;
assert_eq!(st, 405);
assert!(head.contains("allow: get, head"), "{head}");
let (st, head, nothing) = call(addr, "HEAD", "/metrics", None, "").await;
assert_eq!((st, nothing.as_str()), (200, ""));
assert!(head.contains(&format!("content-type: {}", wai_metrics::CONTENT_TYPE)), "{head}");
// Through a front, the scrape is a path the service does not have.
let (st, _, answer) = call(addr, "GET", "/metrics", None, "198.51.100.9").await;
assert_eq!(st, 404, "{answer}");
assert!(!answer.contains("wai_room"), "{answer}");
for (field, given) in [("X-Forwarded-For", "198.51.100.9"), ("X-Forwarded-Host", "room.test")] {
let (st, _, answer) = request(addr, "GET", "/metrics", &[(field, given.to_string())], None, "").await;
assert_eq!(st, 404, "{field}: {answer}");
}
let (text, f) = scrape(addr).await;
let req = |route: &str, method: &str, code: &str| value(&f, "wai_room_requests_total", &[("route", route), ("method", method), ("code", code)]);
assert_eq!(value(&f, "wai_room_grants_total", &[("role", "publisher")]), Some(1.0));
assert_eq!(value(&f, "wai_room_grants_total", &[("role", "viewer")]), Some(1.0));
assert_eq!((req("/grant", "POST", "200"), req("/grant", "POST", "400")), (Some(2.0), Some(1.0)));
assert_eq!((req("/room", "GET", "200"), req("other", "GET", "404")), (Some(1.0), Some(1.0)));
assert_eq!((req("/metrics", "GET", "200"), req("/metrics", "PUT", "405"), req("/metrics", "GET", "404")), (Some(1.0), Some(1.0), Some(3.0)));
assert_eq!(req("/metrics", "HEAD", "200"), Some(1.0));
assert_eq!(value(&f, "wai_room_publishers", &[]), Some(1.0), "granted, and waited for");
assert_eq!(value(&f, "wai_room_viewer_grants", &[]), Some(1.0));
assert_eq!(value(&f, "wai_room_present", &[]), Some(0.0), "nobody has entered");
// The publisher enters, and leaves: the room's gauges follow her.
let room = a["room"]["room"].as_str().unwrap().to_string();
let (session, held) = enter(&server, &room, &hex::decode(a["grant"].as_str().unwrap()).unwrap(), &alice).await;
settles(addr, "wai_room_present", 1.0).await;
let (text_present, f_present) = scrape(addr).await;
assert_eq!(value(&f_present, "wai_room_publishers", &[]), Some(1.0), "present, and holding her place");
drop(held);
session.close(0, "done");
settles(addr, "wai_room_present", 0.0).await;
let (_, f_left) = scrape(addr).await;
assert_eq!(value(&f_left, "wai_room_publishers", &[]), Some(0.0), "her place is given up");
assert_eq!(value(&f_left, "wai_room_rooms_total", &[]), Some(1.0));
// Her room is spent: the next publisher's grant starts another, and it is counted.
let next = grant(addr, &SigningKey::from_bytes(&[0x4b; 32]), "publisher", "203.0.113.51").await;
assert_ne!(next["room"]["room"], a["room"]["room"], "another room");
let (_, f_next) = scrape(addr).await;
assert_eq!(value(&f_next, "wai_room_rooms_total", &[]), Some(2.0));
// No party and no thing of one: not a label, and nowhere in the text.
let ids = [
pubhex(&alice),
pubhex(&viewer),
a["room"]["room"].as_str().unwrap().to_string(),
a["grant"].as_str().unwrap().to_string(),
a["digest"].as_str().unwrap().to_string(),
v["digest"].as_str().unwrap().to_string(),
hex::encode(keys.authority.verifying_key().to_bytes()),
hex::encode(keys.clock.verifying_key().to_bytes()),
"203.0.113.4".to_string(),
"198.51.100.9".to_string(),
"room.test".to_string(),
];
for f in [&f, &f_present] {
for l in label_values(f) {
assert!(!identifier_like(l), "a label value {l:?}");
}
}
for id in &ids {
assert!(!text.contains(id.as_str()) && !text_present.contains(id.as_str()), "{id} is in the metrics");
}
// The relay stops: the clock's session ends, and the gauge says so.
server.shutdown().await;
settles(addr, "wai_room_relay_connected", 0.0).await;
}
/// A room with co-performance: a viewer's grant is an audience member's, holding a place.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn the_audience_of_a_room_with_co_performance() {
timeout(Duration::from_secs(120), async {
let keys = Keys::generate();
let server = relay(&keys).await;
let rest = "[room]\nmax_publishers = 3\njoin_wait_s = 30\n[rate]\nclient_burst = 20\n[co]\naudience_input = 2\nlatency_budget_ms = 1500\nlead = \"first\"\nfidelity_every_s = 0\ncommit_after_ms = 800\nlead_ms = 200\n";
let (_svc, addr) = service(&keys, &server, rest, |s| s.start()).await;
let (_, f) = scrape(addr).await;
assert_eq!(value(&f, "wai_room_audience", &[]), Some(0.0));
grant(addr, &SigningKey::from_bytes(&[0x43; 32]), "viewer", "203.0.113.44").await;
let (_, f) = scrape(addr).await;
assert_eq!(value(&f, "wai_room_audience", &[]), Some(1.0), "granted, and waited for");
assert_eq!(value(&f, "wai_room_viewer_grants", &[]), Some(1.0));
assert_eq!(value(&f, "wai_room_publishers", &[]), Some(0.0));
server.shutdown().await;
})
.await
.expect("the test finishes in time");
}
/// A watch room: a viewer joins by a proof and is not among the room's members, a publisher is
/// both among them and followed as a viewer is. Each counts once, and a viewer's leave counts.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_watch_room_counts_the_viewers_who_joined() {
timeout(Duration::from_secs(120), run_watch()).await.expect("the test finishes in time");
}
async fn run_watch() {
let keys = Keys::generate();
let server = relay(&keys).await;
let dir = std::env::temp_dir().join(format!("joule-moq-room-metrics-watch-{}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
let file = dir.join("source.wav");
std::fs::write(&file, (0..4096u32).map(|i| (i % 251) as u8).collect::<Vec<u8>>()).unwrap();
// The clock's session without the watch loop, which takes a viewer that says nothing for gone.
let rest = format!("[room]\nwatch_source = \"{}\"\nwatch_duration_ms = 60000\n", file.display());
let (svc, addr) = service(&keys, &server, &rest, |s| {
tokio::spawn(s.clone().run_clock());
})
.await;
let (publisher, viewer) = (SigningKey::from_bytes(&[0x44; 32]), SigningKey::from_bytes(&[0x45; 32]));
let p = grant(addr, &publisher, "publisher", "203.0.113.45").await;
let v = grant(addr, &viewer, "viewer", "203.0.113.46").await;
let room = p["room"]["room"].as_str().unwrap().to_string();
assert_eq!(v["room"]["room"].as_str(), Some(room.as_str()));
let (_, f) = scrape(addr).await;
assert_eq!(value(&f, "wai_room_present", &[]), Some(0.0), "nobody has entered");
assert_eq!(value(&f, "wai_room_viewer_grants", &[]), Some(1.0));
// The viewer joins, as a page does.
let g = hex::decode(v["grant"].as_str().unwrap()).unwrap();
let (_, _, c) = call(addr, "GET", "/watch/challenge", None, "203.0.113.46").await;
let c: Value = serde_json::from_str(&c).unwrap();
let c: [u8; 32] = hex::decode(c["challenge"].as_str().unwrap()).unwrap().try_into().unwrap();
let digest: [u8; 32] = Sha256::digest(&g[..g.len() - 64]).into();
let proof = wai::live_room::Proof::sign(digest, &viewer, c, "POST //room.test/watch/join").to_bytes();
let body = json!({"room": room, "grant": hex::encode(&g), "proof": hex::encode(proof)});
let (st, _, j) = call(addr, "POST", "/watch/join", Some(&body), "203.0.113.46").await;
assert_eq!(st, 200, "{j}");
settles(addr, "wai_room_present", 1.0).await;
// The publisher enters: in the room's members and followed as a viewer is, and counted once.
let (session, held) = enter(&server, &room, &hex::decode(p["grant"].as_str().unwrap()).unwrap(), &publisher).await;
settles(addr, "wai_room_publishers", 1.0).await;
settles(addr, "wai_room_present", 2.0).await;
// Its join reaches the watch room's own record a moment after the member's state.
tokio::time::sleep(Duration::from_millis(300)).await;
let (_, f) = scrape(addr).await;
assert_eq!(value(&f, "wai_room_present", &[]), Some(2.0));
// The viewer falls silent, and the clock signs its leave.
let (generation, gone) = {
let mut st = svc.state.lock().unwrap();
let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_micros() as u64;
let gone = st.epoch.watch.as_mut().expect("a watch room").ended(now + joule_moq_room::watch::SILENT_US + 1_000_000);
(st.epoch.generation, gone)
};
assert_eq!(gone.len(), 1, "the viewer, which joined by a proof");
assert!(matches!(svc.watch_leave(generation, gone[0].0).await, Ok(Some(_))), "its leave is signed");
settles(addr, "wai_room_present", 1.0).await;
drop(held);
session.close(0, "done");
settles(addr, "wai_room_present", 0.0).await;
let _ = std::fs::remove_dir_all(&dir);
server.shutdown().await;
}
/// A remote desktop room grants by its own roles: its host holds a publisher's place, and its
/// controllers and viewers are its audience and its viewers' grants.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_remote_room_counts_its_controllers_and_viewers() {
timeout(Duration::from_secs(120), async {
let keys = Keys::generate();
let server = relay(&keys).await;
let rest = "[room]\nmax_present = 8\njoin_wait_s = 30\n[rate]\nclient_burst = 20\nclient_every_s = 1\n[remote]\nwidth = 320\nheight = 192\nfps = 20\nallow = \"all\"\nmax_age_ms = 3000\nlead_ms = 400\n";
let (_svc, addr) = service(&keys, &server, rest, |s| s.start()).await;
let (_, f) = scrape(addr).await;
for role in ["host", "controller", "viewer"] {
assert_eq!(value(&f, "wai_room_grants_total", &[("role", role)]), Some(0.0), "{role}");
}
assert_eq!(value(&f, "wai_room_grants_total", &[("role", "publisher")]), None, "this room grants no publisher");
let host = SigningKey::from_bytes(&[0x46; 32]);
let body = json!({"principal": pubhex(&host), "role": "host", "invitation": remote::new_invitation(&keys, 600)});
let (st, _, g) = call(addr, "POST", "/grant", Some(&body), "203.0.113.47").await;
assert_eq!(st, 200, "{g}");
let (_, f) = scrape(addr).await;
assert_eq!(value(&f, "wai_room_publishers", &[]), Some(1.0), "the host's place");
assert_eq!(value(&f, "wai_room_viewer_grants", &[]), Some(0.0), "the host's is not a viewer's grant");
grant(addr, &SigningKey::from_bytes(&[0x47; 32]), "controller", "203.0.113.48").await;
grant(addr, &SigningKey::from_bytes(&[0x48; 32]), "viewer", "203.0.113.49").await;
let (_, f) = scrape(addr).await;
assert_eq!(value(&f, "wai_room_viewer_grants", &[]), Some(2.0));
assert_eq!(value(&f, "wai_room_audience", &[]), Some(2.0));
assert_eq!(value(&f, "wai_room_publishers", &[]), Some(1.0));
for role in ["host", "controller", "viewer"] {
assert_eq!(value(&f, "wai_room_grants_total", &[("role", role)]), Some(1.0), "{role}");
}
server.shutdown().await;
})
.await
.expect("the test finishes in time");
}
/// A conversation's request of `k`, with its grant if it has one, and a proof over a fresh
/// challenge (live-rooms §10.5.3).
async fn conversation(addr: std::net::SocketAddr, k: &SigningKey, grant: Option<&[u8]>, path: &str) -> (u16, Value) {
let (_, _, c) = call(addr, "GET", "/msg/challenge", None, "203.0.113.50").await;
let c: Value = serde_json::from_str(&c).unwrap();
let c: [u8; 32] = hex::decode(c["challenge"].as_str().unwrap()).unwrap().try_into().unwrap();
let digest = grant.map_or([0; 32], |g| gp::grant_digest_and_principal(g).unwrap().0);
let proof = GrantProof::sign(digest, k, c, &gp::context("POST", "room.test", path)).unwrap();
let mut headers = vec![(gp::HEADER_PROOF, gp::base64url(&proof.to_bytes()))];
if let Some(g) = grant {
headers.push(("Authorization", format!("Bearer {}", gp::base64url(g))));
}
let (st, _, v) = request(addr, "POST", path, &headers, Some(&json!({})), "203.0.113.50").await;
(st, serde_json::from_str(&v).unwrap_or(Value::Null))
}
/// A conversation keeps its members itself, and they are the principals present: none of them is
/// a publisher or a viewer of a room.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_conversation_counts_its_members() {
timeout(Duration::from_secs(120), async {
let keys = Keys::generate();
let dir = std::env::temp_dir().join(format!("joule-moq-room-metrics-msg-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
let cfg = Config::from_toml(&format!("listen = \"127.0.0.1:0\"\nkeys = \"/nonexistent\"\nrelay_url = \"ws://127.0.0.1:1/moq\"\n[messaging]\nstore = \"{}\"\nmax_members = 8\nmax_text = 512\n[rate]\nclient_burst = 64\n", dir.display())).unwrap();
let svc = Service::new(Keys::from_toml(&keys.to_toml()).unwrap(), &cfg);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(http::serve(svc.clone(), listener));
let (_, f) = scrape(addr).await;
assert_eq!(value(&f, "wai_room_present", &[]), Some(0.0));
assert_eq!(value(&f, "wai_room_relay_connected", &[]), Some(0.0), "the clock's session was never opened");
let (alice, bob) = (SigningKey::from_bytes(&[0x49; 32]), SigningKey::from_bytes(&[0x4a; 32]));
let (st, a) = conversation(addr, &alice, None, "/msg/join").await;
assert_eq!(st, 200, "{a}");
let (st, b) = conversation(addr, &bob, None, "/msg/join").await;
assert_eq!(st, 200, "{b}");
let (_, f) = scrape(addr).await;
assert_eq!(value(&f, "wai_room_present", &[]), Some(2.0));
assert_eq!(value(&f, "wai_room_publishers", &[]), Some(0.0));
assert_eq!(value(&f, "wai_room_viewer_grants", &[]), Some(0.0));
let g = hex::decode(b["grant"].as_str().unwrap()).unwrap();
let (st, v) = conversation(addr, &bob, Some(&g), "/msg/leave").await;
assert_eq!(st, 200, "{v}");
let (_, f) = scrape(addr).await;
assert_eq!(value(&f, "wai_room_present", &[]), Some(1.0), "the one that left is not");
let _ = std::fs::remove_dir_all(&dir);
})
.await
.expect("the test finishes in time");
}