7e92f0fa64
A SUBSCRIBE_TRACKS was not forwarded on an upstream link, so it offered the tracks this relay had an upstream of and not those an upstream relay held. -22 §7.6 has a relay send a PUBLISH "to each subscriber that has sent SUBSCRIBE_TRACKS" for each PUBLISH it receives, and -22's security text has a relay send a SUBSCRIBE_TRACKS upstream and receive "a number of new Subscriptions" by it. A SUBSCRIBE_TRACKS whose prefix overlaps a link's is now taken on to that link (the link's prefix where it is the longer and names no wildcard field): one SUBSCRIBE_TRACKS a link and prefix, shared by every request here that asks the same, sent on each session the link opens, and cancelled (its stream reset) when the last request holding it goes, before which it is in place by the time the request is answered. The upstream relay's PUBLISHes it brings are taken as a publisher's, under no grant, the link routing them (one it does not route refused UNINTERESTED), and offered on; the subscriptions they made stand when the request is cancelled (-23 §3.6). A request that came from another relay is answered with what this relay holds and taken no further: RELAY_PATH is no parameter of SUBSCRIBE_TRACKS, and a request taken on once cannot go round a loop of relays. Relay::tracks_up and Relay::subscribe_tracks_held state them for diagnosis. tests/cascade.rs, subscribe_tracks_reaches_an_upstream_relay: the home relay's track published before an edge subscriber's request and one published after offered at the edge and their objects arriving, a track the link does not route offered to no one, two requests one SUBSCRIBE_TRACKS on the link held twice, and once both are cancelled the home relay's copy cancelled, a later track not reaching the edge, the earlier subscriptions standing. Run against four faults planted in turn (no request taken to the link, a PUBLISH of a track the link does not route taken, the last request's end not cancelling the link's, a second request's hold not counted): each fails it. Co-Authored-By: Claude Opus 5.5 <[email protected]>
1174 lines
68 KiB
Rust
1174 lines
68 KiB
Rust
//! Relays that cascade (`crate::cascade`): an edge relay that dials a home relay, and chains
|
|
//! and loops of them, every relay in this process on ephemeral loopback ports.
|
|
//!
|
|
//! - Two relays, over each transport the edge can dial (native QUIC and WebTransport with a
|
|
//! pinned certificate, and the WebSocket binding): a publisher on each, a subscriber of the
|
|
//! other's track on each, every object verified against its publisher's receipts at the
|
|
//! subscriber and by the door of every relay it crossed; namespaces published both ways and
|
|
//! discovered from either relay; each relay's summary of the link names the other.
|
|
//! - Three relays in a chain, with a room's tracks both ways: the path a namespace's
|
|
//! announcement took, and a per-hop record from every hop, whose entries chain hop to hop.
|
|
//! - Receipts at every door: a tampered object a relay without a door forwards is refused at
|
|
//! the next relay's door.
|
|
//! - A loop of three relays: an announcement never comes back to its origin, and a request for
|
|
//! a track nobody publishes is refused `relay_loop` at once.
|
|
//! - Failover: a link to an upstream that is down is passed over for the next; with none up, a
|
|
//! SUBSCRIBE is refused `TIMEOUT` `no_upstream`.
|
|
//! - GOAWAY: an upstream's GOAWAY moves the edge's subscriptions to the URI it names, and its
|
|
//! subscriber sees no GOAWAY and loses no group.
|
|
//! - The DVR window of the home relay, fetched from the edge: served with the seek power, and
|
|
//! refused `seek_not_granted` at the edge without it.
|
|
//! - Co-performance across relays: a member's votes counted in its chain order at the relay it
|
|
//! publishes to, and a watcher at the other relay given exactly what one beside it is; and the
|
|
//! floor held once, at the performer's relay.
|
|
//! - SUBSCRIBE_TRACKS across relays: the edge takes a subscriber's request to its link, and the
|
|
//! home relay's tracks under what the link routes come back as PUBLISHes the edge offers on,
|
|
//! one SUBSCRIBE_TRACKS on the link for every request of the edge that asks the same, and
|
|
//! none once the last of them is cancelled.
|
|
|
|
mod common;
|
|
|
|
use common::*;
|
|
use ed25519_dalek::{Signer, SigningKey};
|
|
use joule_moq::carriage::{object_properties, ObjectExtras, ReceiptBundle};
|
|
use joule_moq::draft22::data::{FetchEntry, ObjectStatus as WireStatus, SubgroupHeader, SubgroupId, SubgroupObject};
|
|
use joule_moq::draft22::{ControlMessage, Location, LocationFilter, NamespaceRequest, Parameter, Parameters, Publish, RequestOk, Token};
|
|
use joule_moq::messages::ObjectHeader;
|
|
use joule_moq::receipt::ReceiptVerifier;
|
|
use joule_moq::{GroupId, ObjectId, ObjectStatus, TrackAlias, TrackNamespace};
|
|
use joule_moq_relay::client::{Client, ClientOptions};
|
|
use joule_moq_relay::config::{Authority, DoorRule, DvrConfig, DvrWindow, FloorConfig, Grant, UpstreamConfig, UpstreamTransport};
|
|
use joule_moq_relay::wire::ReqRx;
|
|
use joule_moq_relay::{RelayConfig, RelayServer};
|
|
use std::collections::BTreeSet;
|
|
use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
|
use tokio::time::timeout;
|
|
|
|
fn hexkey(k: &SigningKey) -> String {
|
|
hex::encode(k.verifying_key().to_bytes())
|
|
}
|
|
|
|
fn now_us() -> u64 {
|
|
SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_micros() as u64
|
|
}
|
|
|
|
fn sha256(b: &[u8]) -> [u8; 32] {
|
|
ring::digest::digest(&ring::digest::SHA256, b).as_ref().try_into().unwrap()
|
|
}
|
|
|
|
/// A loopback port nothing listens on once this returns.
|
|
fn free_port() -> u16 {
|
|
std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port()
|
|
}
|
|
|
|
fn upstream(url: String, transport: UpstreamTransport, prefixes: &[&[&str]]) -> UpstreamConfig {
|
|
UpstreamConfig {
|
|
name: url.clone(),
|
|
url,
|
|
transport,
|
|
prefixes: prefixes.iter().map(|p| p.iter().map(|f| f.to_string()).collect()).collect(),
|
|
ca_path: None,
|
|
cert_sha256: None,
|
|
native_roots: false,
|
|
token_sha256: None,
|
|
token_path: None,
|
|
relay_key: None,
|
|
}
|
|
}
|
|
|
|
/// An upstream entry for `server`, dialed over `via`: native QUIC and WebTransport trust its
|
|
/// certificate by its pinned SHA-256.
|
|
fn upstream_to(server: &RelayServer, via: Via, prefixes: &[&[&str]]) -> UpstreamConfig {
|
|
let path = server.relay().config().path.clone();
|
|
let pin = Some(hex::encode(server.identity.leaf_sha256));
|
|
let mut u = match via {
|
|
Via::Quic => upstream(format!("moqt://127.0.0.1:{}{path}", server.quic_addr.unwrap().port()), UpstreamTransport::Quic, prefixes),
|
|
Via::WebTransport => upstream(format!("https://127.0.0.1:{}{path}", server.quic_addr.unwrap().port()), UpstreamTransport::WebTransport, prefixes),
|
|
_ => upstream(format!("ws://{}{path}", server.websocket_addr.unwrap()), UpstreamTransport::WebSocket, prefixes),
|
|
};
|
|
if matches!(via, Via::Quic | Via::WebTransport) {
|
|
u.cert_sha256 = pin;
|
|
}
|
|
u.relay_key = Some(hex::encode(server.relay().relay_id()));
|
|
u
|
|
}
|
|
|
|
/// A token in a file of its own, for an upstream entry: (its SHA-256, the file).
|
|
fn token_file(tag: &str, token: &[u8]) -> (String, std::path::PathBuf) {
|
|
let p = std::env::temp_dir().join(format!("joule-moq-relay-cascade-{}-{tag}.token", std::process::id()));
|
|
std::fs::write(&p, [token, b"\n"].concat()).unwrap();
|
|
(hex::encode(sha256(token)), p)
|
|
}
|
|
|
|
fn quick(c: &mut RelayConfig) {
|
|
c.limits.goaway_timeout_ms = 500;
|
|
c.limits.upstream_reconnect_ms = 100;
|
|
}
|
|
|
|
/// Wait until `server` has `n` upstream sessions open.
|
|
async fn linked(server: &RelayServer, n: usize) {
|
|
assert!(until(STEP, || server.relay().links().iter().filter(|l| l.session().is_some()).count() >= n).await, "the upstream links open");
|
|
}
|
|
|
|
/// The namespaces a SUBSCRIBE_NAMESPACE's stream names, as NAMESPACE and NAMESPACE_DONE leave
|
|
/// them, once it holds `want` of them.
|
|
async fn namespaces_seen(rx: &mut ReqRx, want: usize) -> BTreeSet<TrackNamespace> {
|
|
let mut seen = BTreeSet::new();
|
|
while seen.len() < want {
|
|
match timeout(STEP, rx.next()).await.expect("NAMESPACE in time").expect("a message") {
|
|
Some(ControlMessage::Namespace(s)) => {
|
|
seen.insert(s);
|
|
}
|
|
Some(ControlMessage::NamespaceDone(s)) => {
|
|
seen.remove(&s);
|
|
}
|
|
other => panic!("expected NAMESPACE, got {other:?}"),
|
|
}
|
|
}
|
|
seen
|
|
}
|
|
|
|
fn door(keys: &[&SigningKey]) -> Vec<DoorRule> {
|
|
vec![DoorRule { prefix: vec!["live".into()], trust: keys.iter().map(|k| hexkey(k)).collect(), trust_field: None, hold_ms: 5_000 }]
|
|
}
|
|
|
|
// ── Two relays, over each transport ────────────────────────────────────────
|
|
|
|
async fn two_relays(via: Via) {
|
|
const G: u64 = 8;
|
|
let key_a = SigningKey::from_bytes(&[0xc1; 32]);
|
|
let key_b = SigningKey::from_bytes(&[0xc2; 32]);
|
|
let home = relay(|c| {
|
|
quick(c);
|
|
c.receipts.door = door(&[&key_a, &key_b]);
|
|
})
|
|
.await;
|
|
let edge = relay(|c| {
|
|
quick(c);
|
|
c.receipts.door = door(&[&key_a, &key_b]);
|
|
c.upstream = vec![upstream_to(&home, via, &[&["live"]])];
|
|
})
|
|
.await;
|
|
linked(&edge, 1).await;
|
|
let (home_id, edge_id) = (home.relay().relay_id(), edge.relay().relay_id());
|
|
assert_eq!(edge.relay().links()[0].session().unwrap().peer.unwrap().id, home_id, "the edge's link names the home relay by its id");
|
|
|
|
// A publisher on each: live/a on the home relay by PUBLISH, live/b on the edge by
|
|
// PUBLISH_NAMESPACE; live/c, a namespace alone, on the home relay.
|
|
let ta = Tracks::new(ns(&["live", "a"]), "cam", 10, 8, 64);
|
|
let tb = Tracks::new(ns(&["live", "b"]), "cam", 20, 9, 96);
|
|
let (pa, _ea) = connect(&home, Via::WebSocket, ClientOptions::default()).await;
|
|
let (pb, _eb) = connect(&edge, Via::WebSocket, ClientOptions::default()).await;
|
|
let (pc, _ec) = connect(&home, Via::WebSocket, ClientOptions::default()).await;
|
|
let _pub_a = publish_tracks(&pa, &ta, None).await;
|
|
let _ns_b = pb.publish_namespace(tb.ns.clone(), Parameters::new()).await.expect("PUBLISH_NAMESPACE on the edge");
|
|
let _ns_c = pc.publish_namespace(ns(&["live", "c"]), Parameters::new()).await.expect("PUBLISH_NAMESPACE on the home relay");
|
|
// The edge publishes live/b on its link: the home relay holds it, its path the edge alone.
|
|
assert!(
|
|
until(STEP, || home.relay().namespaces().iter().any(|(n, _, p)| *n == tb.ns && *p == vec![edge_id])).await,
|
|
"the edge's namespace reaches the home relay, its path the edge"
|
|
);
|
|
|
|
// Discovery from either relay: live/b and live/c, wherever each is published.
|
|
let (dh, _) = connect(&home, Via::WebSocket, ClientOptions::default()).await;
|
|
let (de, _) = connect(&edge, Via::WebSocket, ClientOptions::default()).await;
|
|
let (_dht, mut dhr) = dh.subscribe_namespace(ns(&["live"])).await.expect("SUBSCRIBE_NAMESPACE on the home relay");
|
|
let (_det, mut der) = de.subscribe_namespace(ns(&["live"])).await.expect("SUBSCRIBE_NAMESPACE on the edge");
|
|
let want: BTreeSet<TrackNamespace> = [ns(&["b"]), ns(&["c"])].into_iter().collect();
|
|
assert_eq!(namespaces_seen(&mut dhr, 2).await, want, "the home relay names the edge's namespace and its own");
|
|
assert_eq!(namespaces_seen(&mut der, 2).await, want, "the edge names its own namespace and the home relay's");
|
|
|
|
// A subscriber of the other's track on each relay.
|
|
let (sb, _e1) = connect(&edge, via, ClientOptions::default()).await;
|
|
let (sa, _e2) = connect(&home, via, ClientOptions::default()).await;
|
|
let vb = subscribe_both(&sb, &ta, Parameters::new()).await;
|
|
let (held_b, va) = tokio::join!(answer_subscribes(&pb, &tb), subscribe_both(&sa, &tb, Parameters::new()));
|
|
let pub_a = async {
|
|
for g in 0..G {
|
|
publish_group(&pa, &key_a, &ta, g, &payload("a", g, 700)).await;
|
|
}
|
|
};
|
|
let pub_b = async {
|
|
for g in 0..G {
|
|
publish_group(&pb, &key_b, &tb, g, &payload("b", g, 900)).await;
|
|
}
|
|
};
|
|
let ((), (), got_b, got_a) = tokio::join!(pub_a, pub_b, collect(&sb, STEP * 2, |g| g.len() as u64 >= 2 * G), collect(&sa, STEP * 2, |g| g.len() as u64 >= 2 * G));
|
|
let (ob, bb, _, _) = verify(&got_b, &vb, &key_a);
|
|
let (oa, ba, _, _) = verify(&got_a, &va, &key_b);
|
|
assert_eq!((ob, bb, oa, ba), (G as usize, G as usize, G as usize, G as usize), "every object of each publisher, verified, at the other relay");
|
|
assert_eq!(home.relay().stats().refused_at_door + edge.relay().stats().refused_at_door, 0);
|
|
assert_eq!(edge.relay().stats().forwarded_upstream, 2, "the media track and its receipts track, each forwarded upstream once");
|
|
drop(held_b);
|
|
|
|
for c in [&pa, &pb, &pc, &dh, &de, &sb, &sa] {
|
|
c.close(0, "done");
|
|
}
|
|
let (er, hr) = (edge.relay().clone(), home.relay().clone());
|
|
edge.shutdown().await;
|
|
home.shutdown().await;
|
|
// A session leaves the relay's table before its summary is signed: wait for every one.
|
|
assert!(until(STEP, || er.summaries().len() >= 4 && hr.summaries().len() >= 5).await, "every session's summary at each relay");
|
|
// Every hop's door verified the objects it forwarded: the publishers' signed figures, at
|
|
// the subscriber's relay, for the other relay's media objects.
|
|
// (The link's own summary at each end counts what that relay forwarded over it, verified at
|
|
// its door, too: home's publisher's objects at the home relay, the edge's at the edge.)
|
|
let signed = |r: &joule_moq_relay::Relay, link: bool| {
|
|
r.summaries().iter().filter(|s| s.summary.tally.peer.is_some() == link && s.summary.tally.subscriber.objects_forwarded == 2 * G).map(|s| s.summary.tally.publisher_signed.get("unmetered").map_or(0, |f| f.objects)).collect::<Vec<_>>()
|
|
};
|
|
assert_eq!((signed(&er, false), signed(&hr, false)), (vec![G], vec![G]), "each relay's door verified every media object its subscriber was given");
|
|
assert_eq!((signed(&er, true), signed(&hr, true)), (vec![G], vec![G]), "and every one it sent the other relay");
|
|
// Each relay's summary of the link names the other relay.
|
|
assert!(er.summaries().iter().any(|s| s.summary.tally.peer.as_deref() == Some(hex::encode(home_id).as_str())));
|
|
assert!(hr.summaries().iter().any(|s| s.summary.tally.peer.as_deref() == Some(hex::encode(edge_id).as_str())));
|
|
for s in er.summaries().iter().chain(hr.summaries().iter()) {
|
|
s.verify().expect("every summary verifies under its relay's key");
|
|
}
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn two_relays_over_a_websocket_upstream() {
|
|
timeout(Duration::from_secs(120), two_relays(Via::WebSocket)).await.expect("in time");
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn two_relays_over_a_quic_upstream() {
|
|
timeout(Duration::from_secs(120), two_relays(Via::Quic)).await.expect("in time");
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn two_relays_over_a_webtransport_upstream() {
|
|
timeout(Duration::from_secs(120), two_relays(Via::WebTransport)).await.expect("in time");
|
|
}
|
|
|
|
// ── Three relays in a chain, and their per-hop records ─────────────────────
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn three_relays_in_a_chain() {
|
|
timeout(Duration::from_secs(120), chain()).await.expect("in time");
|
|
}
|
|
|
|
async fn chain() {
|
|
const G: u64 = 6;
|
|
let t0 = now_us() - 1_000_000;
|
|
let room = [0x5a; 32];
|
|
let rh = hex::encode(room);
|
|
let (kp, kq) = (SigningKey::from_bytes(&[0xd1; 32]), SigningKey::from_bytes(&[0xd2; 32]));
|
|
let tp = Tracks::new(ns(&[&rh, &hexkey(&kp), "audio"]), "0", 30, 31, 32);
|
|
let tq = Tracks::new(ns(&[&rh, &hexkey(&kq), "audio"]), "0", 40, 41, 32);
|
|
let door = |c: &mut RelayConfig| c.receipts.door = vec![DoorRule { prefix: vec![rh.clone()], trust: vec![hexkey(&kp), hexkey(&kq)], trust_field: None, hold_ms: 5_000 }];
|
|
// C dials B, B dials A, each routing the room.
|
|
let a = relay(|c| {
|
|
quick(c);
|
|
door(c);
|
|
})
|
|
.await;
|
|
let b = relay(|c| {
|
|
quick(c);
|
|
door(c);
|
|
c.upstream = vec![upstream_to(&a, Via::WebSocket, &[&[&rh]])];
|
|
})
|
|
.await;
|
|
let cc = relay(|c| {
|
|
quick(c);
|
|
door(c);
|
|
c.upstream = vec![upstream_to(&b, Via::WebSocket, &[&[&rh]])];
|
|
})
|
|
.await;
|
|
linked(&b, 1).await;
|
|
linked(&cc, 1).await;
|
|
let (ida, idb, idc) = (a.relay().relay_id(), b.relay().relay_id(), cc.relay().relay_id());
|
|
|
|
// P publishes on A by PUBLISH; Q on C by PUBLISH_NAMESPACE, which reaches A through B.
|
|
let (pp, _) = connect(&a, Via::WebSocket, ClientOptions::default()).await;
|
|
let (pq, _) = connect(&cc, Via::WebSocket, ClientOptions::default()).await;
|
|
let _pub_p = publish_tracks(&pp, &tp, None).await;
|
|
let _ns_q = pq.publish_namespace(tq.ns.clone(), Parameters::new()).await.expect("PUBLISH_NAMESPACE on C");
|
|
assert!(until(STEP, || a.relay().namespaces().iter().any(|(n, _, p)| *n == tq.ns && *p == vec![idc, idb])).await, "Q's namespace reaches A, its path C then B");
|
|
|
|
// A viewer of P on C (C → B → A by route), a viewer of Q on A (A → B → C by announcement).
|
|
let (vc, _) = connect(&cc, Via::WebSocket, ClientOptions::default()).await;
|
|
let (va, _) = connect(&a, Via::WebSocket, ClientOptions::default()).await;
|
|
let sp = subscribe_both(&vc, &tp, Parameters::new()).await;
|
|
let (held_q, sq) = tokio::join!(answer_subscribes(&pq, &tq), subscribe_both(&va, &tq, Parameters::new()));
|
|
let pub_p = async {
|
|
for g in 0..G {
|
|
publish_group(&pp, &kp, &tp, g, &payload("p", g, 500)).await;
|
|
}
|
|
};
|
|
let pub_q = async {
|
|
for g in 0..G {
|
|
publish_group(&pq, &kq, &tq, g, &payload("q", g, 600)).await;
|
|
}
|
|
};
|
|
let ((), (), got_c, got_a) = tokio::join!(pub_p, pub_q, collect(&vc, STEP * 2, |g| g.len() as u64 >= 2 * G), collect(&va, STEP * 2, |g| g.len() as u64 >= 2 * G));
|
|
assert_eq!(verify(&got_c, &sp, &kp).0, G as usize, "P's objects, verified, at C");
|
|
assert_eq!(verify(&got_a, &sq, &kq).0, G as usize, "Q's objects, verified, at A");
|
|
drop(held_q);
|
|
for c in [&pp, &pq, &vc, &va] {
|
|
c.close(0, "done");
|
|
}
|
|
let (ra, rb, rc) = (a.relay().clone(), b.relay().clone(), cc.relay().clone());
|
|
cc.shutdown().await;
|
|
b.shutdown().await;
|
|
a.shutdown().await;
|
|
// A session leaves the relay's table before its summary is signed: wait for every one (A:
|
|
// P, its viewer, the link from B; B: its two links; C: Q, its viewer, the link to B).
|
|
assert!(until(STEP, || ra.summaries().len() >= 3 && rb.summaries().len() >= 2 && rc.summaries().len() >= 3).await, "every session's summary at each relay");
|
|
|
|
// A per-hop record from every hop, each signed by its own relay; P's media track and Q's
|
|
// chain hop to hop: what one hop sent the next, the next received.
|
|
let render = |t: &Tracks| joule_moq::draft22::names::render(&t.ns, Some(&t.media));
|
|
let digest = |t: &Tracks| match joule_moq_relay::metering::classify(&render(t)).expect("a track of the room").1 {
|
|
joule_moq_relay::metering::Kind::Media { digest, .. } => digest,
|
|
other => panic!("not a media track: {other:?}"),
|
|
};
|
|
let (dp, dq) = (digest(&tp), digest(&tq));
|
|
let recs: Vec<_> = [&ra, &rb, &rc].iter().map(|r| r.hop_record(room, t0)).collect();
|
|
for (rec, id) in recs.iter().zip([ida, idb, idc]) {
|
|
rec.verify().expect("the record verifies under its hop's key");
|
|
assert_eq!(rec.hop, id, "each hop signs its own record");
|
|
// Each hop's own figure: the unmetered marker on a host with no counter, or its own
|
|
// counter's advance (OnChipCounter) where the relay reads one, as on macOS.
|
|
match rec.class_tag {
|
|
0 => assert_eq!(rec.joules, 0, "the unmetered marker is 0 with class tag 0"),
|
|
4 => {}
|
|
t => panic!("class tag {t}: the relay signs the unmetered marker or OnChipCounter"),
|
|
}
|
|
}
|
|
let objects = |rec: &joule_moq_relay::metering::HopRecord, track: [u8; 32], dir: u8| rec.entries.iter().find(|e| e.kind == 1 && e.track == track && e.dir == dir).map_or(0, |e| e.objects);
|
|
for (track, name) in [(dp, "P"), (dq, "Q")] {
|
|
let flow: Vec<(u64, u64)> = recs.iter().map(|r| (objects(r, track, 1), objects(r, track, 2))).collect();
|
|
assert_eq!(flow, vec![(G, G); 3], "{name}'s media objects in and out at A, B and C");
|
|
}
|
|
// Only the viewers are subscriber sessions: a hop link is counted at neither end.
|
|
assert_eq!(recs.iter().map(|r| r.sessions).collect::<Vec<_>>(), vec![1, 0, 1], "one subscriber session at A, none at B, one at C");
|
|
assert!(recs[1].ledger.is_empty(), "B served no principal: its ledger is empty");
|
|
assert_eq!(recs[0].ledger.iter().map(|l| l.sessions).sum::<u32>(), 2, "A's ledger: P's session and the viewer's, not the link");
|
|
}
|
|
|
|
// ── Receipts at every door ─────────────────────────────────────────────────
|
|
|
|
/// Group `g` of `t` signed over `signed`, its object carrying `sent`.
|
|
async fn publish_group_as(c: &Client, key: &SigningKey, t: &Tracks, g: u64, signed: &[u8], sent: &[u8]) {
|
|
let (receipt, bundle) = signed_group(key, t.receipt_alias, g, signed);
|
|
let rh = SubgroupHeader { track_alias: t.receipts_alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(t.priority), properties: false, end_of_group: true, first_object: true };
|
|
c.send_subgroup(rh, &[SubgroupObject::normal(0, bundle.to_payload().unwrap())]).await.unwrap();
|
|
let header = ObjectHeader { track_alias: TrackAlias(t.media_alias), group_id: GroupId(g), object_id: ObjectId(0), publisher_priority: t.priority, status: ObjectStatus::Normal, payload_len: sent.len() as u64, mrl_layer: 0, mrl_total_layers: 0, wai_envelope: false };
|
|
let props = object_properties(&header, &ObjectExtras { receipt: Some(receipt), wai_receipt: None, delivery_timeout_ms: None }).unwrap();
|
|
let mh = SubgroupHeader { track_alias: t.media_alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(t.priority), properties: true, end_of_group: true, first_object: true };
|
|
c.send_subgroup(mh, &[SubgroupObject { object_id: 0, properties: props, status: WireStatus::Normal, payload: sent.to_vec() }]).await.unwrap();
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn receipts_are_checked_at_every_door() {
|
|
timeout(Duration::from_secs(120), every_door()).await.expect("in time");
|
|
}
|
|
|
|
async fn every_door() {
|
|
const G: u64 = 6;
|
|
const BAD: u64 = 3;
|
|
let key = SigningKey::from_bytes(&[0xe1; 32]);
|
|
// The home relay keeps no door; the edge does.
|
|
let home = relay(quick).await;
|
|
let edge = relay(|c| {
|
|
quick(c);
|
|
c.receipts.door = door(&[&key]);
|
|
c.upstream = vec![upstream_to(&home, Via::WebSocket, &[&["live"]])];
|
|
})
|
|
.await;
|
|
linked(&edge, 1).await;
|
|
let t = Tracks::new(ns(&["live", "t"]), "cam", 10, 8, 64);
|
|
let (p, _) = connect(&home, Via::WebSocket, ClientOptions::default()).await;
|
|
let _pubs = publish_tracks(&p, &t, None).await;
|
|
let (vh, _) = connect(&home, Via::WebSocket, ClientOptions::default()).await;
|
|
let (ve, _) = connect(&edge, Via::WebSocket, ClientOptions::default()).await;
|
|
let sh = subscribe_both(&vh, &t, Parameters::new()).await;
|
|
let se = subscribe_both(&ve, &t, Parameters::new()).await;
|
|
for g in 0..G {
|
|
let good = payload("t", g, 800);
|
|
let mut sent = good.clone();
|
|
if g == BAD {
|
|
// Changed after it was signed: its receipt names the bytes it no longer holds.
|
|
sent[20] ^= 0x01;
|
|
}
|
|
publish_group_as(&p, &key, &t, g, &good, &sent).await;
|
|
}
|
|
let got_h = collect(&vh, STEP, |g| g.len() as u64 >= 2 * G).await;
|
|
let got_e = collect(&ve, STEP, |g| g.iter().filter(|x| x.header.track_alias == se.receipts_alias).count() as u64 >= G && g.iter().filter(|x| x.header.track_alias == se.media_alias).count() as u64 >= G - 1).await;
|
|
// The home relay, with no door, forwarded the tampered object: its own subscriber has it,
|
|
// and it does not verify.
|
|
let bad_at_home = got_h.iter().find(|x| x.header.track_alias == sh.media_alias && x.header.group_id == BAD).expect("the home relay forwarded group 3");
|
|
let mut changed = payload("t", BAD, 800);
|
|
changed[20] ^= 0x01;
|
|
assert_eq!(bad_at_home.objects[0].payload, changed, "the home relay forwarded the changed bytes");
|
|
assert_eq!(home.relay().stats().refused_at_door, 0, "the home relay checked nothing");
|
|
// The edge checked it again at its door and refused it, though its upstream forwarded it.
|
|
let media_e: Vec<u64> = got_e.iter().filter(|x| x.header.track_alias == se.media_alias && x.reset.is_none() && !x.objects.is_empty()).map(|x| x.header.group_id).collect();
|
|
assert!(!media_e.contains(&BAD), "the edge forwarded no tampered object: {media_e:?}");
|
|
assert_eq!(media_e.len() as u64, G - 1, "every other group reached the edge's subscriber");
|
|
// Each group's door verdict is reached in that group's own task, and nothing the
|
|
// subscriber was given waits for group BAD's: every other group can reach it while BAD is
|
|
// still on its way to the door. So the refusal is waited for, then counted.
|
|
assert!(until(STEP, || edge.relay().stats().refused_at_door >= 1).await, "the edge's door refused the tampered object");
|
|
assert_eq!(edge.relay().stats().refused_at_door, 1, "the edge's door refused the tampered object, and nothing else");
|
|
let verifier = ReceiptVerifier::with_anchors(vec![key.verifying_key()]);
|
|
for x in got_e.iter().filter(|x| x.header.track_alias == se.receipts_alias) {
|
|
verifier.verify_group(&ReceiptBundle::from_payload(&x.objects[0].payload).unwrap().group).expect("every receipt verifies");
|
|
}
|
|
for c in [&p, &vh, &ve] {
|
|
c.close(0, "done");
|
|
}
|
|
let er = edge.relay().clone();
|
|
edge.shutdown().await;
|
|
home.shutdown().await;
|
|
assert!(until(STEP, || er.summaries().len() >= 2).await, "the edge's viewer's and its link's summaries");
|
|
// The refusal is counted against the link it arrived on: the home relay's session.
|
|
let link = er.summaries().into_iter().find(|s| s.summary.tally.peer.is_some()).expect("the link's summary");
|
|
assert_eq!(link.summary.tally.publisher.refused_at_door, 1);
|
|
// A refusal counted is not a group withheld, and the subscriber was read only until it
|
|
// held every other group: a door that counted late and then forwarded would pass both.
|
|
// What the edge gave its viewer, counted once the session ended, is every group but BAD.
|
|
let viewer = er.summaries().into_iter().find(|s| s.summary.tally.peer.is_none()).expect("the viewer's summary");
|
|
let m = viewer.summary.tally.subscriber.tracks.get(&joule_moq::draft22::names::render(&t.ns, Some(&t.media))).cloned().unwrap_or_default();
|
|
assert_eq!(m.objects_forwarded + m.undelivered_at_close, G - 1, "the edge gave its viewer no tampered object");
|
|
}
|
|
|
|
// ── A loop ─────────────────────────────────────────────────────────────────
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn a_loop_of_relays() {
|
|
timeout(Duration::from_secs(120), relay_loop()).await.expect("in time");
|
|
}
|
|
|
|
async fn relay_loop() {
|
|
// A dials B, B dials C, C dials A, each routing everything under live: a loop.
|
|
// Each relay dials the next, so the three ports are chosen before any of them listens, and
|
|
// another test can bind one in between: a start refused for a port in use is tried again
|
|
// on three new ports, at most five times. Anything else a start refuses fails the test.
|
|
let (ra, rb, rc) = {
|
|
let mut attempt = 0;
|
|
'start: loop {
|
|
attempt += 1;
|
|
let ports = [free_port(), free_port(), free_port()];
|
|
let mut started = Vec::new();
|
|
for i in 0..3 {
|
|
let mut c = RelayConfig::loopback();
|
|
quick(&mut c);
|
|
c.websocket_listen = Some(format!("127.0.0.1:{}", ports[i]).parse().unwrap());
|
|
c.upstream = vec![upstream(format!("ws://127.0.0.1:{}/moq", ports[(i + 1) % 3]), UpstreamTransport::WebSocket, &[&["live"]])];
|
|
match RelayServer::start(c).await {
|
|
Ok(r) => started.push(r),
|
|
Err(e) if attempt < 5 && e.contains("Address already in use") => {
|
|
for r in started {
|
|
r.shutdown().await;
|
|
}
|
|
continue 'start;
|
|
}
|
|
Err(e) => panic!("the relay starts: {e}"),
|
|
}
|
|
}
|
|
let mut it = started.into_iter();
|
|
break (it.next().unwrap(), it.next().unwrap(), it.next().unwrap());
|
|
}
|
|
};
|
|
for r in [&ra, &rb, &rc] {
|
|
linked(r, 1).await;
|
|
}
|
|
let (ida, idb) = (ra.relay().relay_id(), rb.relay().relay_id());
|
|
let la = ns(&["live", "a"]);
|
|
let t = Tracks::new(la.clone(), "cam", 10, 8, 64);
|
|
let (p, _) = connect(&ra, Via::WebSocket, ClientOptions::default()).await;
|
|
let _ns = p.publish_namespace(la.clone(), Parameters::new()).await.expect("PUBLISH_NAMESPACE on A");
|
|
// B holds it through A, C through B; it never comes back to A.
|
|
assert!(until(STEP, || rb.relay().namespaces().iter().any(|(n, _, path)| *n == la && *path == vec![ida])).await, "B holds A's namespace");
|
|
assert!(until(STEP, || rc.relay().namespaces().iter().any(|(n, _, path)| *n == la && *path == vec![ida, idb])).await, "C holds it through B");
|
|
tokio::time::sleep(Duration::from_millis(500)).await;
|
|
let at_a: Vec<_> = ra.relay().namespaces().into_iter().filter(|(n, _, _)| *n == la).collect();
|
|
assert_eq!(at_a.len(), 1, "A holds its own namespace once: it never came back around the loop");
|
|
assert!(at_a[0].2.is_empty());
|
|
let loops: u64 = [&ra, &rb, &rc].iter().map(|r| r.relay().stats().relay_loops).sum();
|
|
assert_eq!(loops, 0, "no relay was sent a request its path refuses: none was forwarded toward a relay on its path");
|
|
assert_eq!(rc.relay().links()[0].stats.announced.load(std::sync::atomic::Ordering::Relaxed), 0, "C published nothing back to A");
|
|
|
|
// A viewer on C reaches A's publisher through B.
|
|
let (v, _) = connect(&rc, Via::WebSocket, ClientOptions::default()).await;
|
|
let (held, s) = tokio::join!(answer_subscribes(&p, &t), subscribe_both(&v, &t, Parameters::new()));
|
|
let key = SigningKey::from_bytes(&[0xa7; 32]);
|
|
for g in 0..3 {
|
|
publish_group(&p, &key, &t, g, &payload("l", g, 300)).await;
|
|
}
|
|
let got = collect(&v, STEP, |g| g.len() >= 6).await;
|
|
assert_eq!(verify(&got, &s, &key).0, 3, "every object, through the loop's relays, verified");
|
|
drop(held);
|
|
|
|
// A track nobody publishes, from B: forwarded to C, then to A, whose only link leads back to
|
|
// B, on the path. Refused at once with relay_loop, not held until a timeout.
|
|
let (w, _) = connect(&rb, Via::WebSocket, ClientOptions::default()).await;
|
|
let asked = std::time::Instant::now();
|
|
let refused = w.subscribe(ns(&["live", "nobody"]), name("x"), Parameters::new()).await.map(|_| ()).expect_err("refused");
|
|
assert!(refused.contains("relay_loop"), "{refused}");
|
|
assert!(asked.elapsed() < Duration::from_secs(4), "refused at once: {:?}", asked.elapsed());
|
|
// A refused it, the relay whose only link leads back to B: B was never sent a request that
|
|
// names it, so no relay refused one where it arrived.
|
|
let loops: Vec<u64> = [&ra, &rb, &rc].iter().map(|r| r.relay().stats().relay_loops).collect();
|
|
assert_eq!(loops, vec![1, 0, 0], "the loop is refused at the relay that would close it, before it is sent");
|
|
for c in [&p, &v, &w] {
|
|
c.close(0, "done");
|
|
}
|
|
for r in [ra, rb, rc] {
|
|
r.shutdown().await;
|
|
}
|
|
}
|
|
|
|
// ── Failover ───────────────────────────────────────────────────────────────
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn an_upstream_that_is_down_is_passed_over() {
|
|
timeout(Duration::from_secs(120), failover()).await.expect("in time");
|
|
}
|
|
|
|
async fn failover() {
|
|
let up = relay(quick).await;
|
|
let dead = format!("ws://127.0.0.1:{}/moq", free_port());
|
|
let edge = relay(|c| {
|
|
quick(c);
|
|
c.upstream = vec![upstream(dead.clone(), UpstreamTransport::WebSocket, &[&["live"]]), upstream_to(&up, Via::WebSocket, &[&["live"]])];
|
|
})
|
|
.await;
|
|
let lonely = relay(|c| {
|
|
quick(c);
|
|
c.upstream = vec![upstream(dead.clone(), UpstreamTransport::WebSocket, &[&["live"]])];
|
|
})
|
|
.await;
|
|
linked(&edge, 1).await;
|
|
assert!(edge.relay().links()[0].session().is_none() && edge.relay().links()[1].session().is_some(), "the first upstream is down, the second up");
|
|
let t = Tracks::new(ns(&["live", "f"]), "cam", 10, 8, 64);
|
|
let (p, _) = connect(&up, Via::WebSocket, ClientOptions::default()).await;
|
|
let _pubs = publish_tracks(&p, &t, None).await;
|
|
let (v, _) = connect(&edge, Via::WebSocket, ClientOptions::default()).await;
|
|
let s = subscribe_both(&v, &t, Parameters::new()).await;
|
|
let key = SigningKey::from_bytes(&[0xf1; 32]);
|
|
for g in 0..3 {
|
|
publish_group(&p, &key, &t, g, &payload("f", g, 300)).await;
|
|
}
|
|
let got = collect(&v, STEP, |g| g.len() >= 6).await;
|
|
assert_eq!(verify(&got, &s, &key).0, 3, "served through the second upstream");
|
|
// No upstream up: a typed refusal, at once.
|
|
let (w, _) = connect(&lonely, Via::WebSocket, ClientOptions::default()).await;
|
|
let refused = w.subscribe(t.ns.clone(), t.media.clone(), Parameters::new()).await.map(|_| ()).expect_err("refused");
|
|
assert!(refused.contains("REQUEST_ERROR TIMEOUT") && refused.contains("no_upstream"), "{refused}");
|
|
assert_eq!(lonely.relay().stats().no_upstream, 1);
|
|
// With a rendezvous, the request waits for an upstream within it, then is refused the same.
|
|
let mut rz = Parameters::new();
|
|
rz.push(Parameter::RendezvousTimeout(300));
|
|
let refused = w.subscribe(t.ns.clone(), t.media.clone(), rz).await.map(|_| ()).expect_err("refused");
|
|
assert!(refused.contains("no_upstream"), "{refused}");
|
|
for c in [&p, &v, &w] {
|
|
c.close(0, "done");
|
|
}
|
|
for r in [edge, lonely, up] {
|
|
r.shutdown().await;
|
|
}
|
|
}
|
|
|
|
// ── GOAWAY ─────────────────────────────────────────────────────────────────
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn an_upstream_goaway_moves_the_subscriptions() {
|
|
timeout(Duration::from_secs(120), goaway()).await.expect("in time");
|
|
}
|
|
|
|
async fn goaway() {
|
|
let key = SigningKey::from_bytes(&[0x9a; 32]);
|
|
let t = Tracks::new(ns(&["live", "g"]), "cam", 10, 8, 64);
|
|
let h2 = relay(quick).await;
|
|
let to_h2 = format!("ws://{}{}", h2.websocket_addr.unwrap(), h2.relay().config().path);
|
|
let h1 = relay(|c| {
|
|
quick(c);
|
|
c.limits.goaway_timeout_ms = 3_000;
|
|
c.goaway_uri = to_h2.clone();
|
|
})
|
|
.await;
|
|
let edge = relay(|c| {
|
|
quick(c);
|
|
c.upstream = vec![upstream(format!("ws://{}{}", h1.websocket_addr.unwrap(), h1.relay().config().path), UpstreamTransport::WebSocket, &[&["live"]])];
|
|
})
|
|
.await;
|
|
linked(&edge, 1).await;
|
|
// The same track published to both, as a publisher with a standby does.
|
|
let (p1, _) = connect(&h1, Via::WebSocket, ClientOptions::default()).await;
|
|
let (p2, _) = connect(&h2, Via::WebSocket, ClientOptions::default()).await;
|
|
let _a = publish_tracks(&p1, &t, None).await;
|
|
let _b = publish_tracks(&p2, &t, None).await;
|
|
let (v, _) = connect(&edge, Via::WebSocket, ClientOptions::default()).await;
|
|
let s = subscribe_both(&v, &t, Parameters::new()).await;
|
|
for g in 0..4 {
|
|
let pl = payload("g", g, 400);
|
|
publish_group(&p1, &key, &t, g, &pl).await;
|
|
publish_group(&p2, &key, &t, g, &pl).await;
|
|
}
|
|
let mut got = collect(&v, STEP, |g| g.len() >= 8).await;
|
|
// H1 goes away, its GOAWAY naming H2: the edge moves its subscriptions there.
|
|
let stopping = tokio::spawn(h1.shutdown());
|
|
let link = edge.relay().links()[0].clone();
|
|
assert!(until(STEP, || link.stats.moved.load(std::sync::atomic::Ordering::Relaxed) >= 2).await, "both subscriptions moved to H2");
|
|
assert!(until(STEP, || link.session().is_some_and(|c| c.peer.unwrap().id == h2.relay().relay_id())).await, "the link's session is to H2");
|
|
for g in 4..8 {
|
|
publish_group(&p2, &key, &t, g, &payload("g", g, 400)).await;
|
|
}
|
|
got.extend(collect(&v, STEP, |g| g.len() >= 8).await);
|
|
let (objects, groups, _, _) = verify(&got, &s, &key);
|
|
assert_eq!((objects, groups), (8, 8), "every group, before and after the move, once");
|
|
assert!(v.goaway().borrow().is_none(), "the upstream's GOAWAY is not passed on to the edge's subscriber");
|
|
let _ = stopping.await;
|
|
for c in [&p1, &p2, &v] {
|
|
c.close(0, "done");
|
|
}
|
|
edge.shutdown().await;
|
|
h2.shutdown().await;
|
|
}
|
|
|
|
// ── The DVR window across relays ───────────────────────────────────────────
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn the_home_dvr_window_is_fetched_from_the_edge() {
|
|
timeout(Duration::from_secs(120), dvr()).await.expect("in time");
|
|
}
|
|
|
|
fn fetched_objects(entries: &[FetchEntry]) -> Vec<(u64, Vec<u8>)> {
|
|
entries
|
|
.iter()
|
|
.filter_map(|e| match e {
|
|
FetchEntry::Object(o) => Some((o.group_id, o.payload.clone())),
|
|
_ => None,
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
async fn dvr() {
|
|
const G: u64 = 8;
|
|
let key = SigningKey::from_bytes(&[0x3d; 32]);
|
|
// The home relay's store in memory where the host has it (`store_dir`): the wait below for
|
|
// its window to hold the track is for the store, not for a disk busy with other jobs.
|
|
let dir = store_dir("cascade-dvr");
|
|
let (pub_tok, peer_tok, seek_tok, live_tok) = (b"publisher".to_vec(), b"edge relay".to_vec(), b"viewer who seeks".to_vec(), b"viewer at the edge".to_vec());
|
|
let (peer_sha, peer_file) = token_file("dvr-peer", &peer_tok);
|
|
let rule = |t: &[u8], prefix: &[&str], publish: bool, subscribe: bool, seek: bool| Grant { prefix: prefix.iter().map(|s| s.to_string()).collect(), token_sha256: hex::encode(sha256(t)), publish, subscribe, proof: false, unlimited: true, seek };
|
|
let home = relay(|c| {
|
|
quick(c);
|
|
c.auth.required = true;
|
|
c.auth.grants = vec![rule(&pub_tok, &["live"], true, false, false), rule(&peer_tok, &[], true, true, true)];
|
|
c.dvr = Some(DvrConfig { path: dir.clone(), window: vec![DvrWindow { prefix: vec!["live".into()], max_age_s: 600, max_bytes: 1 << 30 }], room_keys: vec![], takedown_keys: vec![] });
|
|
})
|
|
.await;
|
|
let edge = relay(|c| {
|
|
quick(c);
|
|
c.auth.required = true;
|
|
c.auth.grants = vec![rule(&seek_tok, &["live"], false, true, true), rule(&live_tok, &["live"], false, true, false)];
|
|
let mut u = upstream_to(&home, Via::WebSocket, &[&["live"]]);
|
|
u.token_sha256 = Some(peer_sha.clone());
|
|
u.token_path = Some(peer_file.clone());
|
|
c.upstream = vec![u];
|
|
})
|
|
.await;
|
|
linked(&edge, 1).await;
|
|
let with = |t: &[u8]| ClientOptions { tokens: vec![Token::UseValue { token_type: 0, value: t.to_vec() }], ..ClientOptions::default() };
|
|
let t = Tracks::new(ns(&["live", "d"]), "cam", 10, 8, 64);
|
|
let (p, _) = connect(&home, Via::WebSocket, with(&pub_tok)).await;
|
|
let _pubs = publish_tracks(&p, &t, None).await;
|
|
for g in 0..G {
|
|
publish_group(&p, &key, &t, g, &payload("d", g, 600)).await;
|
|
}
|
|
assert!(until(STEP, || home.relay().dvr().unwrap().largest(&(t.ns.clone(), t.media.clone())).is_some_and(|l| l.group == G - 1)).await, "the home relay's window holds the track");
|
|
let from = |g: u64| Parameters::new().with(Parameter::LocationFilter(LocationFilter::AbsoluteStart { start: Location::new(g, 0) }));
|
|
// A viewer with the seek power at the edge: the whole window, from the home relay, and every
|
|
// object verifies against the receipts fetched the same way.
|
|
let (vs, _) = connect(&edge, Via::WebSocket, with(&seek_tok)).await;
|
|
let (_, media) = vs.fetch(t.ns.clone(), t.media.clone(), from(0)).await.expect("the window, fetched from the edge");
|
|
let (_, receipts) = vs.fetch(t.ns.clone(), t.receipts.clone(), from(0)).await.expect("its receipts");
|
|
let media = fetched_objects(&media);
|
|
assert_eq!(media.iter().map(|(g, _)| *g).collect::<Vec<_>>(), (0..G).collect::<Vec<_>>());
|
|
let verifier = ReceiptVerifier::with_anchors(vec![key.verifying_key()]);
|
|
for ((g, body), (rg, r)) in media.iter().zip(fetched_objects(&receipts)) {
|
|
assert_eq!(*g, rg);
|
|
let b = ReceiptBundle::from_payload(&r).unwrap();
|
|
let vg = verifier.verify_group(&b.group).unwrap();
|
|
verifier.verify_payload(body, &b.objects[0], &vg).expect("a fetched object verifies");
|
|
}
|
|
// A viewer without it, at an edge that knows nothing of the track yet: refused there, below
|
|
// the live edge it asked the home relay for, and served at the live edge.
|
|
let (vl, _) = connect(&edge, Via::WebSocket, with(&live_tok)).await;
|
|
let refused = vl.fetch(t.ns.clone(), t.media.clone(), from(0)).await.map(|_| ()).expect_err("refused");
|
|
assert!(refused.contains("seek_not_granted"), "{refused}");
|
|
assert_eq!(edge.relay().stats().seek_refused, 1);
|
|
let (_, last) = vl.fetch(t.ns.clone(), t.media.clone(), from(G - 1)).await.expect("the live edge");
|
|
assert_eq!(fetched_objects(&last).len(), 1);
|
|
// A SUBSCRIBE whose fill starts at group 0, at an edge that learns the live edge only from
|
|
// its upstream's SUBSCRIBE_OK: refused there once it does.
|
|
let fill = Parameters::new().with(Parameter::FillParameters(Parameters::new().with(Parameter::LocationFilter(LocationFilter::AbsoluteStart { start: Location::new(0, 0) }))));
|
|
let refused = vl.subscribe(t.ns.clone(), t.media.clone(), fill).await.map(|_| ()).expect_err("a fill from group 0, refused");
|
|
assert!(refused.contains("seek_not_granted"), "{refused}");
|
|
assert_eq!(edge.relay().stats().seek_refused, 2);
|
|
for c in [&p, &vs, &vl] {
|
|
c.close(0, "done");
|
|
}
|
|
edge.shutdown().await;
|
|
home.shutdown().await;
|
|
let _ = std::fs::remove_dir_all(&dir);
|
|
let _ = std::fs::remove_file(&peer_file);
|
|
}
|
|
|
|
// ── Co-performance across relays ───────────────────────────────────────────
|
|
|
|
/// A `meta` record of live-events §3.1, signed by `key`, after `prev` when given.
|
|
fn meta(key: &SigningKey, room: [u8; 32], seq: u64, prev: Option<&[u8]>, t: u64, payload: &[u8]) -> Vec<u8> {
|
|
let mut b = b"WEV1\x01".to_vec();
|
|
b.extend_from_slice(&room);
|
|
b.extend_from_slice(&key.verifying_key().to_bytes());
|
|
b.extend_from_slice(&seq.to_le_bytes());
|
|
b.extend_from_slice(&prev.map_or([0; 32], |p| sha256(&p[..p.len() - 64])));
|
|
b.extend_from_slice(&t.to_le_bytes());
|
|
b.push(0);
|
|
b.push(0x01);
|
|
b.extend_from_slice(&(payload.len() as u16).to_le_bytes());
|
|
b.extend_from_slice(payload);
|
|
let mut m = b"wai:live-event\x01".to_vec();
|
|
m.extend_from_slice(&b);
|
|
b.extend_from_slice(&key.sign(&m).to_bytes());
|
|
b
|
|
}
|
|
|
|
fn msf(record: &[u8]) -> Vec<u8> {
|
|
use base64::Engine;
|
|
format!("[{{\"t\":1,\"data\":{{\"wev1\":\"{}\"}}}}]", base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(record)).into_bytes()
|
|
}
|
|
|
|
/// A viewer's grant of `room` for `principal`, as live-rooms §4.1 lays it out, with `attrs`.
|
|
fn viewer_grant(authority: &SigningKey, room: [u8; 32], principal: &SigningKey, attrs: &[(&str, &str)]) -> Vec<u8> {
|
|
let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs();
|
|
let mut g = b"WRG1\x01".to_vec();
|
|
g.extend_from_slice(&room);
|
|
g.extend_from_slice(&principal.verifying_key().to_bytes());
|
|
g.extend_from_slice(&[0; 40]);
|
|
g.extend_from_slice(&(now - 60).to_le_bytes());
|
|
g.extend_from_slice(&(now + 600).to_le_bytes());
|
|
g.extend_from_slice(&[1, 0, 1, 0, 0x10, 0]);
|
|
g.extend_from_slice(&[0; 32]);
|
|
g.extend_from_slice(b"WEP1\x01");
|
|
g.extend_from_slice(&[0x0b; 32]);
|
|
g.extend_from_slice(&1024u16.to_le_bytes());
|
|
g.extend_from_slice(&500u16.to_le_bytes());
|
|
g.extend_from_slice(&50u32.to_le_bytes());
|
|
g.extend_from_slice(&1_000_000u64.to_le_bytes());
|
|
g.extend_from_slice(&1u32.to_le_bytes());
|
|
g.extend_from_slice(&1_000_000u64.to_le_bytes());
|
|
g.extend_from_slice(&[0xff; 16]);
|
|
g.extend_from_slice(&[0; 3]);
|
|
g.extend_from_slice(&[0, 0]);
|
|
let mut a = Vec::new();
|
|
for (k, v) in attrs {
|
|
a.push(k.len() as u8);
|
|
a.extend_from_slice(k.as_bytes());
|
|
a.extend_from_slice(&(v.len() as u16).to_le_bytes());
|
|
a.extend_from_slice(v.as_bytes());
|
|
}
|
|
g.extend_from_slice(&(a.len() as u16).to_le_bytes());
|
|
g.extend_from_slice(&a);
|
|
let mut m = joule_moq::grant_proof::GRANT_DOMAIN.to_vec();
|
|
m.extend_from_slice(&g);
|
|
g.extend_from_slice(&authority.sign(&m).to_bytes());
|
|
g
|
|
}
|
|
|
|
/// One object of group `g`, on a stream of its own.
|
|
async fn send(c: &Client, alias: u64, g: u64, payload: &[u8]) {
|
|
let h = SubgroupHeader { track_alias: alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(16), properties: false, end_of_group: true, first_object: true };
|
|
c.send_subgroup(h, &[SubgroupObject::normal(0, payload.to_vec())]).await.expect("an object");
|
|
}
|
|
|
|
fn room_track(room: [u8; 32], who: &SigningKey, field: &str) -> TrackNamespace {
|
|
ns(&[&hex::encode(room), &hexkey(who), field])
|
|
}
|
|
|
|
/// What `c` receives until `want` arrives, then a second more: the payloads.
|
|
async fn payloads_until(c: &Client, want: &[u8]) -> Vec<Vec<u8>> {
|
|
let mut got = collect(c, STEP, |g| g.iter().any(|x| x.objects.iter().any(|o| o.payload == want))).await;
|
|
got.extend(collect(c, Duration::from_secs(1), |_| false).await);
|
|
got.iter().flat_map(|g| g.objects.iter().map(|o| o.payload.clone())).collect()
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn co_performance_votes_are_counted_once_in_chain_order() {
|
|
timeout(Duration::from_secs(120), votes()).await.expect("in time");
|
|
}
|
|
|
|
async fn votes() {
|
|
let room = [0xb1; 32];
|
|
let authority = SigningKey::from_bytes(&[0xb2; 32]);
|
|
let clock = SigningKey::from_bytes(&[0xb3; 32]);
|
|
let (member, watcher_h, watcher_e) = (SigningKey::from_bytes(&[0xb4; 32]), SigningKey::from_bytes(&[0xb5; 32]), SigningKey::from_bytes(&[0xb6; 32]));
|
|
let (clock_tok, peer_tok) = (b"the clock's token".to_vec(), b"the edge relay".to_vec());
|
|
let (peer_sha, peer_file) = token_file("votes-peer", &peer_tok);
|
|
let floor = FloorConfig { room_keys: vec![hexkey(&clock)], hold_ms: 60_000 };
|
|
// The clock and a watcher on the home relay; the member and another watcher on the edge.
|
|
let home = relay(|c| {
|
|
quick(c);
|
|
c.auth.required = true;
|
|
c.auth.authorities = vec![Authority { key: hexkey(&authority), rooms: vec![] }];
|
|
c.auth.grants = vec![
|
|
Grant { prefix: vec!["*".into(), hexkey(&clock), "events".into()], token_sha256: hex::encode(sha256(&clock_tok)), publish: true, subscribe: true, proof: false, unlimited: true, seek: false },
|
|
Grant { prefix: vec![], token_sha256: peer_sha.clone(), publish: true, subscribe: true, proof: false, unlimited: true, seek: true },
|
|
];
|
|
c.floor = Some(floor.clone());
|
|
})
|
|
.await;
|
|
let edge = relay(|c| {
|
|
quick(c);
|
|
c.auth.required = true;
|
|
c.auth.authorities = vec![Authority { key: hexkey(&authority), rooms: vec![] }];
|
|
c.floor = Some(floor.clone());
|
|
let mut u = upstream_to(&home, Via::WebSocket, &[&["*"]]);
|
|
u.token_sha256 = Some(peer_sha.clone());
|
|
u.token_path = Some(peer_file.clone());
|
|
c.upstream = vec![u];
|
|
})
|
|
.await;
|
|
linked(&edge, 1).await;
|
|
let t0 = now_us() - 10_000_000;
|
|
let target = |s: &RelayServer| format!("//{}{}", s.websocket_addr.unwrap(), s.relay().config().path);
|
|
let with = |s: &RelayServer, g: Vec<u8>, k: &SigningKey| ClientOptions { grants: vec![(g, k.clone())], target: target(s), ..ClientOptions::default() };
|
|
|
|
let (c, _) = connect(&home, Via::WebSocket, ClientOptions { tokens: vec![Token::UseValue { token_type: 0, value: clock_tok.clone() }], ..ClientOptions::default() }).await;
|
|
let _ce = c.publish(room_track(room, &clock, "events"), name("0"), 1, Parameters::new(), track_props(16, None)).await.expect("PUBLISH the clock's events");
|
|
// The edge's watcher follows the clock's events first, so the edge reads the window as it
|
|
// passes, over its link.
|
|
let (we, _) = connect(&edge, Via::WebSocket, with(&edge, viewer_grant(&authority, room, &watcher_e, &[]), &watcher_e)).await;
|
|
let _wec = we.subscribe(room_track(room, &clock, "events"), name("0"), Parameters::new()).await.expect("the edge's watcher, of the clock's events");
|
|
let mut w = b"WPW1\x01".to_vec();
|
|
w.extend_from_slice(&[1, 0b100, 3, 1]);
|
|
w.extend_from_slice(&(now_us() - t0 + 60_000_000).to_le_bytes());
|
|
w.extend_from_slice(&[0; 32]);
|
|
w.extend_from_slice(&0u16.to_le_bytes());
|
|
let wr = meta(&clock, room, 0, None, now_us() - t0, &w);
|
|
let window = sha256(&wr[..wr.len() - 64]);
|
|
send(&c, 1, 0, &msf(&wr)).await;
|
|
let (fh, fe) = (home.relay().floor().unwrap(), edge.relay().floor().unwrap());
|
|
assert!(until(STEP, || fh.holds_window(&room, &window) && fe.holds_window(&room, &window)).await, "both relays hold the clock's window");
|
|
|
|
// The member publishes its events on the edge; a watcher on each relay follows them.
|
|
let (pm, _) = connect(&edge, Via::WebSocket, with(&edge, viewer_grant(&authority, room, &member, &[("audience_input", "1")]), &member)).await;
|
|
let _pme = pm.publish(room_track(room, &member, "events"), name("0"), 2, Parameters::new(), track_props(16, None)).await.expect("PUBLISH the member's events");
|
|
let events = room_track(room, &member, "events");
|
|
assert!(until(STEP, || home.relay().namespaces().iter().any(|(n, _, _)| *n == events)).await, "the edge publishes the member's namespace upstream");
|
|
let (wh, _) = connect(&home, Via::WebSocket, with(&home, viewer_grant(&authority, room, &watcher_h, &[]), &watcher_h)).await;
|
|
let _whm = wh.subscribe(room_track(room, &member, "events"), name("0"), Parameters::new()).await.expect("the home watcher, of the member's events, through the edge");
|
|
let _wem = we.subscribe(room_track(room, &member, "events"), name("0"), Parameters::new()).await.expect("the edge watcher, of the member's events");
|
|
let vote = |seq: u64, prev: Option<&[u8]>, o: u8| {
|
|
let mut b = b"WPA1\x01".to_vec();
|
|
b.extend_from_slice(&window);
|
|
b.extend_from_slice(&[3, o]);
|
|
meta(&member, room, seq, prev, now_us() - t0, &b)
|
|
};
|
|
let first = vote(0, None, 1);
|
|
let second = vote(1, Some(&first), 2);
|
|
// Seq 1 first, on a stream of its own: the edge holds it for seq 0, counts seq 0, and then
|
|
// refuses seq 1, past the member's power.
|
|
send(&pm, 2, 1, &msf(&second)).await;
|
|
assert!(until(STEP, || fe.waiting(&room) == 1).await, "the edge holds seq 1 for seq 0");
|
|
send(&pm, 2, 0, &msf(&first)).await;
|
|
let want = msf(&first);
|
|
let (got_h, got_e) = tokio::join!(payloads_until(&wh, &want), payloads_until(&we, &want));
|
|
let only_first = vec![msf(&first)];
|
|
assert_eq!(got_e.iter().filter(|p| **p != msf(&wr)).cloned().collect::<Vec<_>>(), only_first, "the edge's watcher: seq 0 alone");
|
|
assert_eq!(got_h, only_first, "the home watcher is given exactly what the edge's is");
|
|
// Seq 1 is refused in its own object's task, woken once seq 0 is counted: nothing the
|
|
// watchers were given waits for that, so the refusal is waited for, then counted.
|
|
assert!(until(STEP, || fe.refused_inputs() >= 1).await, "the edge refuses seq 1");
|
|
assert_eq!((fe.refused_inputs(), fh.refused_inputs()), (1, 0), "decided once, at the edge, where the member publishes");
|
|
for x in [&c, &we, &pm, &wh] {
|
|
x.close(0, "done");
|
|
}
|
|
let er = edge.relay().clone();
|
|
edge.shutdown().await;
|
|
home.shutdown().await;
|
|
let _ = std::fs::remove_file(&peer_file);
|
|
// The watchers were read for a second past seq 0, so a refusal counted and then forwarded
|
|
// later than that would pass them. What the edge gave its watcher, counted once its
|
|
// sessions ended, is seq 0 alone.
|
|
assert!(until(STEP, || er.summaries().len() >= 3).await, "the edge's sessions' summaries");
|
|
let member_events = joule_moq::draft22::names::render(&room_track(room, &member, "events"), Some(&name("0")));
|
|
let given: u64 = er.summaries().iter().filter(|s| s.summary.tally.peer.is_none()).filter_map(|s| s.summary.tally.subscriber.tracks.get(&member_events).cloned()).map(|t| t.objects_forwarded + t.undelivered_at_close).sum();
|
|
assert_eq!(given, 1, "the edge gave its watcher seq 0 alone");
|
|
}
|
|
|
|
/// A `WPT1` schedule (co-performance §6.1) under refuse: from `from`, one segment of two turns
|
|
/// of `len` µs, held by `holders` in turn.
|
|
fn schedule(t0: u64, from: u64, len: u32, holders: &[&SigningKey]) -> Vec<u8> {
|
|
let mut b = b"WPT1\x01".to_vec();
|
|
b.extend_from_slice(&t0.to_le_bytes());
|
|
b.extend_from_slice(&from.to_le_bytes());
|
|
b.push(1);
|
|
b.extend_from_slice(&0u32.to_le_bytes());
|
|
b.push(0);
|
|
b.push(1);
|
|
b.extend_from_slice(&from.to_le_bytes());
|
|
b.extend_from_slice(&len.to_le_bytes());
|
|
b.extend_from_slice(&2u16.to_le_bytes());
|
|
b.push(0);
|
|
b.push(holders.len() as u8);
|
|
for h in holders {
|
|
b.extend_from_slice(&h.verifying_key().to_bytes());
|
|
}
|
|
b
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn the_floor_is_held_once_where_the_performer_publishes() {
|
|
timeout(Duration::from_secs(120), floor_once()).await.expect("in time");
|
|
}
|
|
|
|
async fn floor_once() {
|
|
let room = [0xc7; 32];
|
|
let clock = SigningKey::from_bytes(&[0xc8; 32]);
|
|
let (x, y) = (SigningKey::from_bytes(&[0xc9; 32]), SigningKey::from_bytes(&[0xca; 32]));
|
|
// The home relay holds the floor; the edge, where X publishes, holds none.
|
|
let home = relay(|c| {
|
|
quick(c);
|
|
c.floor = Some(FloorConfig { room_keys: vec![hexkey(&clock)], ..FloorConfig::default() });
|
|
})
|
|
.await;
|
|
let edge = relay(|c| {
|
|
quick(c);
|
|
c.upstream = vec![upstream_to(&home, Via::WebSocket, &[&["*"]])];
|
|
})
|
|
.await;
|
|
linked(&edge, 1).await;
|
|
// The clock's schedule on the home relay gives Y the turn now and for a minute.
|
|
let t0 = now_us();
|
|
let (c, _) = connect(&home, Via::WebSocket, ClientOptions::default()).await;
|
|
let _ce = c.publish(room_track(room, &clock, "events"), name("0"), 1, Parameters::new(), track_props(16, None)).await.expect("PUBLISH the clock's events");
|
|
send(&c, 1, 0, &msf(&meta(&clock, room, 0, None, 0, &schedule(t0, 0, 60_000_000, &[&y, &x])))).await;
|
|
let fh = home.relay().floor().unwrap();
|
|
let audio = room_track(room, &x, "audio");
|
|
assert!(until(STEP, || fh.place(&audio, now_us()).is_some()).await, "the home relay holds the schedule");
|
|
assert_eq!(fh.place(&audio, now_us()), Some(joule_moq_relay::floor::Place::OutOfTurn(1)), "X is out of turn at the home relay");
|
|
// X publishes on the edge, a viewer on the home relay follows: what the edge forwarded is
|
|
// forwarded, not held to the floor a second time by its arrival a link later.
|
|
let (px, _) = connect(&edge, Via::WebSocket, ClientOptions::default()).await;
|
|
let _pa = px.publish(audio.clone(), name("0"), 5, Parameters::new(), track_props(16, None)).await.expect("PUBLISH X's audio on the edge");
|
|
let (v, _) = connect(&home, Via::WebSocket, ClientOptions::default()).await;
|
|
assert!(until(STEP, || home.relay().namespaces().iter().any(|(n, _, _)| *n == audio)).await, "the edge publishes X's namespace upstream");
|
|
let _sv = v.subscribe(audio.clone(), name("0"), Parameters::new()).await.expect("the viewer, of X's audio, through the edge");
|
|
for g in 0..3 {
|
|
send(&px, 5, g, format!("x{g}").as_bytes()).await;
|
|
}
|
|
let got = collect(&v, STEP, |g| g.len() >= 3).await;
|
|
assert_eq!(got.len(), 3, "every object the edge forwarded reaches the home relay's viewer");
|
|
assert_eq!(fh.refused(), 0, "the home relay refused none of them");
|
|
for k in [&c, &px, &v] {
|
|
k.close(0, "done");
|
|
}
|
|
edge.shutdown().await;
|
|
home.shutdown().await;
|
|
}
|
|
|
|
// ── A relay path, where it arrives ─────────────────────────────────────────
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn a_relay_path_is_checked_where_it_arrives() {
|
|
timeout(Duration::from_secs(60), arrives()).await.expect("in time");
|
|
}
|
|
|
|
/// A client that declares a relay id, and proves it unless `prove` is false.
|
|
async fn as_relay(server: &RelayServer, key: &SigningKey, prove: bool) -> Result<Client, String> {
|
|
as_relay_at(server, key, prove, joule_moq::draft::Draft::D22).await
|
|
}
|
|
|
|
/// [`as_relay`], the session at `draft` over the WebSocket binding.
|
|
async fn as_relay_at(server: &RelayServer, key: &SigningKey, prove: bool, draft: joule_moq::draft::Draft) -> Result<Client, String> {
|
|
let url = format!("ws://{}{}", server.websocket_addr.unwrap(), server.relay().config().path);
|
|
let dial = || async {
|
|
let conn = joule_moq_relay::transport::websocket::connect_draft(&url, joule_moq_relay::transport::mux::MuxConfig::default(), draft).await.map_err(|e| e.to_string())?;
|
|
Ok::<_, String>(joule_moq_relay::transport::bridge::wrap(std::sync::Arc::new(conn), draft, joule_moq::draft22::Perspective::Client))
|
|
};
|
|
let opts = ClientOptions {
|
|
extra_setup: vec![joule_moq::draft22::kvp::KeyValuePair::bytes(joule_moq_relay::cascade::RELAY_ID_OPTION, key.verifying_key().to_bytes().to_vec())],
|
|
room_keys: if prove { vec![key.clone()] } else { Vec::new() },
|
|
target: format!("//{}{}", server.websocket_addr.unwrap(), server.relay().config().path),
|
|
..ClientOptions::default()
|
|
};
|
|
// The relay reads this SETUP before it sends its own: the id is proved over a challenge an
|
|
// earlier SETUP of its gave, so the first session is given one and the second proves.
|
|
Client::connect_proving(dial, opts).await
|
|
}
|
|
|
|
async fn closed_with(c: &Client) -> u64 {
|
|
match timeout(STEP, c.conn.closed()).await.expect("the session closes in time") {
|
|
joule_moq_relay::transport::TransportError::Closed { code, .. } => code,
|
|
other => panic!("closed otherwise: {other}"),
|
|
}
|
|
}
|
|
|
|
async fn arrives() {
|
|
let server = relay(quick).await;
|
|
let me = server.relay().relay_id();
|
|
let peer = SigningKey::from_bytes(&[0x4e; 32]);
|
|
let pid = peer.verifying_key().to_bytes();
|
|
let path = |p: Vec<[u8; 32]>| Parameters::new().with(Parameter::RelayPath(p));
|
|
// A relay whose id is proved: a path naming the relay that receives it is refused
|
|
// relay_loop, one past the bound relay_hops_exceeded, and one whose last relay is not the
|
|
// peer closes the session.
|
|
let c = as_relay(&server, &peer, true).await.expect("a peer's SETUP");
|
|
let r = c.subscribe(ns(&["live", "x"]), name("y"), path(vec![me, pid])).await.map(|_| ()).expect_err("refused");
|
|
assert!(r.contains("DOES_NOT_EXIST") && r.contains("relay_loop"), "{r}");
|
|
let r = c.subscribe(ns(&["live", "x"]), name("y"), path([[7u8; 32]; 8].into_iter().chain([pid]).collect())).await.map(|_| ()).expect_err("refused");
|
|
assert!(r.contains("relay_hops_exceeded"), "{r}");
|
|
let r = c.publish_namespace(ns(&["live", "x"]), path(vec![me, pid])).await.map(|_| ()).expect_err("refused");
|
|
assert!(r.contains("UNINTERESTED") && r.contains("relay_loop"), "{r}");
|
|
assert_eq!(server.relay().stats().relay_loops, 3);
|
|
let _ = c.subscribe(ns(&["live", "x"]), name("y"), path(vec![[9; 32]])).await;
|
|
assert_eq!(closed_with(&c).await, joule_moq::draft22::SessionError::PROTOCOL_VIOLATION.0, "a path whose last relay is not the peer");
|
|
// A relay id declared without its proof is refused at SETUP, the reason stated.
|
|
let refused = as_relay(&server, &peer, false).await.err().expect("a relay id without its proof");
|
|
assert!(refused.contains("relay_id_unproved"), "{refused}");
|
|
// RELAY_PATH from a session that negotiated none closes it PROTOCOL_VIOLATION (§9.20).
|
|
let (plain, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let _ = plain.subscribe(ns(&["live", "x"]), name("y"), path(vec![pid])).await;
|
|
assert_eq!(closed_with(&plain).await, joule_moq::draft22::SessionError::PROTOCOL_VIOLATION.0, "RELAY_PATH not negotiated");
|
|
|
|
// The same at -23, where relays peer too: the path is negotiated by RELAY_ID, read by the
|
|
// -23 codec and bridged, a loop refused and the hop limit held; and from a session that
|
|
// negotiated none it is an unknown parameter, INVALID_PARAMETER (-23 §9.19).
|
|
use joule_moq::draft::Draft;
|
|
let c = as_relay_at(&server, &peer, true, Draft::D23).await.expect("a peer's SETUP at -23");
|
|
assert_eq!(c.conn.draft(), Draft::D23);
|
|
let loops = server.relay().stats().relay_loops;
|
|
let r = c.subscribe(ns(&["live", "x"]), name("y"), path(vec![me, pid])).await.map(|_| ()).expect_err("refused");
|
|
assert!(r.contains("DOES_NOT_EXIST") && r.contains("relay_loop"), "{r}");
|
|
let r = c.subscribe(ns(&["live", "x"]), name("y"), path([[7u8; 32]; 8].into_iter().chain([pid]).collect())).await.map(|_| ()).expect_err("refused");
|
|
assert!(r.contains("relay_hops_exceeded"), "{r}");
|
|
let r = c.publish_namespace(ns(&["live", "x"]), path(vec![me, pid])).await.map(|_| ()).expect_err("refused");
|
|
assert!(r.contains("UNINTERESTED") && r.contains("relay_loop"), "{r}");
|
|
assert_eq!(server.relay().stats().relay_loops, loops + 3, "each refused for its path, at -23");
|
|
assert!(c.conn.notes().iter().all(|n| !n.contains("RELAY_PATH")), "the bridge carried the path, and dropped none: {:?}", c.conn.notes());
|
|
let _ = c.subscribe(ns(&["live", "x"]), name("y"), path(vec![[9; 32]])).await;
|
|
assert_eq!(closed_with(&c).await, joule_moq::draft23::codes::SessionError::INVALID_PARAMETER.0, "a path whose last relay is not the peer, at -23");
|
|
let (plain, _) = connect(&server, Via::WsDraft(Draft::D23), ClientOptions::default()).await;
|
|
let _ = plain.subscribe(ns(&["live", "x"]), name("y"), path(vec![pid])).await;
|
|
assert_eq!(closed_with(&plain).await, joule_moq::draft23::codes::SessionError::INVALID_PARAMETER.0, "RELAY_PATH not negotiated, at -23");
|
|
// At -18 and -16 there is no form for it: a relay id declared there is left unread, the
|
|
// session an ordinary one, and its proof alone makes it no peer.
|
|
let declares = ClientOptions { extra_setup: vec![joule_moq::draft22::kvp::KeyValuePair::bytes(joule_moq_relay::cascade::RELAY_ID_OPTION, pid.to_vec())], ..ClientOptions::default() };
|
|
let (old, _ep) = connect(&server, Via::QuicDraft(Draft::D18), declares).await;
|
|
let r = old.subscribe(ns(&["live", "x"]), name("y"), path(vec![pid])).await.map(|_| ()).expect_err("a path at -18");
|
|
assert!(old.conn.notes().iter().any(|n| n.contains("RELAY_PATH")) || r.contains("NOT_SUPPORTED"), "refused by name at -18: {r}; {:?}", old.conn.notes());
|
|
old.close(0, "done");
|
|
server.shutdown().await;
|
|
}
|
|
|
|
// ── SUBSCRIBE_TRACKS across relays ─────────────────────────────────────────
|
|
|
|
/// SUBSCRIBE_TRACKS for `prefix` on `c`, its answer, and its stream.
|
|
async fn tracks_request(c: &Client, prefix: TrackNamespace) -> (ControlMessage, joule_moq_relay::wire::ReqTx, ReqRx) {
|
|
let id = c.next_request_id();
|
|
let (tx, mut rx) = c.request(&ControlMessage::SubscribeTracks(NamespaceRequest { request_id: id, prefix, params: Parameters::new() })).await.expect("the request goes");
|
|
let answer = Client::response(&mut rx).await.expect("an answer");
|
|
(answer, tx, rx)
|
|
}
|
|
|
|
/// The next PUBLISH the relay opens to `c`, within `wait`, accepted (PUBLISH_OK); `None` if
|
|
/// none comes.
|
|
async fn publish_offered(c: &Client, wait: Duration) -> Option<(Publish, joule_moq_relay::wire::ReqTx, ReqRx)> {
|
|
let (tx, rx, first) = c.next_request(wait).await?;
|
|
let ControlMessage::Publish(p) = first else { panic!("the relay opened {} where a PUBLISH was expected", first.name()) };
|
|
tx.send(&ControlMessage::RequestOk(RequestOk { params: Parameters::new(), track_properties: Default::default() })).await.expect("PUBLISH_OK goes");
|
|
Some((p, tx, rx))
|
|
}
|
|
|
|
/// One object of group `g`, on the track published under `alias`.
|
|
async fn one_object(c: &Client, alias: u64, g: u64, payload: &[u8]) {
|
|
let h = SubgroupHeader { track_alias: alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(128), properties: false, end_of_group: false, first_object: true };
|
|
c.send_subgroup(h, &[SubgroupObject::normal(0, payload.to_vec())]).await.expect("the object goes");
|
|
}
|
|
|
|
/// Whether an object of the subscription `alias` carrying `payload` reaches `c`.
|
|
async fn reaches(c: &Client, alias: u64, payload: &[u8]) -> bool {
|
|
let carries = |g: &Got| g.header.track_alias == alias && g.objects.iter().any(|o| o.payload == payload);
|
|
collect(c, STEP, |got| got.iter().any(carries)).await.iter().any(carries)
|
|
}
|
|
|
|
/// SUBSCRIBE_TRACKS across relays (-22 §7.6, -23 §3.6): a subscriber of the edge asks for
|
|
/// every track; the edge takes the request to its upstream link, and the tracks the home relay
|
|
/// holds under what the link routes, now and later, come back as PUBLISHes the edge accepts and
|
|
/// offers on, their objects arriving on the edge's PUBLISH. A track the link does not route is
|
|
/// refused UNINTERESTED at the edge and offered to no one. Two requests of the edge that ask for
|
|
/// the same thing of the link are one SUBSCRIBE_TRACKS on it; the last to go cancels it, and the
|
|
/// home relay offers the edge nothing more.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn subscribe_tracks_reaches_an_upstream_relay() {
|
|
timeout(Duration::from_secs(120), tracks_across()).await.expect("in time");
|
|
}
|
|
|
|
async fn tracks_across() {
|
|
let home = relay(quick).await;
|
|
// The link routes live/<any>: a prefix with a wildcard field, so the edge asks the home
|
|
// relay for the request's own prefix and holds what comes back to the link's.
|
|
let edge = relay(|c| {
|
|
quick(c);
|
|
c.upstream = vec![upstream_to(&home, Via::Quic, &[&["live", "*"]])];
|
|
})
|
|
.await;
|
|
linked(&edge, 1).await;
|
|
let (hr, er) = (home.relay().clone(), edge.relay().clone());
|
|
let (pa, _ea) = connect(&home, Via::Quic, ClientOptions::default()).await;
|
|
let (_a, _arx) = pa.publish(ns(&["live", "a"]), name("cam"), 71, Parameters::new(), Default::default()).await.expect("PUBLISH live/a on the home relay");
|
|
let (_x, _xrx) = pa.publish(ns(&["other", "x"]), name("cam"), 72, Parameters::new(), Default::default()).await.expect("PUBLISH other/x on the home relay");
|
|
|
|
// A subscriber of the edge asks for every track (a prefix of no field).
|
|
let (se, _es) = connect(&edge, Via::Quic, ClientOptions::default()).await;
|
|
let (answer, se_tx, _se_rx) = tracks_request(&se, ns(&[])).await;
|
|
assert!(matches!(answer, ControlMessage::RequestOk(_)), "{answer:?}");
|
|
let (first, _ftx, _frx) = publish_offered(&se, STEP).await.expect("a PUBLISH, at the edge, of the home relay's track");
|
|
assert_eq!((first.namespace.clone(), first.name.clone()), (ns(&["live", "a"]), name("cam")));
|
|
one_object(&pa, 71, 0, b"home-a-0").await;
|
|
assert!(reaches(&se, first.track_alias, b"home-a-0").await, "an object of the home relay's track reaches the edge's subscriber");
|
|
|
|
// A track the home relay gains later, under the link's prefix: offered too.
|
|
let (_b, _brx) = pa.publish(ns(&["live", "b"]), name("cam"), 73, Parameters::new(), Default::default()).await.expect("PUBLISH live/b on the home relay");
|
|
let (second, _stx, _srx) = publish_offered(&se, STEP).await.expect("a PUBLISH of the track published later");
|
|
assert_eq!((second.namespace.clone(), second.name.clone()), (ns(&["live", "b"]), name("cam")));
|
|
one_object(&pa, 73, 0, b"home-b-0").await;
|
|
assert!(reaches(&se, second.track_alias, b"home-b-0").await, "an object of the later track reaches the edge's subscriber");
|
|
// Nothing the link does not route.
|
|
assert!(publish_offered(&se, Duration::from_millis(800)).await.is_none(), "other/x, which the link does not route, is not offered at the edge");
|
|
assert_eq!(er.largest(&(ns(&["other", "x"]), name("cam"))), None, "nor held there");
|
|
|
|
// A second request of the edge for the same prefix: the one SUBSCRIBE_TRACKS on the link.
|
|
let (se2, _es2) = connect(&edge, Via::Quic, ClientOptions::default()).await;
|
|
let (answer, se2_tx, _se2_rx) = tracks_request(&se2, ns(&[])).await;
|
|
assert!(matches!(answer, ControlMessage::RequestOk(_)), "{answer:?}");
|
|
assert_eq!(er.tracks_up(), vec![(0, ns(&[]), 2)], "one SUBSCRIBE_TRACKS on the link, held by both requests");
|
|
assert_eq!(hr.subscribe_tracks_held().len(), 1, "the home relay holds one SUBSCRIBE_TRACKS of the edge's");
|
|
|
|
// Both requests cancelled: the edge cancels its own on the link, and a track the home relay
|
|
// gains after that does not reach the edge.
|
|
for tx in [&se_tx, &se2_tx] {
|
|
tx.reset(joule_moq::draft22::StreamResetCode::CANCELLED.0).await;
|
|
}
|
|
assert!(until(STEP, || er.tracks_up().is_empty()).await, "the edge lets its SUBSCRIBE_TRACKS go: {:?}", er.tracks_up());
|
|
assert!(until(STEP, || hr.subscribe_tracks_held().is_empty()).await, "the home relay's copy of it is cancelled: {:?}", hr.subscribe_tracks_held());
|
|
let (_c, _crx) = pa.publish(ns(&["live", "c"]), name("cam"), 74, Parameters::new(), Default::default()).await.expect("PUBLISH live/c on the home relay");
|
|
one_object(&pa, 74, 0, b"home-c-0").await;
|
|
assert!(until(STEP, || hr.largest(&(ns(&["live", "c"]), name("cam"))).is_some()).await, "the home relay holds live/c");
|
|
tokio::time::sleep(Duration::from_millis(500)).await;
|
|
assert_eq!(er.largest(&(ns(&["live", "c"]), name("cam"))), None, "a track published after the cancellation does not reach the edge");
|
|
// The subscriptions made before stand (-23 §3.6): the edge still holds live/b.
|
|
one_object(&pa, 73, 1, b"home-b-1").await;
|
|
assert!(reaches(&se, second.track_alias, b"home-b-1").await, "the subscription the cancelled request made stands");
|
|
for c in [&pa, &se, &se2] {
|
|
c.close(0, "done");
|
|
}
|
|
edge.shutdown().await;
|
|
home.shutdown().await;
|
|
}
|