From 7a188614b1718adc796bfaa747efa8095df15fcb Mon Sep 17 00:00:00 2001 From: Transaction Science Date: Sat, 24 Oct 2026 21:30:05 -0400 Subject: [PATCH 1/3] joule-moq-relay: count each author's audience inputs in its chain's order, not in arrival order Each group of a member's events track rides a stream of its own, and the relay reads each stream in a task of its own, so two inputs sent back to back reached the floor in either order. The floor counted them as they came: with a limit of one, it could forward a member's second input (seq 1) and refuse its first (seq 0), leaving subscribers with seq 1 and a gap at seq 0, the opposite of what a verifier counting in chain order decides. The floor now keeps each author's chain as it has received it (the first seq it lacks, the seqs above it, and the inputs waiting) and decides an input once it holds every record of the chain below it and no input below it waits. Floor::events_filter is async: an input ahead of a gap waits, with the rest of its stream, until the gap fills or the floor's hold runs out ([floor] hold_ms, 2 000 by default). One whose hold runs out is refused input_out_of_order, whichever object of its author's finds it so, and the chain is 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. Floor::events_step is the synchronous step, tested with explicit instants; Floor::waiting counts the inputs held. Tests: tests/floor.rs sends a member's seq 1 first, waits until the relay holds it, then sends seq 0: seq 0 forwarded, seq 1 refused input_over_limit; and seq 1 alone, refused input_out_of_order no sooner than the hold, after which a late seq 0 is forwarded. Unit tests cover a chat between two votes, an object whose records are out of seq order, the chain counted from after a refused input, and an input whose stream is gone. The hardened test from Train D is kept; its config gains `..FloorConfig::default()`. Co-Authored-By: Claude Opus 5.5 --- .../crates/joule-moq-relay/relay.example.toml | 6 +- .../crates/joule-moq-relay/src/config.rs | 19 +- .../crates/joule-moq-relay/src/floor.rs | 450 +++++++++++++++--- .../crates/joule-moq-relay/src/relay.rs | 2 +- .../crates/joule-moq-relay/src/session.rs | 6 +- .../joule-moq-relay/tests/age_room_service.rs | 2 +- .../crates/joule-moq-relay/tests/floor.rs | 150 +++++- .../joule-moq-room/tests/co_performance.rs | 2 +- 8 files changed, 558 insertions(+), 79 deletions(-) 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..09732528b 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,137 @@ 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; + assert!(until(STEP, || f.waiting(&a.room) == 1).await, "the relay holds seq 1 for seq 0"); + 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. + let got = a.received(&msf(&first)).await; + assert!(until(STEP, || f.refused_inputs() >= 1).await, "the relay refuses seq 1"); + assert_eq!(got, vec![msf(&first)], "seq 0 forwarded, seq 1 withheld"); + 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") } -- 2.54.0 (Apple Git-157) From 0517a563c981774c8480846e477ee28f1e83388f Mon Sep 17 00:00:00 2001 From: Transaction Science Date: Sat, 24 Oct 2026 21:30:36 -0400 Subject: [PATCH 2/3] relay floor test: the payload check is the one an arrival-order relay fails first Co-Authored-By: Claude Opus 5.5 --- .../joule-code-rs/crates/joule-moq-relay/tests/floor.rs | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) 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 09732528b..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 @@ -403,13 +403,15 @@ async fn chain_order() { // 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; - assert!(until(STEP, || f.waiting(&a.room) == 1).await, "the relay holds seq 1 for seq 0"); + 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 seq 1"); + 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)]); } -- 2.54.0 (Apple Git-157) From 6b65ab5ffb62fd7ddc3be838031ce988fc9fd301 Mon Sep 17 00:00:00 2001 From: Transaction Science Date: Sat, 24 Oct 2026 21:34:49 -0400 Subject: [PATCH 3/3] =?UTF-8?q?wai.live.coperformance=20=C2=A77.3:=20a=20r?= =?UTF-8?q?elay=20counts=20an=20author's=20inputs=20in=20its=20chain's=20o?= =?UTF-8?q?rder?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The relay MUST count an author's inputs in the order of the author's chain (live-events §6.2), as a verifier does, never in the order they reach it: one chain's records can reach it on several streams, one for each group. It 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, refuses one whose gap has not filled by then as input_out_of_order and counts 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, now naming both): a verifier holds each chain in order and counts each input where §9.2 step 4 meets it, so §10, the corpus, wai-rs and co_performance_verify.py are unchanged. §12 notes that an author leaving a gap in its own chain holds up only its own inputs, for no longer than the bound. CHANGELOG. Co-Authored-By: Claude Opus 5.5 --- wai/CHANGELOG.md | 24 ++++++++++++++++++++++++ wai/extensions/co-performance.md | 20 +++++++++++++++++++- 2 files changed, 43 insertions(+), 1 deletion(-) 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 -- 2.54.0 (Apple Git-157)