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`

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<String>,
+    /// 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");
     }
 }

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<u8>)> {
     (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<u64>,
+    /// 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<u64>,
+}
+
+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<Schedule>,
@@ -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<u64>>,
+    /// 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<HashMap<[u8; 32], RoomFloor>>,
     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<Vec<u8>>,
     /// 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<u32>, grant: Option<&GrantHead>, pass: &mut Pass) -> Option<Option<&'static str>> {
+    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<u32>, 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<u32>, 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<u32>, 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<u32>, 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<Instant> = 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<Filtered, Instant> {
+        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<GrantHead> = (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<u8>], ms: u64, p: &mut Pass| {
+            let obj = format!("[{}]", recs.iter().map(|r| entry(r, 2000)).collect::<Vec<_>>().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<Filtered, Instant>, why: &str| {
+            let x = x.expect("decided");
+            assert!(x.withheld && x.refused == vec![why], "{why}: {x:?}");
+        };
+        let forwarded = || Ok::<Filtered, Instant>(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<GrantHead> = (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<Filtered, Instant>, 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];

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,
         };

joule-code/joule-code-rs/crates/joule-moq-relay/src/session.rs

@@ -663,9 +663,11 @@ async fn upstream_subgroup(relay: Relay, ctx: Arc<SessionCtx>, 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 {

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);

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<u8>, k: &SigningKey| ClientOptions { grants: vec![(g, k.clone())], target: target.clone(), ..ClientOptions::default() };
+        let t0 = now_us() - 10_000_000;
+        let (c, _) = connect(&server, Via::WebSocket, ClientOptions { tokens: vec![joule_moq::draft22::Token::UseValue { token_type: 0, value: token }], ..ClientOptions::default() }).await;
+        let mut held = vec![c.publish(track(room, &clock, "events"), name("0"), 1, Parameters::new(), track_props(16, None)).await.expect("PUBLISH the clock's events")];
+        let mut w = b"WPW1\x01".to_vec();
+        w.extend_from_slice(&[1, 0b100, 3, 1]);
+        w.extend_from_slice(&(now_us() - t0 + 60_000_000).to_le_bytes());
+        w.extend_from_slice(&[0; 32]);
+        w.extend_from_slice(&0u16.to_le_bytes());
+        let wr = meta(&clock, room, 0, now_us() - t0, &w);
+        send(&c, 1, 0, &msf(&wr)).await;
+        let (s, _) = connect(&server, Via::WebSocket, with(viewer_grant(&authority, room, &watcher, &[]), &watcher)).await;
+        let (pm, _) = connect(&server, Via::WebSocket, with(viewer_grant(&authority, room, &member, &[("audience_input", "1")]), &member)).await;
+        held.push(pm.publish(track(room, &member, "events"), name("0"), 2, Parameters::new(), track_props(16, None)).await.expect("PUBLISH the member's events"));
+        let (_, tx, rx) = s.subscribe(track(room, &member, "events"), name("0"), Parameters::new()).await.expect("SUBSCRIBE the member's events");
+        held.push((tx, rx));
+        let window = digest(&wr);
+        let f = server.relay().floor().unwrap();
+        assert!(until(STEP, || f.holds_window(&room, &window)).await, "the relay holds the clock's window");
+        Audience { server, room, member, window, t0, clients: [c, pm, s], _held: held }
+    }
+
+    /// The member's vote for option `o` at seq `seq`, chained after `prev`.
+    fn vote(&self, seq: u64, prev: Option<&[u8]>, o: u8) -> Vec<u8> {
+        let mut b = b"WPA1\x01".to_vec();
+        b.extend_from_slice(&self.window);
+        b.extend_from_slice(&[3, o]);
+        let t = now_us() - self.t0;
+        match prev {
+            None => meta(&self.member, self.room, seq, t, &b),
+            Some(p) => meta_after(&self.member, self.room, seq, digest(p), t, &b),
+        }
+    }
+
+    /// What the watcher receives of the member's events: until `want` arrives, then a second more.
+    async fn received(&self, want: &[u8]) -> Vec<Vec<u8>> {
+        let s = &self.clients[2];
+        let mut got = collect(s, STEP, |g| g.iter().any(|x| x.objects.iter().any(|o| o.payload == want))).await;
+        got.extend(collect(s, Duration::from_secs(1), |_| false).await);
+        got.iter().flat_map(|g| g.objects.iter().map(|o| o.payload.clone())).collect()
+    }
+
+    /// Every session closed: the refusals its summaries count, by reason.
+    async fn close(self) -> Vec<(String, u64)> {
+        for x in &self.clients {
+            x.close(0, "done");
+        }
+        let relay = self.server.relay().clone();
+        assert!(until(STEP, || relay.summaries().len() >= 3).await, "every session's summary");
+        let mut refused: Vec<(String, u64)> = relay.summaries().iter().flat_map(|x| x.summary.tally.publisher.co_refused.clone()).collect();
+        refused.sort();
+        self.server.shutdown().await;
+        refused
+    }
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn the_relay_counts_inputs_in_chain_order() {
+    timeout(Duration::from_secs(120), chain_order()).await.expect("the test finishes in time");
+}
+
+async fn chain_order() {
+    // A hold past any wait of this test, however loaded the host: the relay is to wait for seq 0.
+    let a = Audience::open([0x81; 32], 0x82, 60_000).await;
+    let f = a.server.relay().floor().unwrap();
+    let first = a.vote(0, None, 1);
+    let second = a.vote(1, Some(&first), 2);
+    // The member's two votes, each a group on a stream of its own, the second opened first: the
+    // relay holds seq 1 until seq 0 arrives.
+    send(&a.clients[1], 2, 1, &msf(&second)).await;
+    let held = until(STEP, || f.waiting(&a.room) == 1).await;
+    assert_eq!(f.refused_inputs(), 0);
+    send(&a.clients[1], 2, 0, &msf(&first)).await;
+    // Seq 0 is counted first and forwarded; seq 1, counted after it, is past the member's power.
+    // (A relay that counted them as they came would have forwarded seq 1 and refused seq 0.)
+    let got = a.received(&msf(&first)).await;
+    assert!(until(STEP, || f.refused_inputs() >= 1).await, "the relay refuses one of them");
+    assert_eq!(got, vec![msf(&first)], "seq 0 forwarded, seq 1 withheld");
+    assert!(held, "the relay held seq 1 for seq 0");
+    assert_eq!((f.refused_inputs(), f.waiting(&a.room)), (1, 0));
+    assert_eq!(a.close().await, vec![("input_over_limit".to_string(), 1)]);
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn the_relay_refuses_an_input_whose_gap_never_fills() {
+    timeout(Duration::from_secs(120), gap_never_fills()).await.expect("the test finishes in time");
+}
+
+async fn gap_never_fills() {
+    let hold = Duration::from_millis(400);
+    let a = Audience::open([0x91; 32], 0x92, hold.as_millis() as u64).await;
+    let f = a.server.relay().floor().unwrap();
+    let first = a.vote(0, None, 1);
+    let second = a.vote(1, Some(&first), 2);
+    // Seq 1 alone, seq 0 held back by its sender: the relay refuses seq 1 once its hold runs out.
+    let sent = Instant::now();
+    send(&a.clients[1], 2, 1, &msf(&second)).await;
+    assert!(until(STEP, || f.refused_inputs() >= 1).await, "the relay refuses seq 1");
+    assert!(sent.elapsed() >= hold, "not before its hold ran out: {:?}", sent.elapsed());
+    assert_eq!(f.waiting(&a.room), 0);
+    // Seq 0, sent now, is forwarded: no input after it was.
+    send(&a.clients[1], 2, 0, &msf(&first)).await;
+    let got = a.received(&msf(&first)).await;
+    assert_eq!(got, vec![msf(&first)], "seq 1 withheld, seq 0 forwarded");
+    assert_eq!(f.refused_inputs(), 1);
+    assert_eq!(a.close().await, vec![("input_out_of_order".to_string(), 1)]);
+}

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")
 }
 

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

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