diff --git a/joule-code/joule-code-rs/crates/joule-moq-relay/relay.example.toml b/joule-code/joule-code-rs/crates/joule-moq-relay/relay.example.toml index 1c1e54119..d99ef2ba4 100644 --- a/joule-code/joule-code-rs/crates/joule-moq-relay/relay.example.toml +++ b/joule-code/joule-code-rs/crates/joule-moq-relay/relay.example.toml @@ -111,9 +111,13 @@ max_bytes = 4294967296 # and handovers from the events tracks it forwards, and does not forward a media object a # performer publishes out of turn under a schedule that refuses. A schedule is honoured from a # room key (these, and the keys of `auth.authorities`) or from a principal whose session proved -# a grant with the direct power. Absent: the relay holds no floor. +# a grant with the direct power. Absent: the relay holds no floor. The relay counts each +# author's audience inputs in the order of its chain (co-performance §7.3): an input that arrives +# ahead of a gap in its author's chain waits up to `hold_ms` for the gap to fill, and is refused +# `input_out_of_order` if it has not. [floor] room_keys = [] +hold_ms = 2000 # A room authority (an example; not a default): every WRG1 grant its key signs is honoured, # beside its proof and within its window, with the rights live-rooms §10.2 gives it. `rooms` diff --git a/joule-code/joule-code-rs/crates/joule-moq-relay/src/config.rs b/joule-code/joule-code-rs/crates/joule-moq-relay/src/config.rs index 308cfb836..77e48f3d0 100644 --- a/joule-code/joule-code-rs/crates/joule-moq-relay/src/config.rs +++ b/joule-code/joule-code-rs/crates/joule-moq-relay/src/config.rs @@ -385,7 +385,7 @@ pub struct DvrConfig { } /// The floor of WAI rooms (WAI co-performance §6, `crate::floor`). -#[derive(Clone, Debug, Default, Deserialize)] +#[derive(Clone, Debug, Deserialize)] #[serde(deny_unknown_fields)] pub struct FloorConfig { /// Room keys (Ed25519, hex) whose schedules and handovers the relay honours in every room: @@ -394,6 +394,21 @@ pub struct FloorConfig { /// keys of `auth.authorities` are room keys too. #[serde(default)] pub room_keys: Vec, + /// How long an audience input that arrives ahead of a gap in its author's chain waits for + /// the gap to fill before it is refused `input_out_of_order` (WAI co-performance §7.3): the + /// relay counts each author's inputs in the order of its chain. + #[serde(default = "d_floor_hold")] + pub hold_ms: u64, +} + +fn d_floor_hold() -> u64 { + crate::floor::DEFAULT_HOLD_MS +} + +impl Default for FloorConfig { + fn default() -> Self { + FloorConfig { room_keys: Vec::new(), hold_ms: d_floor_hold() } + } } impl FloorConfig { @@ -534,6 +549,6 @@ mod tests { assert!(long.check().is_err(), "a window past 24 hours"); let floor = c.floor.as_ref().expect("the example holds the floor"); assert!(floor.keys().expect("its room keys read").is_empty()); - assert!(FloorConfig { room_keys: vec!["00".into()] }.keys().is_err(), "a room key of one byte"); + assert!(FloorConfig { room_keys: vec!["00".into()], ..FloorConfig::default() }.keys().is_err(), "a room key of one byte"); } } diff --git a/joule-code/joule-code-rs/crates/joule-moq-relay/src/floor.rs b/joule-code/joule-code-rs/crates/joule-moq-relay/src/floor.rs index 646ff932f..24cc7e169 100644 --- a/joule-code/joule-code-rs/crates/joule-moq-relay/src/floor.rs +++ b/joule-code/joule-code-rs/crates/joule-moq-relay/src/floor.rs @@ -43,14 +43,26 @@ //! A refused record is withheld and never replaced (live-moderation §8.1): its author's chain //! shows the gap. A repeat of a record already decided (a group's first object repeats the //! chain, live-events §8.2) is decided as the first was. +//! +//! The relay counts an author's inputs in the order of the author's chain (live-events §6.2), +//! never in the order they reach it (§7.3): each group of an events track rides a stream of its +//! own (live-events §8.1), and each stream is read in a task of its own, so two inputs sent one +//! after the other can be taken in either order. An input is decided once the relay has +//! received every record of its author's chain below its `seq` and decided every input among +//! them. One that arrives ahead of a gap waits for it ([`Floor::events_filter`]), at most the +//! floor's hold, and is refused `input_out_of_order` if the gap has not filled by then; the +//! author's chain is then counted from the record after it. An input that arrives after a later +//! input of its author was forwarded is refused `input_out_of_order` too: counting it then would +//! count it after an input that follows it in the chain. use crate::events::{self, Event}; use joule_moq::grant_proof::GrantHead; use joule_moq::TrackNamespace; -use std::collections::{HashMap, HashSet, VecDeque}; +use std::collections::{BTreeSet, HashMap, HashSet, VecDeque}; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Mutex; -use std::time::Instant; +use std::time::{Duration, Instant}; +use tokio::sync::Notify; /// A schedule's magic (§6.1) and a handover's (§6.3). pub const SCHEDULE_MAGIC: &[u8; 4] = b"WPT1"; @@ -66,6 +78,15 @@ pub const AUDIENCE_ATTR: &str = "audience_input"; const MAX_ROOMS: usize = 256; /// The records a room keeps the decision of, for the repeats of a group's first object. const MAX_DECIDED: usize = 65_536; +/// How long an input that arrives ahead of a gap in its author's chain waits for the gap to +/// fill, by default (§7.3). +pub const DEFAULT_HOLD_MS: u64 = 2_000; +/// The reason of an input the relay cannot count in its author's chain order (§7.3). +pub const OUT_OF_ORDER: &str = "input_out_of_order"; +/// The seqs of an author's chain a room keeps above the first it lacks. +const MAX_AHEAD: usize = 1024; +/// The authors' chains a room keeps; past it, one with no input waiting is dropped. +const MAX_CHAINS: usize = 4096; /// The `meta` kind (live-events §4.2). pub const KIND_META: u8 = 0x01; /// A schedule's modes (§6.5). @@ -203,6 +224,71 @@ fn read_input(b: &[u8]) -> Option<([u8; 32], u8, Option)> { (r.p == b.len()).then_some((window, ty, option)) } +/// One author's chain as the relay has received it on the author's events track (live-events +/// §6.2): where its inputs stand in it, so that they are counted in its order (§7.3). +#[derive(Default)] +struct Chain { + /// The first seq the relay lacks: every seq below it it has received, or passed when a hold + /// ran out. + next: u64, + /// The seqs above `next` it has received. + ahead: BTreeSet, + /// The inputs received and not yet decided, by digest: each one's seq and the instant its + /// hold runs out. + waiting: HashMap<[u8; 32], (u64, Instant)>, + /// The greatest seq of an input of the author's the relay has forwarded. + forwarded: Option, +} + +impl Chain { + /// The record at `seq` has arrived; whether the run of seqs the relay holds grew. + fn receive(&mut self, seq: u64) -> bool { + if seq == self.next { + self.advance(); + return true; + } + if seq > self.next && self.ahead.len() < MAX_AHEAD { + self.ahead.insert(seq); + } + false + } + + /// The chain is counted from the record after `seq`: a hold ran out on the gap below it. + fn pass(&mut self, seq: u64) { + if seq >= self.next { + self.next = seq; + self.ahead = self.ahead.split_off(&seq); + self.advance(); + } + } + + fn advance(&mut self) { + self.ahead.remove(&self.next); + self.next = self.next.saturating_add(1); + while self.ahead.remove(&self.next) { + self.next = self.next.saturating_add(1); + } + } + + /// Whether the input at `seq` may be decided: the relay holds every record below it, and no + /// input below it waits. + fn ready(&self, seq: u64) -> bool { + self.next > seq && !self.waiting.values().any(|&(s, _)| s < seq) + } +} + +/// What the floor has decided of one events-track object so far, over the steps of its hold +/// ([`Floor::events_step`]). +#[derive(Debug, Default)] +pub struct Pass { + /// Each input of the object decided: `None` forwarded, or the reason refused. + verdicts: HashMap<[u8; 32], Option<&'static str>>, + /// When each input of the object stops waiting for its turn. + ends: HashMap<[u8; 32], Instant>, + /// The reason of each input the relay refused while it took the object. + refused: Vec<&'static str>, +} + struct RoomFloor { /// In the order the relay honoured them: the last whose `from` is reached is in effect. schedules: Vec, @@ -217,6 +303,8 @@ struct RoomFloor { /// Each author's records forwarded, by digest, and their `t`s, latest last: its rate. counted: HashSet<[u8; 32]>, times: HashMap<[u8; 32], VecDeque>, + /// Each author's chain as the relay has received it. + chains: HashMap<[u8; 32], Chain>, touched: Instant, } @@ -232,6 +320,7 @@ impl Default for RoomFloor { decided: HashMap::new(), counted: HashSet::new(), times: HashMap::new(), + chains: HashMap::new(), touched: Instant::now(), } } @@ -280,6 +369,10 @@ pub struct Floor { rooms: Mutex>, refused: AtomicU64, refused_inputs: AtomicU64, + /// How long an input ahead of a gap in its author's chain waits for it. + hold: Duration, + /// An author's chain moved on: a record arrived, or an input was decided. + chained: Notify, } /// What the relay forwards of one events-track object ([`Floor::events_filter`]). @@ -290,7 +383,8 @@ pub struct Filtered { pub payload: Option>, /// Every record of it is withheld: the object is not forwarded. pub withheld: bool, - /// The reason of each input this object refused for the first time. + /// The reason of each input the relay refused while it took this object: its own refused + /// for the first time, and any of its author's whose hold ran out meanwhile. pub refused: Vec<&'static str>, } @@ -318,9 +412,59 @@ fn count(f: &mut RoomFloor, e: &Event) { } } +/// Record a decision, forgetting every earlier one once a room holds its bound of them. +fn decide(decided: &mut HashMap<[u8; 32], Option<&'static str>>, digest: [u8; 32], v: Option<&'static str>) { + if decided.len() >= MAX_DECIDED { + decided.clear(); + } + decided.insert(digest, v); +} + +/// The input `e` of `owner`'s own track, decided if its turn in its author's chain has come +/// (§7.3): `Some` its verdict, or `None` while it waits. +#[allow(clippy::too_many_arguments)] +fn take_input(f: &mut RoomFloor, room: [u8; 32], owner: [u8; 32], e: &Event, rec: &[u8], limit: Option, grant: Option<&GrantHead>, pass: &mut Pass) -> Option> { + if let Some(v) = pass.verdicts.get(&e.digest) { + return Some(*v); + } + if let Some(v) = f.decided.get(&e.digest).copied() { + // A repeat: decided as the first was, unless its signature is another's. + let v = if v.is_none() && !e.verifies(rec) { Some("bad_signature") } else { v }; + if let Some(c) = f.chains.get_mut(&owner) { + c.waiting.remove(&e.digest); + } + pass.verdicts.insert(e.digest, v); + return Some(v); + } + let chain = f.chains.get(&owner)?; + if !chain.ready(e.seq) { + return None; + } + let behind = chain.forwarded.is_some_and(|m| m > e.seq); + let v = input_verdict(f, e, rec, limit, grant, behind.then_some(OUT_OF_ORDER)); + decide(&mut f.decided, e.digest, v); + let chain = f.chains.get_mut(&owner).expect("the author's chain"); + chain.waiting.remove(&e.digest); + match v { + None => { + chain.forwarded = Some(chain.forwarded.map_or(e.seq, |m| m.max(e.seq))); + let (w, _, _) = read_input(&e.payload).expect("forwarded inputs read"); + *f.sent.entry((w, owner)).or_default() += 1; + count(f, e); + } + Some(why) => { + pass.refused.push(why); + tracing::info!(target: "joule_moq_relay::floor", room = %hex::encode(room), principal = %hex::encode(owner), seq = e.seq, reason = why, "an audience input refused"); + } + } + pass.verdicts.insert(e.digest, v); + Some(v) +} + /// §7.3's checks over an input record of its author's own track: `None` to forward it, or the -/// reason it is refused. -fn input_verdict(f: &RoomFloor, e: &Event, rec: &[u8], limit: Option, grant: Option<&GrantHead>) -> Option<&'static str> { +/// reason it is refused. `order` is the reason to refuse it with when it cannot be counted in +/// its author's chain order, once the checks that do not count have passed. +fn input_verdict(f: &RoomFloor, e: &Event, rec: &[u8], limit: Option, grant: Option<&GrantHead>, order: Option<&'static str>) -> Option<&'static str> { if !e.verifies(rec) { return Some("bad_signature"); } @@ -333,6 +477,9 @@ fn input_verdict(f: &RoomFloor, e: &Event, rec: &[u8], limit: Option, grant if w.types & (1 << (ty - 1)) == 0 || option.is_some_and(|o| o >= w.options) { return Some("input_type_mismatch"); } + if order.is_some() { + return order; + } if f.sent.get(&(window, e.principal)).copied().unwrap_or(0) + 1 > limit { return Some("input_over_limit"); } @@ -346,9 +493,22 @@ fn input_verdict(f: &RoomFloor, e: &Event, rec: &[u8], limit: Option, grant } impl Floor { - /// A floor whose room keys are `room_keys` (the authorities' and the configured ones). + /// A floor whose room keys are `room_keys` (the authorities' and the configured ones), + /// holding an input ahead of a gap in its author's chain for [`DEFAULT_HOLD_MS`]. pub fn new(room_keys: Vec<[u8; 32]>) -> Floor { - Floor { room_keys, rooms: Mutex::new(HashMap::new()), refused: AtomicU64::new(0), refused_inputs: AtomicU64::new(0) } + Floor { + room_keys, + rooms: Mutex::new(HashMap::new()), + refused: AtomicU64::new(0), + refused_inputs: AtomicU64::new(0), + hold: Duration::from_millis(DEFAULT_HOLD_MS), + chained: Notify::new(), + } + } + + /// The floor, holding an input ahead of a gap in its author's chain for `hold` (§7.3). + pub fn holding(self, hold: Duration) -> Floor { + Floor { hold, ..self } } /// Whether the floor holds the selection window `window` (its record's digest) of `room`: @@ -357,35 +517,67 @@ impl Floor { self.rooms.lock().unwrap().get(room).is_some_and(|f| f.windows.contains_key(window)) } + /// The audience inputs of `room` that wait for their turn in their authors' chains (§7.3). + pub fn waiting(&self, room: &[u8; 32]) -> usize { + self.rooms.lock().unwrap().get(room).map_or(0, |f| f.chains.values().map(|c| c.waiting.len()).sum()) + } + /// The objects held back so far, out of turn under refuse. pub fn refused(&self) -> u64 { self.refused.load(Ordering::Relaxed) } - /// One object of the events track `ns` the relay forwards: [`Floor::events_filter`] with no - /// session grant, its outcome left aside. For the tests of the floor alone. + /// One object of the events track `ns` the relay forwards: one [`Floor::events_step`] with + /// no session grant, its outcome left aside. For the tests of the floor alone. pub fn events_object(&self, ns: &TrackNamespace, payload: &[u8], directors: &[[u8; 32]]) { - let _ = self.events_filter(ns, payload, directors, &[], 0); + let _ = self.events_step(ns, payload, directors, &[], 0, Instant::now(), &mut Pass::default()); } /// One object of the events track `ns`, before the relay forwards it: each schedule, /// handover and window of it the relay honours, and each commitment of a room key, goes - /// into its room's floor; each audience input of it is held to §7.3, and one refused is - /// withheld. `directors` are the principals of the grants with the direct power the - /// publishing session proved in this room, and `grants` those grants' heads, every one it - /// proved in this room; `now_unix_s` is when it arrived. - pub fn events_filter(&self, ns: &TrackNamespace, payload: &[u8], directors: &[[u8; 32]], grants: &[GrantHead], now_unix_s: u64) -> Filtered { - let mut out = Filtered::default(); - let (Some(room), Some(owner)) = (crate::dvr::room_of(ns), key_field(ns, 1)) else { return out }; - if !(ns.0.len() == 3 && ns.0[2] == b"events") { - return out; + /// into its room's floor; each audience input of it is held to §7.3, in its author's chain + /// order, and one refused is withheld. An input that arrives ahead of a gap in its author's + /// chain waits, and the object with it, until the gap fills or the floor's hold runs out. + /// `directors` are the principals of the grants with the direct power the publishing + /// session proved in this room, and `grants` those grants' heads, every one it proved in + /// this room; `now_unix_s` is when it arrived. + pub async fn events_filter(&self, ns: &TrackNamespace, payload: &[u8], directors: &[[u8; 32]], grants: &[GrantHead], now_unix_s: u64) -> Filtered { + let mut pass = Pass::default(); + let mut reached: Option = None; + loop { + let notified = self.chained.notified(); + tokio::pin!(notified); + notified.as_mut().enable(); + let now = reached.map_or_else(Instant::now, |t| t.max(Instant::now())); + match self.events_step(ns, payload, directors, grants, now_unix_s, now, &mut pass) { + Ok(out) => return out, + Err(until) => { + if tokio::time::timeout_at(tokio::time::Instant::from_std(until), notified).await.is_err() { + reached = Some(until); + } + } + } } - let Some(recs) = events::records(payload) else { return out }; + } + + /// One step of [`Floor::events_filter`] at `now`, under `pass`, what the object's earlier + /// steps decided: what the relay forwards of the object, or, while one of its inputs waits + /// for its turn, the instant by which to take it again. Every record of the object is in + /// its author's chain from its first step, and the floor holds its schedules, handovers, + /// windows and commitments from then. + #[allow(clippy::too_many_arguments)] + pub fn events_step(&self, ns: &TrackNamespace, payload: &[u8], directors: &[[u8; 32]], grants: &[GrantHead], now_unix_s: u64, now: Instant, pass: &mut Pass) -> Result { + let (Some(room), Some(owner)) = (crate::dvr::room_of(ns), key_field(ns, 1)) else { return Ok(Filtered::default()) }; + if !(ns.0.len() == 3 && ns.0[2] == b"events") { + return Ok(Filtered::default()); + } + let Some(recs) = events::records(payload) else { return Ok(Filtered::default()) }; let room_key = self.room_keys.contains(&owner); let direct = room_key || directors.contains(&owner); // The author's audience-input power and rate window, from the grant its session proved. let grant = grants.iter().find(|g| g.room == room && g.principal == owner && g.good_at(now_unix_s)); let limit = grant.and_then(|g| g.attr(AUDIENCE_ATTR)).and_then(decimal).filter(|n| (1..=255).contains(n)); + let refused_before = pass.refused.len(); let mut rooms = self.rooms.lock().unwrap(); if !rooms.contains_key(&room) && rooms.len() >= MAX_ROOMS && let Some(old) = rooms.iter().min_by_key(|(_, f)| f.touched).map(|(r, _)| *r) @@ -394,45 +586,48 @@ impl Floor { } let f = rooms.entry(room).or_default(); f.touched = Instant::now(); - let mut withheld: HashSet<[u8; 32]> = HashSet::new(); - for (_, rec) in &recs { - let Some(e) = Event::read(rec) else { continue }; - if e.room != room || e.principal != owner { + let own: Vec<(Event, &[u8])> = recs.iter().filter_map(|(_, rec)| Event::read(rec).filter(|e| e.room == room && e.principal == owner).map(|e| (e, rec.as_slice()))).collect(); + let is_input = |e: &Event| e.kind == KIND_META && e.payload.get(..4) == Some(&INPUT_MAGIC[..]); + // Every record of the object takes its place in its author's chain, and each input of it + // not yet decided waits there until its turn comes (§7.3). + if !f.chains.contains_key(&owner) && f.chains.len() >= MAX_CHAINS + && let Some(idle) = f.chains.iter().find(|(_, c)| c.waiting.is_empty()).map(|(k, _)| *k) + { + f.chains.remove(&idle); + } + let chain = f.chains.entry(owner).or_default(); + let mut moved = false; + for (e, _) in &own { + moved |= chain.receive(e.seq); + if is_input(e) && !pass.verdicts.contains_key(&e.digest) && !f.decided.contains_key(&e.digest) { + let ends = [pass.ends.get(&e.digest).copied(), chain.waiting.get(&e.digest).map(|w| w.1)]; + let end = ends.into_iter().flatten().min().unwrap_or(now + self.hold); + pass.ends.insert(e.digest, end); + chain.waiting.insert(e.digest, (e.seq, end)); + } + } + // An input whose hold has run out is refused, whichever object carries it, and the + // author's chain is counted from the record after it. + let mut out_of_time: Vec<([u8; 32], u64)> = chain.waiting.iter().filter(|(_, w)| w.1 <= now).map(|(d, w)| (*d, w.0)).collect(); + out_of_time.sort_by_key(|x| x.1); + for (d, seq) in out_of_time { + chain.waiting.remove(&d); + chain.pass(seq); + decide(&mut f.decided, d, Some(OUT_OF_ORDER)); + if pass.ends.contains_key(&d) { + pass.verdicts.insert(d, Some(OUT_OF_ORDER)); + } + pass.refused.push(OUT_OF_ORDER); + moved = true; + tracing::info!(target: "joule_moq_relay::floor", room = %hex::encode(room), principal = %hex::encode(owner), seq, reason = OUT_OF_ORDER, "an audience input refused"); + } + for (e, rec) in &own { + if is_input(e) { + moved |= !pass.verdicts.contains_key(&e.digest) && take_input(f, room, owner, e, rec, limit, grant, pass).is_some(); continue; } - let magic = (e.kind == KIND_META && e.payload.len() >= 4).then(|| &e.payload[..4]); - if magic == Some(&INPUT_MAGIC[..]) { - let verdict = match f.decided.get(&e.digest) { - // A repeat: decided as the first was, unless its signature is another's. - Some(None) if !e.verifies(rec) => Some("bad_signature"), - Some(v) => *v, - None => { - let v = input_verdict(f, &e, rec, limit, grant); - if f.decided.len() >= MAX_DECIDED { - f.decided.clear(); - } - f.decided.insert(e.digest, v); - match v { - None => { - let (w, _, _) = read_input(&e.payload).expect("forwarded inputs read"); - *f.sent.entry((w, owner)).or_default() += 1; - count(f, &e); - } - Some(why) => { - out.refused.push(why); - tracing::info!(target: "joule_moq_relay::floor", room = %hex::encode(room), principal = %hex::encode(owner), reason = why, "an audience input refused"); - } - } - v - } - }; - if verdict.is_some() { - withheld.insert(e.digest); - } - continue; - } - count(f, &e); - let Some(magic) = magic else { continue }; + count(f, e); + let Some(magic) = (e.kind == KIND_META && e.payload.len() >= 4).then(|| &e.payload[..4]) else { continue }; let m4: &[u8; 4] = magic.try_into().expect("4 bytes"); if ![SCHEDULE_MAGIC, HANDOVER_MAGIC, WINDOW_MAGIC, COMMIT_MAGIC].contains(&m4) || f.seen.contains(&e.digest) || !e.verifies(rec) { continue; @@ -481,16 +676,33 @@ impl Floor { let order = f.order; f.handovers.push(Handover { schedule: s.digest, segment, turn, at, to, order }); } + // The object's inputs still waiting, in their chain's order: one decided above may have + // been the turn another waited for. + let mut left: Vec<&(Event, &[u8])> = own.iter().filter(|(e, _)| is_input(e) && !pass.verdicts.contains_key(&e.digest)).collect(); + left.sort_by_key(|(e, _)| e.seq); + for (e, rec) in left { + moved |= take_input(f, room, owner, e, rec, limit, grant, pass).is_some(); + } + let waits = own.iter().any(|(e, _)| is_input(e) && !pass.verdicts.contains_key(&e.digest)); + let until = f.chains.get(&owner).and_then(|c| c.waiting.values().map(|w| w.1).min()); + let withheld: HashSet<[u8; 32]> = own.iter().filter(|(e, _)| matches!(pass.verdicts.get(&e.digest), Some(Some(_)))).map(|(e, _)| e.digest).collect(); drop(rooms); + self.refused_inputs.fetch_add((pass.refused.len() - refused_before) as u64, Ordering::Relaxed); + if moved { + self.chained.notify_waiters(); + } + if waits { + return Err(until.unwrap_or(now + self.hold)); + } + let mut out = Filtered { refused: std::mem::take(&mut pass.refused), ..Filtered::default() }; if withheld.is_empty() { - return out; + return Ok(out); } - self.refused_inputs.fetch_add(out.refused.len() as u64, Ordering::Relaxed); match events::withhold(payload, |d| withheld.contains(d)) { None => out.withheld = true, Some(kept) => out.payload = Some(kept.into_owned()), } - out + Ok(out) } /// The audience inputs refused so far (§7.3). @@ -575,6 +787,11 @@ mod tests { const T0: u64 = 1_792_540_800_000_000; const S: u64 = 1_000_000; + /// One events-track object taken by the floor now, which decides it at once. + fn filter(f: &Floor, ns: &TrackNamespace, payload: &[u8], directors: &[[u8; 32]], grants: &[GrantHead], now_s: u64) -> Filtered { + f.events_step(ns, payload, directors, grants, now_s, Instant::now(), &mut Pass::default()).expect("decided at once") + } + fn key(i: u8) -> SigningKey { SigningKey::from_bytes(&[i; 32]) } @@ -759,7 +976,7 @@ mod tests { let input = |who: u8, g: &GrantHead, seq: u64, t_s: u64, payload: &[u8]| { let r = record(&key(who), room, seq, t_s * S, KIND_META, payload); let d = Event::read(&r).unwrap().digest; - (f.events_filter(&ns(room, pk(who), "events"), format!("[{}]", entry(&r, t_s * 1000)).as_bytes(), &[], std::slice::from_ref(g), now_s), d, r) + (filter(&f, &ns(room, pk(who), "events"), format!("[{}]", entry(&r, t_s * 1000)).as_bytes(), &[], std::slice::from_ref(g), now_s), d, r) }; let vw = record(&key(clock), room, 0, S, KIND_META, &window(3, 60)); let vw_d = Event::read(&vw).unwrap().digest; @@ -782,7 +999,7 @@ mod tests { let mut forged = record(&key(m2), room, 0, 3 * S, KIND_META, &vote(&vw_d, 1)); let n = forged.len(); forged[n - 1] ^= 1; - let x = f.events_filter(&ns(room, pk(m2), "events"), format!("[{}]", entry(&forged, 3000)).as_bytes(), &[], std::slice::from_ref(&g2), now_s); + let x = filter(&f, &ns(room, pk(m2), "events"), format!("[{}]", entry(&forged, 3000)).as_bytes(), &[], std::slice::from_ref(&g2), now_s); refused(x, "bad_signature"); // Its second input forwarded, its third over its power of two. assert_eq!(input(m1, &g1, 1, 4, &prompt(&pw_d, "slower")).0, ok, "another window: its own count"); @@ -791,21 +1008,21 @@ mod tests { // m2's policy allows three events a second: the fourth within one is over its rate. let chat = |seq: u64, t_us: u64| record(&key(m2), room, seq, t_us, 0x03, b"\x02\x00hi"); let obj = format!("[{},{}]", entry(&chat(1, 10 * S), 10_000), entry(&chat(2, 10 * S + 1), 10_000)); - assert_eq!(f.events_filter(&ns(room, pk(m2), "events"), obj.as_bytes(), &[], std::slice::from_ref(&g2), now_s), ok); + assert_eq!(filter(&f, &ns(room, pk(m2), "events"), obj.as_bytes(), &[], std::slice::from_ref(&g2), now_s), ok); let r3 = record(&key(m2), room, 3, 10 * S + 2, KIND_META, &vote(&vw_d, 1)); - let x = f.events_filter(&ns(room, pk(m2), "events"), format!("[{}]", entry(&r3, 10_000)).as_bytes(), &[], std::slice::from_ref(&g2), now_s); + let x = filter(&f, &ns(room, pk(m2), "events"), format!("[{}]", entry(&r3, 10_000)).as_bytes(), &[], std::slice::from_ref(&g2), now_s); assert_eq!(x, ok, "the third event within the second"); let r4 = record(&key(m2), room, 4, 10 * S + 3, KIND_META, &vote(&vw_d, 2)); - let x = f.events_filter(&ns(room, pk(m2), "events"), format!("[{}]", entry(&r4, 10_000)).as_bytes(), &[], std::slice::from_ref(&g2), now_s); + let x = filter(&f, &ns(room, pk(m2), "events"), format!("[{}]", entry(&r4, 10_000)).as_bytes(), &[], std::slice::from_ref(&g2), now_s); refused(x, "rate_exceeded"); // An object of a chat and a refused input: the chat goes on alone, byte for byte; the // input's repeat in a later group's first object is withheld again, not counted again. let c5 = chat(5, 20 * S); let bad = record(&key(m2), room, 6, 20 * S, KIND_META, &vote(&vw_d, 7)); let both = format!("[{},{}]", entry(&c5, 20_000), entry(&bad, 20_000)); - let x = f.events_filter(&ns(room, pk(m2), "events"), both.as_bytes(), &[], std::slice::from_ref(&g2), now_s); + let x = filter(&f, &ns(room, pk(m2), "events"), both.as_bytes(), &[], std::slice::from_ref(&g2), now_s); assert_eq!((x.payload, x.withheld, x.refused), (Some(format!("[{}]", entry(&c5, 20_000)).into_bytes()), false, vec!["input_type_mismatch"])); - let x = f.events_filter(&ns(room, pk(m2), "events"), both.as_bytes(), &[], std::slice::from_ref(&g2), now_s); + let x = filter(&f, &ns(room, pk(m2), "events"), both.as_bytes(), &[], std::slice::from_ref(&g2), now_s); assert!(x.refused.is_empty() && x.payload.is_some(), "decided as the first was, counted once"); // Once the clock's commitment passes, the window takes no more. let mut c = b"WPK1\x01".to_vec(); @@ -823,6 +1040,105 @@ mod tests { assert_eq!(f.refused_inputs(), 12); } + #[test] + fn inputs_are_counted_in_their_chains_order() { + let room = [8; 32]; + let (authority, clock) = (key(1), 2); + let f = Floor::new(vec![pk(clock)]).holding(Duration::from_secs(2)); + let now_s = 1_800_000_000; + let vw = record(&key(clock), room, 0, S, KIND_META, &window(3, 60)); + let vw_d = Event::read(&vw).unwrap().digest; + f.events_object(&ns(room, pk(clock), "events"), format!("[{}]", entry(&vw, 1000)).as_bytes(), &[]); + // Each member may send one input to a window; member 6, three. + let grants: Vec = (3..=8).map(|m| grant(&authority, room, m, 50, &[(AUDIENCE_ATTR, if m == 6 { "3" } else { "1" })])).collect(); + let t0 = Instant::now(); + let at = |ms: u64| t0 + Duration::from_millis(ms); + let step = |who: u8, recs: &[Vec], ms: u64, p: &mut Pass| { + let obj = format!("[{}]", recs.iter().map(|r| entry(r, 2000)).collect::>().join(",")); + f.events_step(&ns(room, pk(who), "events"), obj.as_bytes(), &[], &grants, now_s, at(ms), p) + }; + let v = |who: u8, seq: u64, o: u8| record(&key(who), room, seq, 2 * S, KIND_META, &vote(&vw_d, o)); + let refused = |x: Result, why: &str| { + let x = x.expect("decided"); + assert!(x.withheld && x.refused == vec![why], "{why}: {x:?}"); + }; + let forwarded = || Ok::(Filtered::default()); + + // Member 3's seq 1 reaches the relay before its seq 0: it waits, and seq 0 is counted + // first, as a verifier counts them. + let (mut p0, mut p1) = (Pass::default(), Pass::default()); + assert_eq!(step(3, &[v(3, 1, 1)], 0, &mut p1), Err(at(2000)), "seq 1 waits for seq 0"); + assert_eq!(f.waiting(&room), 1); + assert_eq!(step(3, &[v(3, 0, 0)], 10, &mut p0), forwarded(), "seq 0, counted first"); + refused(step(3, &[v(3, 1, 1)], 10, &mut p1), "input_over_limit"); + assert_eq!(f.waiting(&room), 0); + + // Member 4's chain is a vote, a chat and a vote, taken last to first: the chat is not + // held, and fills no gap below the second vote while the first has not arrived. + let chat = record(&key(4), room, 1, 2 * S, 0x03, b"\x02\x00hi"); + let mut p2 = Pass::default(); + assert_eq!(step(4, &[v(4, 2, 2)], 0, &mut p2), Err(at(2000))); + assert_eq!(step(4, &[chat], 5, &mut Pass::default()), forwarded(), "a chat is not held"); + assert_eq!(step(4, &[v(4, 2, 2)], 5, &mut p2), Err(at(2000)), "the vote below it has not arrived"); + assert_eq!(step(4, &[v(4, 0, 0)], 10, &mut Pass::default()), forwarded()); + refused(step(4, &[v(4, 2, 2)], 10, &mut p2), "input_over_limit"); + + // An object of member 5's whose records are out of their chain's order: each input is + // decided in that order, and the object goes on without the one refused. + let (v1, v0) = (v(5, 1, 1), v(5, 0, 0)); + let x = step(5, &[v1, v0.clone()], 0, &mut Pass::default()).expect("both decided"); + assert_eq!((x.payload, x.withheld, x.refused), (Some(format!("[{}]", entry(&v0, 2000)).into_bytes()), false, vec!["input_over_limit"])); + assert_eq!(f.refused_inputs(), 3); + } + + #[test] + fn an_input_whose_gap_never_fills_is_refused_when_its_hold_runs_out() { + let room = [9; 32]; + let (authority, clock) = (key(1), 2); + let f = Floor::new(vec![pk(clock)]).holding(Duration::from_secs(2)); + let now_s = 1_800_000_000; + let vw = record(&key(clock), room, 0, S, KIND_META, &window(3, 60)); + let vw_d = Event::read(&vw).unwrap().digest; + f.events_object(&ns(room, pk(clock), "events"), format!("[{}]", entry(&vw, 1000)).as_bytes(), &[]); + let grants: Vec = (3..=5).map(|m| grant(&authority, room, m, 50, &[(AUDIENCE_ATTR, if m == 3 { "3" } else { "1" })])).collect(); + let t0 = Instant::now(); + let at = |ms: u64| t0 + Duration::from_millis(ms); + let step = |who: u8, rec: &[u8], ms: u64, p: &mut Pass| f.events_step(&ns(room, pk(who), "events"), format!("[{}]", entry(rec, 2000)).as_bytes(), &[], &grants, now_s, at(ms), p); + let v = |who: u8, seq: u64, o: u8| record(&key(who), room, seq, 2 * S, KIND_META, &vote(&vw_d, o)); + let refused = |x: Result, why: &str| { + let x = x.expect("decided"); + assert!(x.withheld && x.refused == vec![why], "{why}: {x:?}"); + }; + + // Member 3 (three inputs) sends seq 1 alone. It waits for the hold, then is refused. + let mut p = Pass::default(); + assert_eq!(step(3, &v(3, 1, 1), 0, &mut p), Err(at(2000))); + assert_eq!(step(3, &v(3, 1, 1), 1999, &mut p), Err(at(2000)), "still waiting"); + refused(step(3, &v(3, 1, 1), 2000, &mut p), OUT_OF_ORDER); + // Its chain is counted from the record after it: seq 2 is decided at once and forwarded, + // and seq 0, arriving now, would be counted after seq 2. + assert_eq!(step(3, &v(3, 2, 2), 2001, &mut Pass::default()), Ok(Filtered::default())); + refused(step(3, &v(3, 0, 0), 2002, &mut Pass::default()), OUT_OF_ORDER); + + // Member 4's seq 1 is refused when its hold runs out; its seq 0, later still, is forwarded: + // nothing after it was counted. + let mut p = Pass::default(); + assert!(step(4, &v(4, 1, 1), 0, &mut p).is_err()); + refused(step(4, &v(4, 1, 1), 2500, &mut p), OUT_OF_ORDER); + assert_eq!(step(4, &v(4, 0, 0), 2600, &mut Pass::default()), Ok(Filtered::default())); + + // Member 5's seq 1 waits on a stream that is then gone. Its author's next object, after + // the hold, finds it out of time: refused, and counted with that object. + assert!(step(5, &v(5, 1, 1), 0, &mut Pass::default()).is_err()); + let chat = record(&key(5), room, 2, 3 * S, 0x03, b"\x02\x00hi"); + let x = step(5, &chat, 2100, &mut Pass::default()).expect("a chat"); + assert_eq!((x.payload, x.withheld, x.refused), (None, false, vec![OUT_OF_ORDER])); + // A repeat of it is decided as it was. + let x = step(5, &v(5, 1, 1), 2200, &mut Pass::default()).expect("a repeat"); + assert!(x.withheld && x.refused.is_empty(), "{x:?}"); + assert_eq!((f.waiting(&room), f.refused_inputs()), (0, 4)); + } + #[test] fn schedules_read_as_section_6_1_has_them() { let d = [0; 32]; diff --git a/joule-code/joule-code-rs/crates/joule-moq-relay/src/relay.rs b/joule-code/joule-code-rs/crates/joule-moq-relay/src/relay.rs index 6dfc7b5b3..7953c9897 100644 --- a/joule-code/joule-code-rs/crates/joule-moq-relay/src/relay.rs +++ b/joule-code/joule-code-rs/crates/joule-moq-relay/src/relay.rs @@ -426,7 +426,7 @@ impl Relay { Some(f) => { let mut keys = f.keys()?; keys.extend(auth.authority_keys()); - Some(crate::floor::Floor::new(keys)) + Some(crate::floor::Floor::new(keys).holding(std::time::Duration::from_millis(f.hold_ms))) } None => None, }; diff --git a/joule-code/joule-code-rs/crates/joule-moq-relay/src/session.rs b/joule-code/joule-code-rs/crates/joule-moq-relay/src/session.rs index b6f702665..374f64771 100644 --- a/joule-code/joule-code-rs/crates/joule-moq-relay/src/session.rs +++ b/joule-code/joule-code-rs/crates/joule-moq-relay/src/session.rs @@ -663,9 +663,11 @@ async fn upstream_subgroup(relay: Relay, ctx: Arc, h: joule_moq::dra continue; } // A room's events: its schedules, handovers, windows and commitments read, and an - // audience input §7.3 refuses withheld, before anything is forwarded or cached. + // audience input §7.3 refuses withheld, before anything is forwarded or cached. An + // input that arrives ahead of a gap in its author's chain waits here, and the rest + // of this stream with it, until the gap fills or the floor's hold runs out. if floor_events && o.status == ObjectStatus::Normal && let Some(f) = floor { - let v = f.events_filter(&key.0, &o.payload, &directors, &room_grants, crate::relay::now_unix_us() / 1_000_000); + let v = f.events_filter(&key.0, &o.payload, &directors, &room_grants, crate::relay::now_unix_us() / 1_000_000).await; if !v.refused.is_empty() { let mut t = ctx.tally.lock().unwrap(); for why in &v.refused { diff --git a/joule-code/joule-code-rs/crates/joule-moq-relay/tests/age_room_service.rs b/joule-code/joule-code-rs/crates/joule-moq-relay/tests/age_room_service.rs index 7efe41e84..f34d8166c 100644 --- a/joule-code/joule-code-rs/crates/joule-moq-relay/tests/age_room_service.rs +++ b/joule-code/joule-code-rs/crates/joule-moq-relay/tests/age_room_service.rs @@ -104,7 +104,7 @@ async fn run() { // The clock's own rule: it subscribes to every track, by its token; exempt from limits. c.auth.grants = vec![Grant { prefix: vec![], token_sha256: hex::encode(joule_moq_relay::auth::sha256(&token)), publish: false, subscribe: true, proof: false, unlimited: true, seek: false }]; // Co-performance: the relay holds the room's floor, the clock a room key. - c.floor = Some(FloorConfig { room_keys: vec![hexkey(&clock)] }); + c.floor = Some(FloorConfig { room_keys: vec![hexkey(&clock)], ..FloorConfig::default() }); }) .await; let target = format!("//{}{}", server.websocket_addr.unwrap(), server.relay().config().path); diff --git a/joule-code/joule-code-rs/crates/joule-moq-relay/tests/floor.rs b/joule-code/joule-code-rs/crates/joule-moq-relay/tests/floor.rs index e3ca4528e..e3ce23015 100644 --- a/joule-code/joule-code-rs/crates/joule-moq-relay/tests/floor.rs +++ b/joule-code/joule-code-rs/crates/joule-moq-relay/tests/floor.rs @@ -10,6 +10,11 @@ //! 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; @@ -21,7 +26,9 @@ 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 std::time::{Duration, SystemTime, UNIX_EPOCH}; +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 { @@ -103,7 +110,7 @@ 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)] })).await; + 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. @@ -231,7 +238,7 @@ async fn inputs() { 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)] }); + c.floor = Some(FloorConfig { room_keys: vec![hexkey(&clock)], ..FloorConfig::default() }); }) .await; let target = format!("//{}{}", server.websocket_addr.unwrap(), server.relay().config().path); @@ -271,7 +278,8 @@ async fn inputs() { 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 limit counts the votes in the order the relay takes them. + // 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; @@ -296,3 +304,139 @@ async fn inputs() { 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, 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 { + 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> { + 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)]); +} diff --git a/joule-code/joule-code-rs/crates/joule-moq-room/tests/co_performance.rs b/joule-code/joule-code-rs/crates/joule-moq-room/tests/co_performance.rs index c37b61b8c..feac5e73a 100644 --- a/joule-code/joule-code-rs/crates/joule-moq-room/tests/co_performance.rs +++ b/joule-code/joule-code-rs/crates/joule-moq-room/tests/co_performance.rs @@ -53,7 +53,7 @@ async fn relay(keys: &Keys) -> RelayServer { Grant { prefix: vec!["*".into(), clock.clone(), "events".into()], token_sha256: t.clone(), publish: true, subscribe: false, proof: false, unlimited: true, seek: false }, Grant { prefix: vec![], token_sha256: t, publish: false, subscribe: true, proof: false, unlimited: false, seek: true }, ]; - c.floor = Some(FloorConfig { room_keys: vec![clock] }); + c.floor = Some(FloorConfig { room_keys: vec![clock], ..FloorConfig::default() }); RelayServer::start(c).await.expect("the relay starts") } diff --git a/wai/CHANGELOG.md b/wai/CHANGELOG.md index 71ee2439f..769bcec12 100644 --- a/wai/CHANGELOG.md +++ b/wai/CHANGELOG.md @@ -6,6 +6,30 @@ wire format may change and such changes are recorded here. ## v0.4.0 — current +### `wai.live.coperformance`: a relay counts audience inputs in their chain's order + +- **The rule** ([co-performance](extensions/co-performance.md) §7.3): a relay counts an + author's inputs in the order of the author's chain, as a verifier does, never in the order + they reach it. The records of one chain can reach a relay on several streams, one for each + group, read side by side, so a member's second input could be taken before its first, and a + relay counting as they came could forward the second and refuse the first, where a verifier + accepts the first and refuses the second, leaving a gap in the member's chain where the first + input was. A relay now decides an input once it has received every record of the chain below + it and decided every input among them; it holds an input that arrives ahead of a gap, for a + bound of its own, and refuses one whose gap has not filled by then as `input_out_of_order`, + counting the chain from the record after it; and it refuses as `input_out_of_order` an input + that arrives after a later input of its author was forwarded. The reason is the relay's + alone, as `window_committed` is (§8): a verifier holds each chain in order, and `wai-rs` and + `co_performance_verify.py` already count each input where the merged order meets it, so + neither changes and the corpus gains no vector. +- **The relay** (`joule-moq-relay`) keeps each author's chain as it has received it, holds an + input ahead of a gap with the rest of its stream until the gap fills or `[floor] hold_ms` + (2 000 by default) runs out, and counts `input_out_of_order` in the session's summary. + `tests/floor.rs`: a member's second vote, sent first on a stream of its own and held, then its + first: the first forwarded, the second refused `input_over_limit`; and a second vote whose first + never comes, refused `input_out_of_order` no sooner than the hold, after which the first, sent + late, is forwarded. + ### `wai.age.assurance`: the room's own service under the serving rule - **The rule** ([age-assurance](extensions/age-assurance.md) §5.5): a session that proves diff --git a/wai/extensions/co-performance.md b/wai/extensions/co-performance.md index 70e89237b..9b9cb8b2b 100644 --- a/wai/extensions/co-performance.md +++ b/wai/extensions/co-performance.md @@ -473,6 +473,20 @@ window's `close` is not refused: it was in flight when the window closed, and it room, under the member's grant; the input names it, and a performer that uses it fetches it by that name. +A relay **MUST** count an author's inputs in the order of the author's chain (live-events +§6.2), as a verifier does, and never in the order they reach it: the records of one chain can +reach it on several streams, one for each group (live-events §8.1), which it may read side by +side. It decides an input once it has received every record of the author's chain below the +input's `seq` and decided every input among them. An input that reaches it ahead of a gap in +the author's chain it holds, for at most a bound of its own, and decides once the gap fills; one +whose gap has not filled when the bound runs out it refuses as `input_out_of_order`, and it +counts the author's chain from the record after it. It also refuses as `input_out_of_order` an +input that reaches it after it has forwarded a later input of the same author, since counting it +then would count it after an input that follows it in the chain. A relay that holds no earlier +record of an author's chain, because it started after the chain did, so refuses that author's +first input once its bound runs out. Neither refusal is a verifier's: a verifier holds each +chain in its order and counts each input where §9.2 step 4 meets it. + ### 7.4 Moderation [Live-moderation](live-moderation.md) applies to inputs as to any `meta` event: a review policy @@ -579,7 +593,8 @@ cites a selection takes its parties from these fields. §9.2 refuses. Refusal is the relay's because only the relay sees each object before every subscriber does. A relay **SHOULD** count what it held back in its per-session record ([live-metering](live-metering.md) §3): the objects out of turn, and the inputs by the reason - §7.3 gives each, or `window_committed` for one naming a window already committed, which is + §7.3 gives each, `window_committed` for one naming a window already committed, or + `input_out_of_order` for one it could not count in its author's chain order (§7.3), which are the relay's alone and no verifier's. - **The room clock** signs marks (§6.7), commitments, selections and fidelity statements: what the room states about what happened. A clock that is the relay's key signs marks as it @@ -787,6 +802,9 @@ to fail. be chosen after the inputs are seen. What remains to the clock is to omit an input from the room, which leaves a gap in the member's chain (live-events §6.2), or to commit before an input in flight arrives, which the input's `t` below `close` shows. +- A relay holds an input that reaches it ahead of a gap in its author's chain for a bound of its + own, and refuses it once the bound runs out (§7.3). An author that leaves a gap in its own + chain holds up only its own inputs, and for no longer than that bound. - A sketch or a movement sample is named by digest, and its object is fetched only by a party the room's grants let subscribe; a movement sample can identify the person who made it, and it is personal data. Taking part in a room under a grant is taking part in its mechanics, and this