8f8f9cd8db
The relay held each ended session's summary until its operator acknowledged it, and past summary.held_max (65 536) with no spill file refused every new session summaries_unacknowledged. A wai-relay with no [control] has nothing that can acknowledge: the live room's relay and its edge are configured so, and could only fill and then refuse. Decision D95. - RelayConfig::summaries_handed_over: no control.socket, control.listen or control.flush_interval_ms, and a summary.out_path. Such a relay writes each summary to the file whole and synced and holds none (Outbox::hand_over, Outbox::keep_as); a summary whose write fails is held as before and counts toward the bound. - The operator is told: a warning when what is held unacknowledged reaches half the bound and an error when it passes it, once each until acknowledgements bring it under half. At start the program says where its summaries go, and with neither a surface nor an out_path that they reach no one. A failed write to out_path is logged; it was dropped silently. - tests/window_cuts.rs: the live room's configuration under a bound of 2, six sessions (four past it), each admitted, each summary whole on the file and verifying, none held; planted and caught: a surface never acknowledged (told twice, the session past the bound refused) and a file that cannot be written (all held, the session past the bound refused). src/outbox.rs drives the default bound: 65 537 handed over, none held or refused; held instead, refused and told twice. Co-Authored-By: Claude Opus 5.5 <[email protected]>
600 lines
30 KiB
Rust
600 lines
30 KiB
Rust
//! A window closed from the operator's surface by a relay of its own process, at one cut across
|
|
//! the relays that carry a room, and the signed summaries drained (`src/window.rs`,
|
|
//! `src/outbox.rs`, `src/control.rs`). Each check is held against a fault planted for it:
|
|
//!
|
|
//! - **A window's counts by session, bound to its record.** The surface's `window` answers a
|
|
//! sealed window's parts signed by the relay; the room's `record` over the window recomputes
|
|
//! from them, so the operator closes the window from the parts. Planted: a part's count raised
|
|
//! and signed by the relay's own key, a session's part left out and signed so, a part changed
|
|
//! after it was signed, and the next window's parts set against this window's record: each
|
|
//! refused.
|
|
//! - **Relays that carry a room close a window at one cut.** A core and two edges, each a viewer
|
|
//! on it; the edges `close` before a group begins and the core after, so the group began
|
|
//! between their instants. Joined, the cut puts it in the closing window at all three, and the
|
|
//! core's and the first edge's statements of the link chain exactly in that window and the
|
|
//! next. Planted: the second edge seals at its own cut, as an edge cutting at its own instant
|
|
//! would, and the chain is caught broken, short in the first window and the contradiction of
|
|
//! live-metering §11 in the next; and a cut below the core's own is refused, nothing sealed.
|
|
//! - **An edge rides through a window end.** Its viewer is sent no GOAWAY across two window ends,
|
|
//! stays open, and receives every group once. Planted: the second edge ends its window by a
|
|
//! restart, as edges did before, and the same check catches the GOAWAY its viewer is sent.
|
|
//! - **Summaries drained.** An acknowledged summary is freed and never handed out again; an
|
|
//! acknowledgement past what was handed out frees nothing; past `summary.held_max` with no
|
|
//! spill file every summary is held and a new session refused `summaries_unacknowledged`
|
|
//! until the operator acknowledges; with a spill file the summaries go there, come back in
|
|
//! order, and a restart hands over those not acknowledged, and none that were.
|
|
//! - **Summaries handed over on a file** (WAI decision D95). A relay configured as the live
|
|
//! room's is, with `summary.out_path` and no operator's surface, under a bound of 2: five
|
|
//! sessions, past the bound, each admitted, each summary on the file whole and verifying, none
|
|
//! held. Planted: the same relay with a surface its operator never acknowledges on refuses the
|
|
//! session past the bound `summaries_unacknowledged`, after telling the operator at half the
|
|
//! bound and past it; and the same relay whose file cannot be written holds the summaries and
|
|
//! refuses past the bound, never dropping one.
|
|
|
|
#![cfg(unix)]
|
|
|
|
mod common;
|
|
|
|
use common::*;
|
|
use ed25519_dalek::SigningKey;
|
|
use joule_moq::draft22::{Parameters, SessionError};
|
|
use joule_moq_relay::client::{Client, ClientOptions};
|
|
use joule_moq_relay::config::{UpstreamConfig, UpstreamTransport};
|
|
use joule_moq_relay::metering::HopRecord;
|
|
use joule_moq_relay::transport::{quic, TransportError};
|
|
use joule_moq_relay::window::{link_chain, Cut, SignedWindow};
|
|
use joule_moq_relay::RelayServer;
|
|
use serde_json::{json, Value};
|
|
use std::path::{Path, PathBuf};
|
|
use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
|
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
|
|
use tokio::time::timeout;
|
|
|
|
fn now_us() -> u64 {
|
|
SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_micros() as u64
|
|
}
|
|
|
|
fn hexkey(k: &SigningKey) -> String {
|
|
hex::encode(k.verifying_key().to_bytes())
|
|
}
|
|
|
|
/// A socket path of this test's own, short enough for every host's limit.
|
|
fn socket_path(what: &str) -> PathBuf {
|
|
std::env::temp_dir().join(format!("wai-relay-cut-{}-{what}.sock", std::process::id()))
|
|
}
|
|
|
|
/// A directory of this test's own, empty.
|
|
fn dir(what: &str) -> PathBuf {
|
|
let d = std::env::temp_dir().join(format!("wai-relay-cut-{}-{what}", std::process::id()));
|
|
let _ = std::fs::remove_dir_all(&d);
|
|
std::fs::create_dir_all(&d).unwrap();
|
|
d
|
|
}
|
|
|
|
/// One request on the control socket at `path`, and its answer.
|
|
async fn ask(path: &Path, req: Value) -> Value {
|
|
timeout(STEP * 2, async {
|
|
let s = tokio::net::UnixStream::connect(path).await.expect("the control socket");
|
|
let (r, mut w) = s.into_split();
|
|
w.write_all(format!("{req}\n").as_bytes()).await.unwrap();
|
|
let mut line = String::new();
|
|
BufReader::new(r).read_line(&mut line).await.unwrap();
|
|
serde_json::from_str::<Value>(&line).unwrap_or_else(|e| panic!("an answer: {e}: {line:?}"))
|
|
})
|
|
.await
|
|
.expect("an answer in time")
|
|
}
|
|
|
|
/// A request the surface must answer `ok`.
|
|
async fn ok(path: &Path, req: Value) -> Value {
|
|
let a = ask(path, req.clone()).await;
|
|
assert_eq!(a["ok"], true, "{req} answered {a}");
|
|
a
|
|
}
|
|
|
|
/// The sealed window's statement, of `room`, from the surface.
|
|
async fn statement(path: &Path, window: u64, room: &str) -> SignedWindow {
|
|
let a = ok(path, json!({"op": "window", "window": window, "room": room})).await;
|
|
serde_json::from_value(a["statement"].clone()).expect("a signed window")
|
|
}
|
|
|
|
/// The room's record over the sealed window, from the surface.
|
|
async fn record(path: &Path, room: &str, t0: u64, window: u64) -> HopRecord {
|
|
let a = ok(path, json!({"op": "record", "room": room, "t0_unix_us": t0, "window": window})).await;
|
|
let h = |k: &str| hex::decode(a[k].as_str().unwrap()).unwrap();
|
|
HopRecord::decode(&h("record"), &h("ledger")).expect("the record reads")
|
|
}
|
|
|
|
/// An upstream entry for `server` over its WebSocket listener, the port it bound itself.
|
|
fn upstream_to(server: &RelayServer, prefixes: &[&[&str]]) -> UpstreamConfig {
|
|
UpstreamConfig {
|
|
name: "core".into(),
|
|
url: format!("ws://{}{}", server.websocket_addr.unwrap(), server.relay().config().path),
|
|
transport: UpstreamTransport::WebSocket,
|
|
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: Some(hex::encode(server.relay().relay_id())),
|
|
}
|
|
}
|
|
|
|
/// Media and receipts objects forwarded by every session of `server` open now.
|
|
fn forwarded(server: &RelayServer) -> u64 {
|
|
server.relay().open_tallies().iter().map(|(_, t)| t.subscriber.objects_forwarded).sum()
|
|
}
|
|
|
|
/// The viewer rode through: it was sent no GOAWAY and its session is open.
|
|
fn rode_through(v: &Client) -> Result<(), String> {
|
|
if let Some(g) = v.goaway().borrow().clone() {
|
|
return Err(format!("the viewer was sent GOAWAY (new session URI {:?})", String::from_utf8_lossy(&g.new_session_uri)));
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
// ── A window's counts by session, bound to its record ──────────────────────
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn a_windows_counts_by_session_are_bound_to_its_record() {
|
|
timeout(Duration::from_secs(180), parts_run()).await.expect("the test finishes in time");
|
|
}
|
|
|
|
async fn parts_run() {
|
|
let t0 = now_us() - 1_000_000;
|
|
let room = [0x61; 32];
|
|
let r = hex::encode(room);
|
|
let pk = SigningKey::from_bytes(&[0x62; 32]);
|
|
let d = dir("parts");
|
|
let seed = [0x63u8; 32];
|
|
std::fs::write(d.join("relay.key"), hex::encode(seed)).unwrap();
|
|
let sock = socket_path("parts");
|
|
let server = relay(|c| {
|
|
c.control.socket = Some(sock.clone());
|
|
c.control.flush_grace_ms = 200;
|
|
c.summary.key_path = Some(d.join("relay.key"));
|
|
c.metering.energy = joule_moq_relay::config::EnergyMode::Off;
|
|
})
|
|
.await;
|
|
let t = Tracks::new(ns(&[&r, &hexkey(&pk), "video"]), "0", 10, 11, 64);
|
|
let media = joule_moq::draft22::names::render(&t.ns, Some(&t.media));
|
|
let (p, _ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
|
|
let _published = publish_tracks(&p, &t, None).await;
|
|
let (v, _ev) = connect(&server, Via::Quic, ClientOptions::default()).await;
|
|
let s = subscribe_both(&v, &t, Parameters::new()).await;
|
|
let mut got = Vec::new();
|
|
for (window, groups) in [(0u64, 0..3u64), (1, 3..5)] {
|
|
for g in groups.clone() {
|
|
publish_group(&p, &pk, &t, g, &payload("w", g, 300)).await;
|
|
}
|
|
got.extend(collect(&v, STEP, |x| x.len() as u64 >= 2 * (groups.end - groups.start)).await);
|
|
assert!(until(STEP, || forwarded(&server) == 2 * groups.end).await, "every object counted out");
|
|
let f = ok(&sock, json!({"op": "flush"})).await;
|
|
assert_eq!(f["window"], window);
|
|
assert_eq!(f["cut"][&media], groups.end, "the cut: the group after the last that began here: {f}");
|
|
}
|
|
assert_eq!(verify(&got, &s, &pk).0, 5, "every group delivered");
|
|
|
|
// The operator closes window 0 from the surface alone: its parts, and the room's record.
|
|
let w0 = statement(&sock, 0, &r).await;
|
|
let rec0 = record(&sock, &r, t0, 0).await;
|
|
w0.check_record(&rec0, t0).expect("the record recomputes from the window's parts");
|
|
assert_eq!(w0.statement.relay_key, hex::encode(server.relay().relay_id()));
|
|
// What the lifecycle is written from: who published which media, who watched which layer.
|
|
let publisher = w0.statement.parts.iter().find(|x| x.tally.publisher.objects_received > 0).expect("the publisher's part");
|
|
assert_eq!(publisher.tally.publisher.tracks[&media].objects_received, 3, "its media in the window");
|
|
let viewer = w0.statement.parts.iter().find(|x| !x.tally.subscriber.layer_spans.is_empty()).expect("the viewer's part");
|
|
assert_eq!(viewer.tally.subscriber.tracks[&media].objects_forwarded, 3);
|
|
assert!(viewer.tally.subscriber.layer_spans.iter().all(|x| x.class == 0 && x.to_unix_us <= w0.statement.to_unix_us), "its layer and its time in the window");
|
|
assert_eq!(w0.statement.parts.len(), 2, "both open, nobody left");
|
|
|
|
// Planted, each refused.
|
|
let key = SigningKey::from_bytes(&seed);
|
|
let mut caught = 0;
|
|
let mut raised = w0.statement.clone();
|
|
let vi = raised.parts.iter().position(|x| !x.tally.subscriber.layer_spans.is_empty()).unwrap();
|
|
raised.parts[vi].tally.subscriber.tracks.get_mut(&media).unwrap().objects_forwarded += 1;
|
|
let e = SignedWindow::sign(raised, &key).check_record(&rec0, t0).expect_err("a count raised, signed by the relay's own key");
|
|
assert!(e.starts_with("parts_mismatch"), "{e}");
|
|
caught += 1;
|
|
let mut fewer = w0.statement.clone();
|
|
fewer.parts.remove(vi);
|
|
let e = SignedWindow::sign(fewer, &key).check_record(&rec0, t0).expect_err("the viewer's part left out, signed so");
|
|
assert!(e.starts_with("parts_mismatch"), "{e}");
|
|
caught += 1;
|
|
let mut changed = w0.clone();
|
|
changed.statement.parts[vi].tally.subscriber.layer_spans[0].to_unix_us -= 1;
|
|
let e = changed.check_record(&rec0, t0).expect_err("a part changed after it was signed");
|
|
assert!(e.starts_with("bad_signature"), "{e}");
|
|
caught += 1;
|
|
let w1 = statement(&sock, 1, &r).await;
|
|
let rec1 = record(&sock, &r, t0, 1).await;
|
|
w1.check_record(&rec1, t0).expect("window 1's record recomputes from its parts");
|
|
let e = w1.check_record(&rec0, t0).expect_err("window 1's parts against window 0's record");
|
|
assert!(e.starts_with("parts_mismatch"), "{e}");
|
|
caught += 1;
|
|
eprintln!("planted faults caught (parts): {caught} of 4");
|
|
assert_eq!(caught, 4);
|
|
|
|
// Unsealed, and not kept: refused.
|
|
let a = ask(&sock, json!({"op": "window", "window": 9})).await;
|
|
assert_eq!(a["ok"], false, "{a}");
|
|
p.close(0, "done");
|
|
v.close(0, "done");
|
|
server.shutdown().await;
|
|
let _ = std::fs::remove_dir_all(&d);
|
|
}
|
|
|
|
// ── Relays that carry a room close a window at one cut ─────────────────────
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn relays_that_carry_a_room_close_a_window_at_one_cut() {
|
|
timeout(Duration::from_secs(240), cuts_run()).await.expect("the test finishes in time");
|
|
}
|
|
|
|
async fn cuts_run() {
|
|
let room = [0x71; 32];
|
|
let r = hex::encode(room);
|
|
let pk = SigningKey::from_bytes(&[0x72; 32]);
|
|
let (sc, s1, s2) = (socket_path("core"), socket_path("edge1"), socket_path("edge2"));
|
|
let quiet = |c: &mut joule_moq_relay::RelayConfig| {
|
|
c.metering.energy = joule_moq_relay::config::EnergyMode::Off;
|
|
c.control.flush_grace_ms = 300;
|
|
c.limits.upstream_reconnect_ms = 100;
|
|
c.limits.goaway_timeout_ms = 1_000;
|
|
};
|
|
let core = relay(|c| {
|
|
quiet(c);
|
|
c.control.socket = Some(sc.clone());
|
|
})
|
|
.await;
|
|
let edge = |sock: PathBuf| {
|
|
let up = upstream_to(&core, &[&[&r]]);
|
|
relay(move |c| {
|
|
quiet(c);
|
|
c.control.socket = Some(sock);
|
|
c.upstream = vec![up];
|
|
})
|
|
};
|
|
let (e1, e2) = (edge(s1.clone()).await, edge(s2.clone()).await);
|
|
for e in [&e1, &e2] {
|
|
assert!(until(STEP, || e.relay().links().iter().any(|l| l.session().is_some())).await, "the edge's link to the core opens");
|
|
}
|
|
let t = Tracks::new(ns(&[&r, &hexkey(&pk), "video"]), "0", 10, 11, 64);
|
|
let media = joule_moq::draft22::names::render(&t.ns, Some(&t.media));
|
|
let (p, _ep) = connect(&core, Via::WebSocket, ClientOptions::default()).await;
|
|
let _published = publish_tracks(&p, &t, None).await;
|
|
let (v1, _) = connect(&e1, Via::WebSocket, ClientOptions::default()).await;
|
|
let (v2, _) = connect(&e2, Via::WebSocket, ClientOptions::default()).await;
|
|
let (sub1, sub2) = (subscribe_both(&v1, &t, Parameters::new()).await, subscribe_both(&v2, &t, Parameters::new()).await);
|
|
let (mut got1, mut got2) = (Vec::new(), Vec::new());
|
|
let sessions = |s: &RelayServer| s.relay().stats().sessions_open;
|
|
// Groups `gs` published, received at both viewers, and counted out at every relay.
|
|
let groups = |gs: std::ops::Range<u64>| {
|
|
let (p, pk, t) = (&p, &pk, &t);
|
|
async move {
|
|
for g in gs {
|
|
publish_group(p, pk, t, g, &payload("c", g, 400)).await;
|
|
}
|
|
}
|
|
};
|
|
let all = 2u64; // a media and a receipts object a group
|
|
let received = async |v: &Client, got: &mut Vec<Got>, n: u64| {
|
|
let more = collect(v, STEP, |x| (got.len() + x.len()) as u64 >= all * n).await;
|
|
got.extend(more);
|
|
};
|
|
let counted = async |n: u64| {
|
|
// At each edge its viewer's; at the core its two links'.
|
|
assert!(until(STEP, || forwarded(&e1) == all * n && forwarded(&e2) == all * n && forwarded(&core) == 2 * all * n).await, "every object counted out at every relay");
|
|
};
|
|
groups(0..4).await;
|
|
received(&v1, &mut got1, 4).await;
|
|
received(&v2, &mut got2, 4).await;
|
|
counted(4).await;
|
|
let before = (sessions(&core), sessions(&e1), sessions(&e2));
|
|
|
|
// The edges close first; group 4 begins after their instants and before the core's.
|
|
let c1 = ok(&s1, json!({"op": "close"})).await;
|
|
let c2 = ok(&s2, json!({"op": "close"})).await;
|
|
groups(4..5).await;
|
|
received(&v1, &mut got1, 5).await;
|
|
received(&v2, &mut got2, 5).await;
|
|
let cc = ok(&sc, json!({"op": "close"})).await;
|
|
let cut_of = |a: &Value| serde_json::from_value::<Cut>(a["cut"].clone()).expect("a cut");
|
|
let (own1, own2, own_core) = (cut_of(&c1), cut_of(&c2), cut_of(&cc));
|
|
assert_eq!((own1.of(&media), own2.of(&media), own_core.of(&media)), (Some(4), Some(4), Some(5)), "each relay's own cut: the group after the last that began there");
|
|
let joined = Cut::join([&own_core, &own1, &own2]);
|
|
assert_eq!(joined.of(&media), Some(5), "the join: group 4 in the closing window, everywhere");
|
|
// Group 5 begins after every instant: it waits for the cut at all three.
|
|
groups(5..6).await;
|
|
received(&v1, &mut got1, 6).await;
|
|
received(&v2, &mut got2, 6).await;
|
|
counted(6).await;
|
|
|
|
let mut caught = 0;
|
|
// Planted: a cut below the core's own, refused with nothing sealed.
|
|
let early = ask(&sc, json!({"op": "seal", "window": cc["window"], "cut": own1})).await;
|
|
assert_eq!(early["ok"], false, "{early}");
|
|
assert!(early["error"].as_str().unwrap().starts_with("cut_too_early"), "{early}");
|
|
assert!(core.relay().window(cc["window"].as_u64().unwrap()).is_none(), "nothing sealed");
|
|
caught += 1;
|
|
|
|
// The core and the first edge seal at the join; the second edge, planted, at its own cut,
|
|
// where an edge cutting at its own instant would.
|
|
let seal = |path: PathBuf, w: Value, cut: Cut| async move { ok(&path, json!({"op": "seal", "window": w, "cut": cut})).await };
|
|
let (a, b, c) = tokio::join!(seal(sc.clone(), cc["window"].clone(), joined.clone()), seal(s1.clone(), c1["window"].clone(), joined.clone()), seal(s2.clone(), c2["window"].clone(), own2.clone()));
|
|
assert_eq!(a["cut_sha256"], b["cut_sha256"], "the core and the first edge sealed at one cut");
|
|
assert_ne!(a["cut_sha256"], c["cut_sha256"]);
|
|
let window = |a: &Value| a["window"].as_u64().unwrap();
|
|
let (wc, w1, w2) = (window(&a), window(&b), window(&c));
|
|
let (stc, st1, st2) = (statement(&sc, wc, &r).await, statement(&s1, w1, &r).await, statement(&s2, w2, &r).await);
|
|
let flows = link_chain(&stc, &st1, true).expect("the core and the first edge chain over their link in the window");
|
|
assert_eq!((flows[&media].out, flows[&media].inn), (5, 5), "groups 0 to 4, out at the core and in at the edge");
|
|
let e = link_chain(&stc, &st2, true).expect_err("the second edge cut one group early");
|
|
assert!(e.starts_with("hop_chain_mismatch"), "{e}");
|
|
caught += 1;
|
|
|
|
// The next window, every relay at one cut.
|
|
groups(6..8).await;
|
|
received(&v1, &mut got1, 8).await;
|
|
received(&v2, &mut got2, 8).await;
|
|
counted(8).await;
|
|
let closes = [ok(&s1, json!({"op": "close"})).await, ok(&s2, json!({"op": "close"})).await, ok(&sc, json!({"op": "close"})).await];
|
|
let cuts: Vec<Cut> = closes.iter().map(cut_of).collect();
|
|
let joined = Cut::join(&cuts);
|
|
assert_eq!(joined.of(&media), Some(8));
|
|
let (a, b, c) = tokio::join!(seal(s1.clone(), closes[0]["window"].clone(), joined.clone()), seal(s2.clone(), closes[1]["window"].clone(), joined.clone()), seal(sc.clone(), closes[2]["window"].clone(), joined.clone()));
|
|
let (st1, st2, stc) = (statement(&s1, window(&a), &r).await, statement(&s2, window(&b), &r).await, statement(&sc, window(&c), &r).await);
|
|
let flows = link_chain(&stc, &st1, true).expect("and in the next");
|
|
assert_eq!((flows[&media].out, flows[&media].inn), (3, 3), "groups 5 to 7");
|
|
let e = link_chain(&stc, &st2, false).expect_err("the group the second edge cut early, counted in this window");
|
|
assert!(e.starts_with("hop_chain_mismatch") && e.contains("4 objects in over the link, 3 sent"), "{e}");
|
|
caught += 1;
|
|
|
|
// The first edge rode through both window ends: its viewer was sent nothing, is open, and
|
|
// has every group once, verified; no session anywhere ended.
|
|
rode_through(&v1).expect("the first edge's viewer rode through");
|
|
assert_eq!(verify(&got1, &sub1, &pk).0, 8, "every group once, at the first edge");
|
|
assert_eq!(verify(&got2, &sub2, &pk).0, 8, "and at the second");
|
|
assert_eq!((sessions(&core), sessions(&e1), sessions(&e2)), before, "nobody left, at the core or at an edge");
|
|
assert!(!e1.relay().going_away());
|
|
|
|
// Planted: the second edge ends its window as edges did, by a restart: its viewer is sent
|
|
// GOAWAY, and the check catches it.
|
|
let stopping = tokio::spawn(e2.shutdown());
|
|
assert!(until(STEP, || v2.goaway().borrow().is_some()).await, "the restarting edge sends GOAWAY");
|
|
let e = rode_through(&v2).expect_err("a GOAWAY at a window's end");
|
|
assert!(e.contains("GOAWAY"), "{e}");
|
|
caught += 1;
|
|
rode_through(&v1).expect("the first edge's viewer still");
|
|
eprintln!("planted faults caught (cuts and ride-through): {caught} of 4");
|
|
assert_eq!(caught, 4);
|
|
|
|
for c in [&p, &v1, &v2] {
|
|
c.close(0, "done");
|
|
}
|
|
let _ = stopping.await;
|
|
e1.shutdown().await;
|
|
core.shutdown().await;
|
|
drop(sub1);
|
|
drop(sub2);
|
|
}
|
|
|
|
// ── Summaries drained ───────────────────────────────────────────────────────
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn summaries_are_drained_and_none_past_the_bound_is_dropped() {
|
|
timeout(Duration::from_secs(180), drain_run()).await.expect("the test finishes in time");
|
|
}
|
|
|
|
/// A session opened over the WebSocket binding and closed, waited for until its summary is kept.
|
|
async fn one_session(server: &RelayServer, kept: usize) {
|
|
let (c, _) = connect(server, Via::WebSocket, ClientOptions::default()).await;
|
|
c.close(0, "done");
|
|
assert!(until(STEP, || server.relay().summaries_held().0 >= kept).await, "the session's summary kept");
|
|
}
|
|
|
|
/// The sessions of the summaries a drain answered, in order.
|
|
fn drained(a: &Value) -> Vec<(u64, u64)> {
|
|
a["summaries"].as_array().unwrap().iter().map(|x| (x["seq"].as_u64().unwrap(), x["summary"]["summary"]["session"].as_u64().unwrap())).collect()
|
|
}
|
|
|
|
async fn drain_run() {
|
|
let mut caught = 0;
|
|
// No spill file: a bound of 2.
|
|
let sock = socket_path("drain");
|
|
let server = relay(|c| {
|
|
c.control.socket = Some(sock.clone());
|
|
c.summary.held_max = 2;
|
|
c.metering.energy = joule_moq_relay::config::EnergyMode::Off;
|
|
})
|
|
.await;
|
|
for n in 1..=3 {
|
|
one_session(&server, n).await;
|
|
}
|
|
// Past the bound: every summary held, none dropped, and a new session refused.
|
|
let held = server.relay().clone();
|
|
assert_eq!(held.summaries().len(), 3, "the third, past the bound, held");
|
|
assert!(held.refusing_for_summaries());
|
|
// Refused before the relay's SETUP is sent: what is held is the close and its reason.
|
|
let ep = quic::client_endpoint(loopback(), vec![server.identity.chain[0].clone()], vec![joule_moq::draft22::ALPN.to_vec()], None).unwrap();
|
|
let conn: joule_moq_relay::transport::SharedConnection = std::sync::Arc::new(quic::connect(&ep, server.quic_addr.unwrap(), "localhost").await.expect("a QUIC session"));
|
|
let client = Client::connect(conn.clone(), ClientOptions::default()).await;
|
|
match timeout(STEP, conn.closed()).await {
|
|
Ok(TransportError::Closed { code, reason, .. }) => {
|
|
assert_eq!(code, SessionError::INTERNAL_ERROR.0, "{reason}");
|
|
assert!(reason.starts_with(joule_moq_relay::outbox::SUMMARIES_UNACKNOWLEDGED), "{reason}");
|
|
}
|
|
other => panic!("a session past the bound was admitted: {other:?} (connect: {:?})", client.as_ref().map(|_| "Ok")),
|
|
}
|
|
drop(client);
|
|
caught += 1;
|
|
// Planted: an acknowledgement of what was never handed out frees nothing.
|
|
let a = ask(&sock, json!({"op": "ack", "through": 3})).await;
|
|
assert_eq!(a["ok"], false, "{a}");
|
|
assert_eq!(held.summaries_held().0, 3);
|
|
caught += 1;
|
|
let first = ok(&sock, json!({"op": "summaries", "limit": 2})).await;
|
|
let seen = drained(&first);
|
|
assert_eq!(seen.iter().map(|x| x.0).collect::<Vec<_>>(), vec![1, 2]);
|
|
let a = ask(&sock, json!({"op": "ack", "through": 3})).await;
|
|
assert_eq!(a["ok"], false, "3 was not handed out: {a}");
|
|
caught += 1;
|
|
let a = ok(&sock, json!({"op": "ack", "through": 1})).await;
|
|
assert_eq!((a["freed"].as_u64(), a["held"].as_u64()), (Some(1), Some(2)));
|
|
// The acknowledged summary is freed, and never handed out again.
|
|
let rest = drained(&ok(&sock, json!({"op": "summaries"})).await);
|
|
assert_eq!(rest.iter().map(|x| x.0).collect::<Vec<_>>(), vec![2, 3], "1 is gone");
|
|
assert!(!held.summaries().iter().any(|s| s.summary.session == seen[0].1), "freed");
|
|
assert!(!held.refusing_for_summaries(), "back at the bound: sessions admitted");
|
|
one_session(&server, 3).await;
|
|
let a = ok(&sock, json!({"op": "summaries", "after": 3})).await;
|
|
assert_eq!(drained(&a).iter().map(|x| x.0).collect::<Vec<_>>(), vec![4], "the session admitted since");
|
|
let a = ok(&sock, json!({"op": "ack", "through": 4})).await;
|
|
assert_eq!(a["held"].as_u64(), Some(0));
|
|
server.shutdown().await;
|
|
|
|
// A spill file: past a bound of 1, the summaries go there and come back in order.
|
|
let d = dir("spill");
|
|
let spill = d.join("summaries.jsonl");
|
|
let sock = socket_path("spill");
|
|
let start = |sp: PathBuf, so: PathBuf| {
|
|
relay(move |c| {
|
|
c.control.socket = Some(so);
|
|
c.summary.held_max = 1;
|
|
c.summary.spill_path = Some(sp);
|
|
c.metering.energy = joule_moq_relay::config::EnergyMode::Off;
|
|
})
|
|
};
|
|
let server = start(spill.clone(), sock.clone()).await;
|
|
for n in 1..=3 {
|
|
one_session(&server, n).await;
|
|
}
|
|
assert_eq!(server.relay().summaries_held(), (3, 2, 0, 0), "one in memory, two spilled");
|
|
assert!(!server.relay().refusing_for_summaries(), "spilled, so every session admitted");
|
|
assert_eq!(std::fs::read_to_string(&spill).unwrap().lines().count(), 2);
|
|
let all = drained(&ok(&sock, json!({"op": "summaries"})).await);
|
|
assert_eq!(all.iter().map(|x| x.0).collect::<Vec<_>>(), vec![1, 2, 3], "read back in order");
|
|
let ids: Vec<u64> = all.iter().map(|x| x.1).collect();
|
|
// Two acknowledged, the one in memory and the first spilled; then the relay restarts.
|
|
ok(&sock, json!({"op": "ack", "through": 2})).await;
|
|
server.shutdown().await;
|
|
let server = start(spill.clone(), sock.clone()).await;
|
|
let back = drained(&ok(&sock, json!({"op": "summaries"})).await);
|
|
assert_eq!(back.iter().map(|x| x.1).collect::<Vec<_>>(), vec![ids[2]], "a restart hands over the one not acknowledged, and none that was");
|
|
ok(&sock, json!({"op": "ack", "through": back[0].0})).await;
|
|
assert_eq!(std::fs::metadata(&spill).unwrap().len(), 0, "every spilled summary acknowledged: the file emptied");
|
|
server.shutdown().await;
|
|
eprintln!("planted faults caught (summaries): {caught} of 3");
|
|
assert_eq!(caught, 3);
|
|
let _ = std::fs::remove_dir_all(&d);
|
|
}
|
|
|
|
// ── Summaries handed over on a file ─────────────────────────────────────────────────────────
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn summaries_handed_over_on_out_path_never_refuse_a_session() {
|
|
timeout(Duration::from_secs(180), handed_over_run()).await.expect("the test finishes in time");
|
|
}
|
|
|
|
/// A session opened over the WebSocket binding and closed, waited for until the relay has its
|
|
/// summary, kept or handed over.
|
|
async fn ended(server: &RelayServer, n: u64) {
|
|
let (c, _) = connect(server, Via::WebSocket, ClientOptions::default()).await;
|
|
c.close(0, "done");
|
|
let r = server.relay().clone();
|
|
assert!(until(STEP, || r.summaries_handed_over().0.unwrap_or(0) + r.summaries_held().0 as u64 >= n).await, "the session's summary handed over or kept");
|
|
}
|
|
|
|
/// A session past the bound: refused before the relay's SETUP, `summaries_unacknowledged`; or
|
|
/// admitted. `true` where it was refused so.
|
|
async fn refused_for_summaries(server: &RelayServer) -> bool {
|
|
let ep = quic::client_endpoint(loopback(), vec![server.identity.chain[0].clone()], vec![joule_moq::draft22::ALPN.to_vec()], None).unwrap();
|
|
let conn: joule_moq_relay::transport::SharedConnection = std::sync::Arc::new(quic::connect(&ep, server.quic_addr.unwrap(), "localhost").await.expect("a QUIC session"));
|
|
let client = Client::connect(conn.clone(), ClientOptions::default()).await;
|
|
let refused = match &client {
|
|
Err(_) => matches!(timeout(STEP, conn.closed()).await, Ok(TransportError::Closed { code, reason, .. }) if code == SessionError::INTERNAL_ERROR.0 && reason.starts_with(joule_moq_relay::outbox::SUMMARIES_UNACKNOWLEDGED)),
|
|
Ok(c) => {
|
|
c.close(0, "admitted");
|
|
false
|
|
}
|
|
};
|
|
drop(client);
|
|
refused
|
|
}
|
|
|
|
async fn handed_over_run() {
|
|
let mut caught = 0;
|
|
// The live room's relay: an out_path, no surface. A bound of 2, and five sessions.
|
|
let d = dir("handed");
|
|
let out = d.join("sessions.jsonl");
|
|
let o = out.clone();
|
|
let server = relay(move |c| {
|
|
c.summary.out_path = Some(o);
|
|
c.summary.held_max = 2;
|
|
c.metering.energy = joule_moq_relay::config::EnergyMode::Off;
|
|
})
|
|
.await;
|
|
assert!(server.relay().config().summaries_handed_over());
|
|
for n in 1..=5 {
|
|
ended(&server, n).await;
|
|
assert!(!server.relay().refusing_for_summaries(), "session {n}: none refused");
|
|
}
|
|
assert!(!refused_for_summaries(&server).await, "a sixth session, past the bound, admitted");
|
|
assert!(until(STEP, || server.relay().summaries_handed_over().0 == Some(6)).await);
|
|
assert_eq!(server.relay().summaries_held().0, 0, "none held");
|
|
assert_eq!(server.relay().summaries_handed_over().1, 0, "nothing to tell: nothing held");
|
|
let lines: Vec<String> = std::fs::read_to_string(&out).unwrap().lines().map(str::to_string).collect();
|
|
assert_eq!(lines.len(), 6, "every summary on the file");
|
|
let mut sessions = std::collections::BTreeSet::new();
|
|
for l in &lines {
|
|
let s: joule_moq_relay::SignedSessionSummary = serde_json::from_str(l).expect("a whole summary a line");
|
|
s.verify().expect("signed by the relay");
|
|
sessions.insert(s.summary.session);
|
|
}
|
|
assert_eq!(sessions.len(), 6, "one line for each session");
|
|
server.shutdown().await;
|
|
|
|
// Planted: a surface the operator never acknowledges on. Told at half the bound and past
|
|
// it, and the session past the bound refused.
|
|
let sock = socket_path("handed");
|
|
let (o, so) = (d.join("surfaced.jsonl"), sock.clone());
|
|
let server = relay(move |c| {
|
|
c.summary.out_path = Some(o);
|
|
c.control.socket = Some(so);
|
|
c.summary.held_max = 2;
|
|
c.metering.energy = joule_moq_relay::config::EnergyMode::Off;
|
|
})
|
|
.await;
|
|
assert!(!server.relay().config().summaries_handed_over());
|
|
for n in 1..=3 {
|
|
ended(&server, n).await;
|
|
}
|
|
assert_eq!(server.relay().summaries_handed_over(), (None, 2), "held, and the operator told at half the bound and past it");
|
|
if refused_for_summaries(&server).await {
|
|
caught += 1;
|
|
}
|
|
server.shutdown().await;
|
|
|
|
// Planted: a file that cannot be written. Held, and the session past the bound refused.
|
|
let o = d.join("no-such-dir").join("sessions.jsonl");
|
|
let server = relay(move |c| {
|
|
c.summary.out_path = Some(o);
|
|
c.summary.held_max = 2;
|
|
c.metering.energy = joule_moq_relay::config::EnergyMode::Off;
|
|
})
|
|
.await;
|
|
for n in 1..=3 {
|
|
ended(&server, n).await;
|
|
}
|
|
assert_eq!((server.relay().summaries_held().0, server.relay().summaries_handed_over().0), (3, Some(0)), "none handed over, every one held");
|
|
if refused_for_summaries(&server).await {
|
|
caught += 1;
|
|
}
|
|
server.shutdown().await;
|
|
eprintln!("planted faults caught (summaries handed over): {caught} of 2");
|
|
assert_eq!(caught, 2);
|
|
let _ = std::fs::remove_dir_all(&d);
|
|
let _ = std::fs::remove_file(&sock);
|
|
}
|