442 lines
24 KiB
Rust
442 lines
24 KiB
Rust
//! The floor of a WAI room (WAI co-performance §6), through a relay on loopback that holds
|
|
//! it: the room's clock publishes a turn schedule that refuses on its events track, and two
|
|
//! performers each publish an audio track that one subscriber follows. The object of the
|
|
//! performer whose turn it is reaches the subscriber, and the other's does not; once the first
|
|
//! hands its turn to the second on its own events track, the second's reaches the subscriber
|
|
//! and the first's does not. Each refusal is counted.
|
|
//!
|
|
//! And the room's audience inputs (§7.3), through a relay that honours a room authority's
|
|
//! grants: the clock opens a vote on its events track; a member whose grant carries the
|
|
//! audience-input power sends one vote that a subscriber of its events track receives, and a
|
|
//! second, past its power, that the relay withholds; a viewer whose grant carries no such power
|
|
//! has its vote withheld; each refusal is counted in its session's summary by reason.
|
|
//!
|
|
//! And the order the relay counts them in (§7.3), its author's chain's: a member's second vote,
|
|
//! sent first on a stream of its own, waits for its first, which is forwarded, and is then past
|
|
//! the member's power; a second vote whose first never comes is refused once the relay's hold
|
|
//! runs out, and the first, sent after, is forwarded.
|
|
|
|
mod common;
|
|
|
|
use common::*;
|
|
use ed25519_dalek::{Signer, SigningKey};
|
|
use joule_moq::draft22::data::{SubgroupHeader, SubgroupId, SubgroupObject};
|
|
use joule_moq::draft22::Parameters;
|
|
use joule_moq::TrackNamespace;
|
|
use joule_moq_relay::client::{Client, ClientOptions};
|
|
use joule_moq_relay::config::{Authority, FloorConfig, Grant};
|
|
use joule_moq_relay::floor::Place;
|
|
use joule_moq_relay::wire::{ReqRx, ReqTx};
|
|
use joule_moq_relay::RelayServer;
|
|
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
|
|
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 `meta` record of live-events §3.1, without an anchor, signed by `key`.
|
|
fn meta(key: &SigningKey, room: [u8; 32], seq: u64, 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(&[0; 32]);
|
|
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 `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
|
|
}
|
|
|
|
/// A `WPH1` handover (co-performance §6.3).
|
|
fn handover(at: u64, to: &SigningKey) -> Vec<u8> {
|
|
let mut b = b"WPH1\x01".to_vec();
|
|
b.extend_from_slice(&at.to_le_bytes());
|
|
b.extend_from_slice(&to.verifying_key().to_bytes());
|
|
b
|
|
}
|
|
|
|
/// 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 track(room: [u8; 32], who: &SigningKey, field: &str) -> TrackNamespace {
|
|
ns(&[&hex::encode(room), &hexkey(who), field])
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn the_relay_holds_the_floor() {
|
|
timeout(Duration::from_secs(120), run()).await.expect("the test finishes in time");
|
|
}
|
|
|
|
async fn run() {
|
|
let room = [0x61; 32];
|
|
let clock = SigningKey::from_bytes(&[0x62; 32]);
|
|
let (a, b) = (SigningKey::from_bytes(&[0x63; 32]), SigningKey::from_bytes(&[0x64; 32]));
|
|
let server = relay(|c| c.floor = Some(FloorConfig { room_keys: vec![hexkey(&clock)], ..FloorConfig::default() })).await;
|
|
let relay = server.relay().clone();
|
|
let floor = || relay.floor().expect("the relay holds the floor");
|
|
// The room began ten seconds ago; its clock gives the floor to a, then b, ten minutes each.
|
|
let t0 = now_us() - 10_000_000;
|
|
let (c, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let _ce = c.publish(track(room, &clock, "events"), name("0"), 1, Parameters::new(), track_props(16, None)).await.expect("PUBLISH the clock's events");
|
|
let rc = now_us() - t0;
|
|
send(&c, 1, 0, &msf(&meta(&clock, room, 0, rc, &schedule(t0, rc, 600_000_000, &[&a, &b])))).await;
|
|
let place = |who: &SigningKey| floor().place(&track(room, who, "audio"), now_us());
|
|
assert!(until(STEP, || place(&a) == Some(Place::InTurn)).await, "the relay reads the schedule");
|
|
assert_eq!(place(&b), Some(Place::OutOfTurn(1)));
|
|
|
|
// Both performers publish audio; one subscriber follows both.
|
|
let (pa, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let (pb, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let _ha = pa.publish(track(room, &a, "audio"), name("0"), 10, Parameters::new(), track_props(0, None)).await.expect("PUBLISH a's audio");
|
|
let _hb = pb.publish(track(room, &b, "audio"), name("0"), 20, Parameters::new(), track_props(0, None)).await.expect("PUBLISH b's audio");
|
|
let (s, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let _sa = s.subscribe(track(room, &a, "audio"), name("0"), Parameters::new()).await.expect("SUBSCRIBE a's audio");
|
|
let _sb = s.subscribe(track(room, &b, "audio"), name("0"), Parameters::new()).await.expect("SUBSCRIBE b's audio");
|
|
send(&pa, 10, 1, b"a in its turn").await;
|
|
send(&pb, 20, 1, b"b out of turn").await;
|
|
// What reaches the subscriber: the object in turn, waited for; then, once the relay has held
|
|
// the other back, anything more for a second.
|
|
let has = |g: &[Got], p: &[u8]| g.iter().any(|x| x.objects.iter().any(|o| o.payload == p));
|
|
let mut got = collect(&s, STEP, |g| has(g, b"a in its turn")).await;
|
|
assert!(until(STEP, || floor().refused() >= 1).await, "the relay holds back the object out of turn");
|
|
got.extend(collect(&s, Duration::from_secs(1), |_| false).await);
|
|
let payloads: Vec<&[u8]> = got.iter().flat_map(|g| g.objects.iter().map(|o| o.payload.as_slice())).collect();
|
|
assert_eq!(payloads, vec![b"a in its turn".as_slice()], "only the object in turn is forwarded");
|
|
assert_eq!(floor().refused(), 1);
|
|
|
|
// a hands its turn to b, from now.
|
|
let _ae = pa.publish(track(room, &a, "events"), name("0"), 11, Parameters::new(), track_props(16, None)).await.expect("PUBLISH a's events");
|
|
let at = now_us() - t0;
|
|
send(&pa, 11, 0, &msf(&meta(&a, room, 0, at, &handover(at, &b)))).await;
|
|
assert!(until(STEP, || place(&b) == Some(Place::InTurn)).await, "the relay reads the handover");
|
|
send(&pa, 10, 2, b"a after its handover").await;
|
|
send(&pb, 20, 2, b"b in the turn it was handed").await;
|
|
let mut got = collect(&s, STEP, |g| has(g, b"b in the turn it was handed")).await;
|
|
assert!(until(STEP, || floor().refused() >= 2).await, "the relay holds back a's object after its handover");
|
|
got.extend(collect(&s, Duration::from_secs(1), |_| false).await);
|
|
let payloads: Vec<&[u8]> = got.iter().flat_map(|g| g.objects.iter().map(|o| o.payload.as_slice())).collect();
|
|
assert_eq!(payloads, vec![b"b in the turn it was handed".as_slice()]);
|
|
assert_eq!(floor().refused(), 2);
|
|
for x in [&c, &pa, &pb, &s] {
|
|
x.close(0, "done");
|
|
}
|
|
server.shutdown().await;
|
|
}
|
|
|
|
/// A grant of `room` for `principal`, as live-rooms §4.1 lays it out, signed by `authority`:
|
|
/// a viewer's (no media, its policy every kind but the lifecycle's at the default bounds),
|
|
/// 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
|
|
}
|
|
|
|
/// A `meta` record with a `prev`, so a chain of several reads as one.
|
|
fn meta_after(key: &SigningKey, room: [u8; 32], seq: u64, prev: [u8; 32], t: u64, payload: &[u8]) -> Vec<u8> {
|
|
let mut r = meta(key, room, seq, t, payload);
|
|
r[77..109].copy_from_slice(&prev);
|
|
let body = r.len() - 64;
|
|
let mut m = b"wai:live-event\x01".to_vec();
|
|
m.extend_from_slice(&r[..body]);
|
|
let sig = key.sign(&m).to_bytes();
|
|
r[body..].copy_from_slice(&sig);
|
|
r
|
|
}
|
|
|
|
fn sha256(b: &[u8]) -> [u8; 32] {
|
|
ring::digest::digest(&ring::digest::SHA256, b).as_ref().try_into().unwrap()
|
|
}
|
|
|
|
/// A record's digest (live-events §3.2): over all of it but its signature.
|
|
fn digest(record: &[u8]) -> [u8; 32] {
|
|
sha256(&record[..record.len() - 64])
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn the_relay_refuses_invalid_audience_inputs() {
|
|
timeout(Duration::from_secs(120), inputs()).await.expect("the test finishes in time");
|
|
}
|
|
|
|
async fn inputs() {
|
|
let room = [0x71; 32];
|
|
let authority = SigningKey::from_bytes(&[0x72; 32]);
|
|
let clock = SigningKey::from_bytes(&[0x73; 32]);
|
|
let (member, viewer, watcher) = (SigningKey::from_bytes(&[0x74; 32]), SigningKey::from_bytes(&[0x75; 32]), SigningKey::from_bytes(&[0x76; 32]));
|
|
let token = b"the clock's token".to_vec();
|
|
let token_sha = hex::encode(sha256(&token));
|
|
let server = relay(|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: token_sha.clone(), publish: true, subscribe: true, proof: false, unlimited: true, seek: false }];
|
|
c.floor = Some(FloorConfig { room_keys: vec![hexkey(&clock)], ..FloorConfig::default() });
|
|
})
|
|
.await;
|
|
let target = format!("//{}{}", server.websocket_addr.unwrap(), server.relay().config().path);
|
|
let with = |g: Vec<u8>, k: &SigningKey| ClientOptions { grants: vec![(g, k.clone())], target: target.clone(), ..ClientOptions::default() };
|
|
let t0 = now_us() - 10_000_000;
|
|
let rc = || now_us() - t0;
|
|
|
|
// The clock opens a vote of three options, closing in a minute.
|
|
let (c, _) = connect(&server, Via::WebSocket, ClientOptions { tokens: vec![joule_moq::draft22::Token::UseValue { token_type: 0, value: token.clone() }], ..ClientOptions::default() }).await;
|
|
let _ce = c.publish(track(room, &clock, "events"), name("0"), 1, Parameters::new(), track_props(16, None)).await.expect("PUBLISH the clock's events");
|
|
let mut w = b"WPW1\x01".to_vec();
|
|
w.extend_from_slice(&[1, 0b100, 3, 1]);
|
|
w.extend_from_slice(&(rc() + 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, rc(), &w);
|
|
let wd = digest(&wr);
|
|
send(&c, 1, 0, &msf(&wr)).await;
|
|
let vote = |o: u8| {
|
|
let mut b = b"WPA1\x01".to_vec();
|
|
b.extend_from_slice(&wd);
|
|
b.extend_from_slice(&[3, o]);
|
|
b
|
|
};
|
|
|
|
// A watcher follows the member's and the viewer's events tracks.
|
|
let (s, _) = connect(&server, Via::WebSocket, with(viewer_grant(&authority, room, &watcher, &[]), &watcher)).await;
|
|
// The member may send one input to a window; the viewer none.
|
|
let (pm, _) = connect(&server, Via::WebSocket, with(viewer_grant(&authority, room, &member, &[("audience_input", "1")]), &member)).await;
|
|
let (pv, _) = connect(&server, Via::WebSocket, with(viewer_grant(&authority, room, &viewer, &[]), &viewer)).await;
|
|
let _pme = pm.publish(track(room, &member, "events"), name("0"), 2, Parameters::new(), track_props(16, None)).await.expect("PUBLISH the member's events");
|
|
let _pve = pv.publish(track(room, &viewer, "events"), name("0"), 3, Parameters::new(), track_props(16, None)).await.expect("PUBLISH the viewer's events");
|
|
let _sm = s.subscribe(track(room, &member, "events"), name("0"), Parameters::new()).await.expect("SUBSCRIBE the member's events");
|
|
let _sv = s.subscribe(track(room, &viewer, "events"), name("0"), Parameters::new()).await.expect("SUBSCRIBE the viewer's events");
|
|
// The inputs name the window: the floor holds it before the first is sent.
|
|
let f = server.relay().floor().unwrap();
|
|
assert!(until(STEP, || f.holds_window(&room, &wd)).await, "the relay holds the clock's window");
|
|
// The member's first vote, forwarded: the watcher's subscription is live. Each group rides a
|
|
// stream of its own, which the relay reads in a task of its own, so the second is sent once
|
|
// the first has been taken (`the_relay_counts_inputs_in_chain_order` sends them the other way
|
|
// round).
|
|
let first = meta(&member, room, 0, rc(), &vote(1));
|
|
send(&pm, 2, 0, &msf(&first)).await;
|
|
let mut got = collect(&s, STEP, |g| g.iter().any(|x| x.objects.iter().any(|o| o.payload == msf(&first)))).await;
|
|
// Its second, past its power, and the viewer's, with no power: both refused, and nothing
|
|
// more reaches the watcher once the relay has refused them.
|
|
let second = meta_after(&member, room, 1, digest(&first), rc(), &vote(2));
|
|
send(&pm, 2, 1, &msf(&second)).await;
|
|
send(&pv, 3, 0, &msf(&meta(&viewer, room, 0, rc(), &vote(0)))).await;
|
|
assert!(until(STEP, || f.refused_inputs() >= 2).await, "the relay refuses the second vote and the viewer's");
|
|
got.extend(collect(&s, Duration::from_secs(1), |_| false).await);
|
|
let payloads: Vec<&[u8]> = got.iter().flat_map(|g| g.objects.iter().map(|o| o.payload.as_slice())).collect();
|
|
assert_eq!(payloads, vec![msf(&first).as_slice()], "only the member's first vote is forwarded");
|
|
assert_eq!(f.refused_inputs(), 2);
|
|
for x in [&pm, &pv, &s, &c] {
|
|
x.close(0, "done");
|
|
}
|
|
let relay = server.relay().clone();
|
|
assert!(until(STEP, || relay.summaries().len() >= 4).await, "every session's summary");
|
|
let refused: Vec<(String, u64)> = relay.summaries().iter().flat_map(|x| x.summary.tally.publisher.co_refused.clone()).collect();
|
|
let mut refused = refused;
|
|
refused.sort();
|
|
assert_eq!(refused, vec![("input_not_permitted".to_string(), 1), ("input_over_limit".to_string(), 1)]);
|
|
server.shutdown().await;
|
|
}
|
|
|
|
/// A room whose clock has opened a vote of three options, a member whose grant allows one input
|
|
/// to a window, and a watcher of the member's events track, through a relay that holds an input
|
|
/// ahead of a gap in its author's chain for `hold_ms`.
|
|
struct Audience {
|
|
server: RelayServer,
|
|
room: [u8; 32],
|
|
member: SigningKey,
|
|
window: [u8; 32],
|
|
t0: u64,
|
|
clients: [Client; 3],
|
|
_held: Vec<(ReqTx, ReqRx)>,
|
|
}
|
|
|
|
impl Audience {
|
|
async fn open(room: [u8; 32], seed: u8, hold_ms: u64) -> Audience {
|
|
let authority = SigningKey::from_bytes(&[seed; 32]);
|
|
let clock = SigningKey::from_bytes(&[seed + 1; 32]);
|
|
let (member, watcher) = (SigningKey::from_bytes(&[seed + 2; 32]), SigningKey::from_bytes(&[seed + 3; 32]));
|
|
let token = b"the clock's token".to_vec();
|
|
let token_sha = hex::encode(sha256(&token));
|
|
let server = relay(|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: token_sha, publish: true, subscribe: true, proof: false, unlimited: true, seek: false }];
|
|
c.floor = Some(FloorConfig { room_keys: vec![hexkey(&clock)], hold_ms });
|
|
})
|
|
.await;
|
|
let target = format!("//{}{}", server.websocket_addr.unwrap(), server.relay().config().path);
|
|
let with = |g: Vec<u8>, k: &SigningKey| ClientOptions { grants: vec![(g, k.clone())], target: target.clone(), ..ClientOptions::default() };
|
|
let t0 = now_us() - 10_000_000;
|
|
let (c, _) = connect(&server, Via::WebSocket, ClientOptions { tokens: vec![joule_moq::draft22::Token::UseValue { token_type: 0, value: token }], ..ClientOptions::default() }).await;
|
|
let mut held = vec![c.publish(track(room, &clock, "events"), name("0"), 1, Parameters::new(), track_props(16, None)).await.expect("PUBLISH 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, now_us() - t0, &w);
|
|
send(&c, 1, 0, &msf(&wr)).await;
|
|
let (s, _) = connect(&server, Via::WebSocket, with(viewer_grant(&authority, room, &watcher, &[]), &watcher)).await;
|
|
let (pm, _) = connect(&server, Via::WebSocket, with(viewer_grant(&authority, room, &member, &[("audience_input", "1")]), &member)).await;
|
|
held.push(pm.publish(track(room, &member, "events"), name("0"), 2, Parameters::new(), track_props(16, None)).await.expect("PUBLISH the member's events"));
|
|
let (_, tx, rx) = s.subscribe(track(room, &member, "events"), name("0"), Parameters::new()).await.expect("SUBSCRIBE the member's events");
|
|
held.push((tx, rx));
|
|
let window = digest(&wr);
|
|
let f = server.relay().floor().unwrap();
|
|
assert!(until(STEP, || f.holds_window(&room, &window)).await, "the relay holds the clock's window");
|
|
Audience { server, room, member, window, t0, clients: [c, pm, s], _held: held }
|
|
}
|
|
|
|
/// The member's vote for option `o` at seq `seq`, chained after `prev`.
|
|
fn vote(&self, seq: u64, prev: Option<&[u8]>, o: u8) -> Vec<u8> {
|
|
let mut b = b"WPA1\x01".to_vec();
|
|
b.extend_from_slice(&self.window);
|
|
b.extend_from_slice(&[3, o]);
|
|
let t = now_us() - self.t0;
|
|
match prev {
|
|
None => meta(&self.member, self.room, seq, t, &b),
|
|
Some(p) => meta_after(&self.member, self.room, seq, digest(p), t, &b),
|
|
}
|
|
}
|
|
|
|
/// What the watcher receives of the member's events: until `want` arrives, then a second more.
|
|
async fn received(&self, want: &[u8]) -> Vec<Vec<u8>> {
|
|
let s = &self.clients[2];
|
|
let mut got = collect(s, STEP, |g| g.iter().any(|x| x.objects.iter().any(|o| o.payload == want))).await;
|
|
got.extend(collect(s, Duration::from_secs(1), |_| false).await);
|
|
got.iter().flat_map(|g| g.objects.iter().map(|o| o.payload.clone())).collect()
|
|
}
|
|
|
|
/// Every session closed: the refusals its summaries count, by reason.
|
|
async fn close(self) -> Vec<(String, u64)> {
|
|
for x in &self.clients {
|
|
x.close(0, "done");
|
|
}
|
|
let relay = self.server.relay().clone();
|
|
assert!(until(STEP, || relay.summaries().len() >= 3).await, "every session's summary");
|
|
let mut refused: Vec<(String, u64)> = relay.summaries().iter().flat_map(|x| x.summary.tally.publisher.co_refused.clone()).collect();
|
|
refused.sort();
|
|
self.server.shutdown().await;
|
|
refused
|
|
}
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn the_relay_counts_inputs_in_chain_order() {
|
|
timeout(Duration::from_secs(120), chain_order()).await.expect("the test finishes in time");
|
|
}
|
|
|
|
async fn chain_order() {
|
|
// A hold past any wait of this test, however loaded the host: the relay is to wait for seq 0.
|
|
let a = Audience::open([0x81; 32], 0x82, 60_000).await;
|
|
let f = a.server.relay().floor().unwrap();
|
|
let first = a.vote(0, None, 1);
|
|
let second = a.vote(1, Some(&first), 2);
|
|
// The member's two votes, each a group on a stream of its own, the second opened first: the
|
|
// relay holds seq 1 until seq 0 arrives.
|
|
send(&a.clients[1], 2, 1, &msf(&second)).await;
|
|
let held = until(STEP, || f.waiting(&a.room) == 1).await;
|
|
assert_eq!(f.refused_inputs(), 0);
|
|
send(&a.clients[1], 2, 0, &msf(&first)).await;
|
|
// Seq 0 is counted first and forwarded; seq 1, counted after it, is past the member's power.
|
|
// (A relay that counted them as they came would have forwarded seq 1 and refused seq 0.)
|
|
let got = a.received(&msf(&first)).await;
|
|
assert!(until(STEP, || f.refused_inputs() >= 1).await, "the relay refuses one of them");
|
|
assert_eq!(got, vec![msf(&first)], "seq 0 forwarded, seq 1 withheld");
|
|
assert!(held, "the relay held seq 1 for seq 0");
|
|
assert_eq!((f.refused_inputs(), f.waiting(&a.room)), (1, 0));
|
|
assert_eq!(a.close().await, vec![("input_over_limit".to_string(), 1)]);
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn the_relay_refuses_an_input_whose_gap_never_fills() {
|
|
timeout(Duration::from_secs(120), gap_never_fills()).await.expect("the test finishes in time");
|
|
}
|
|
|
|
async fn gap_never_fills() {
|
|
let hold = Duration::from_millis(400);
|
|
let a = Audience::open([0x91; 32], 0x92, hold.as_millis() as u64).await;
|
|
let f = a.server.relay().floor().unwrap();
|
|
let first = a.vote(0, None, 1);
|
|
let second = a.vote(1, Some(&first), 2);
|
|
// Seq 1 alone, seq 0 held back by its sender: the relay refuses seq 1 once its hold runs out.
|
|
let sent = Instant::now();
|
|
send(&a.clients[1], 2, 1, &msf(&second)).await;
|
|
assert!(until(STEP, || f.refused_inputs() >= 1).await, "the relay refuses seq 1");
|
|
assert!(sent.elapsed() >= hold, "not before its hold ran out: {:?}", sent.elapsed());
|
|
assert_eq!(f.waiting(&a.room), 0);
|
|
// Seq 0, sent now, is forwarded: no input after it was.
|
|
send(&a.clients[1], 2, 0, &msf(&first)).await;
|
|
let got = a.received(&msf(&first)).await;
|
|
assert_eq!(got, vec![msf(&first)], "seq 1 withheld, seq 0 forwarded");
|
|
assert_eq!(f.refused_inputs(), 1);
|
|
assert_eq!(a.close().await, vec![("input_out_of_order".to_string(), 1)]);
|
|
}
|