From 233f50195ff0facd57fd41f0c0023c9e59628452 Mon Sep 17 00:00:00 2001 From: osmaczko <33099791+osmaczko@users.noreply.github.com> Date: Fri, 21 Aug 2026 08:43:59 +0200 Subject: [PATCH] test: cover GroupV2 group growth against forks A forked group keeps reporting a healthy roster, so a growth test that compares member counts walks straight past the split. What the members on either branch cannot do is read each other's posts. The tests grow a group the way people do, one member at a time and five at a time, and after every add require that every joined member reports the same roster and reads what the others post. The harness can now be told to keep a client running when it rejects an inbound payload, the way a production client does with `Event::InboundError`, so a split group reports who ended up on which branch instead of stopping at the first rejection. --- .../integration_tests_core/src/test_client.rs | 47 ++- core/integration_tests_core/tests/README.md | 63 ++++ .../tests/test_group_v2_scale.rs | 289 ++++++++++++++++++ 3 files changed, 389 insertions(+), 10 deletions(-) create mode 100644 core/integration_tests_core/tests/README.md create mode 100644 core/integration_tests_core/tests/test_group_v2_scale.rs diff --git a/core/integration_tests_core/src/test_client.rs b/core/integration_tests_core/src/test_client.rs index 2c5a0e1..8a57946 100644 --- a/core/integration_tests_core/src/test_client.rs +++ b/core/integration_tests_core/src/test_client.rs @@ -6,7 +6,7 @@ use std::collections::HashMap; use std::fmt::Debug; use std::ops::{Deref, DerefMut}; use std::time::Duration; -use tracing::{info, warn}; +use tracing::{debug, info, warn}; use components::{EphemeralRegistry, LocalBroadcaster, MemStore}; @@ -34,6 +34,8 @@ pub struct ReceivedMessage { pub struct TestClient { inner: ClientType, received_messages: Vec>>, + inbound_errors: Vec, + tolerate_inbound_errors: bool, } impl TestClient { @@ -41,6 +43,8 @@ impl TestClient { Self { inner: client, received_messages: vec![], + inbound_errors: vec![], + tolerate_inbound_errors: false, } } @@ -48,6 +52,12 @@ impl TestClient { self.inner.ident_id().clone() } + /// Inbound payloads this client rejected, in arrival order. Only recorded + /// once the harness tolerates them. + pub fn inbound_errors(&self) -> &[String] { + &self.inbound_errors + } + fn drain_outcomes(&mut self) -> Vec { let mut messages = vec![]; while let Some(data) = self.inner.ds().poll() { @@ -56,7 +66,15 @@ impl TestClient { let mut outcomes = vec![]; for data in messages { - let outcome = self.inner.handle_payload(&data).unwrap(); + let outcome = match self.inner.handle_payload(&data) { + Ok(outcome) => outcome, + Err(e) if self.tolerate_inbound_errors => { + warn!(id = ?self.ident_id(), error = ?e, "INBOUND ERROR"); + self.inbound_errors.push(format!("{e:?}")); + continue; + } + Err(e) => panic!("{:?} rejected an inbound payload: {e:?}", self.ident_id()), + }; warn!(id= ?self.ident_id(),?outcome, "DRAIN CLIENT"); // Copy Convo Messages to received buffer @@ -142,7 +160,7 @@ pub struct TestHarness { impl TestHarness { pub fn new(cb: impl Fn(&TestClient, PayloadOutcome) + 'static) -> Self { const { assert!(N > 0, "TestHarness requires at least one client") }; - const { assert!(N <= 4, "Only 4 clients are supported(Soft Limit") }; + const { assert!(N <= 64, "TestHarness supports at most 64 clients") }; let mut clients = vec![]; let mut addresses = HashMap::new(); @@ -167,7 +185,7 @@ impl TestHarness { clients.push(client); } - dbg!(&rs); + debug!(?rs, "registry"); Self { addresses, @@ -186,13 +204,22 @@ impl TestHarness { &mut self.clients[i] } - fn names(i: usize) -> &'static str { + /// Lets a client keep running when it rejects an inbound payload, the way + /// a production client does with `Event::InboundError`, and records what + /// it rejected. Without this a rejection fails the test where it happens. + pub fn tolerate_inbound_errors(&mut self) { + for client in &mut self.clients { + client.tolerate_inbound_errors = true; + } + } + + fn names(i: usize) -> String { match i { - SARO => "saro", - RAYA => "raya", - PAX => "pax", - MIRA => "mira", - _ => "unnamed", + SARO => "saro".into(), + RAYA => "raya".into(), + PAX => "pax".into(), + MIRA => "mira".into(), + n => format!("m{n:02}"), } } diff --git a/core/integration_tests_core/tests/README.md b/core/integration_tests_core/tests/README.md new file mode 100644 index 0000000..9d3a677 --- /dev/null +++ b/core/integration_tests_core/tests/README.md @@ -0,0 +1,63 @@ +# GroupV2 scale tests + +`test_group_v2_scale.rs` grows a GroupV2 group over the loss-free in-process broadcaster and, after +every add, requires that every joined member reports the same roster **and** can read what the +others post. The exchange is the check that catches a fork: on the de-mls commit before the fix +(see below), the one-at-a-time test reaches six members that all report the same six-member roster +while one of them can no longer decrypt what the others send, so a roster comparison alone would +have called that group converged. Covers libchat#199. + +| test | group | adds | +|---|---|---| +| `groupv2_grows_one_member_at_a_time` | 12 members | one per add | +| `groupv2_grows_in_batches` | 26 members | five per add | + +The clock is virtual, and the two together take well under a minute. + +## Run them + +```sh +# both (needs protoc, as the rest of the workspace does: apt-get install protobuf-compiler) +cargo test -p integration_tests_core --test test_group_v2_scale + +# one of them, with the tracing feed on +LOG=info cargo test -p integration_tests_core --test test_group_v2_scale groupv2_grows_in_batches -- --nocapture +``` + +## Reading a failure + +``` +the group of 6 is no longer one group: members [4] never read the post from member 0 :: +rosters(size -> clients) {6: 6} not_joined 6 distinct_rosters 1 creator_pending 0 +rejected_payloads 16 first member 4: DeMlsError(Mls(ProcessMessage(ValidationError(UnableToDecrypt(AeadError))))) +``` + +- `rosters` maps a member count to the number of clients reporting it, and `not_joined` counts the + clients the test has not added yet. `distinct_rosters` counts how many different rosters those + clients hold, so `1` means they all agree on the membership and the split is in the key material + alone. +- `creator_pending` is the invites the creator still has awaiting a commit. +- `rejected_payloads` counts what the clients refused to process, and `first` quotes the earliest + one held by the lowest-numbered client. `UnableToDecrypt` is the signature of a fork: the payload + is well formed, it just belongs to another branch of the group. + +## Watching the bug they cover + +Point de-mls at the commit before the fix in `core/conversations/Cargo.toml`: + +```toml +de-mls = { git = "https://github.com/vacp2p/de-mls", rev = "5cfce1b97305363466c0e68668fcd85cad4b8996" } +``` + +Both tests then fail within seconds, on the first add that follows a voted steward election: at six +members when they are added one at a time, at sixteen when they are added five at a time. + +## Local overrides + +Timing comes from constants at the top of the file; group size is the const generic on `run` and +batch size its argument. Two environment variables cover what a local run usually needs to change: + +| var | default | meaning | +|---|---|---| +| `BUDGET` | 30 | virtual seconds a settle gets before the group is called split | +| `LOG` | off | `warn`, `info` or `debug` turns the tracing feed on | diff --git a/core/integration_tests_core/tests/test_group_v2_scale.rs b/core/integration_tests_core/tests/test_group_v2_scale.rs new file mode 100644 index 0000000..db25e7c --- /dev/null +++ b/core/integration_tests_core/tests/test_group_v2_scale.rs @@ -0,0 +1,289 @@ +//! GroupV2 group growth (regression for libchat#199). +//! +//! One creator grows a group over the loss-free in-process broadcaster, and +//! after every add the group has to converge twice over: every joined member +//! reports the same roster, and every joined member can still read what the +//! others post. The second check is the one that catches a fork, because two +//! branches of a split group can carry the same members while sharing no key +//! material. + +use integration_tests_core::TestHarness; +use shared_traits::IdentId; +use std::collections::{BTreeMap, BTreeSet}; +use std::time::Duration; + +/// Granularity the virtual clock advances in, and the steps between two checks +/// of a settle condition. Checking queries every client, which costs far more +/// than a step does. +const STEP: Duration = Duration::from_millis(50); +const STEPS_PER_CHECK: usize = 10; + +/// Virtual seconds a settle gets before the group is called split. `BUDGET` +/// raises it for a local run, and `LOG=warn|info|debug` turns on the tracing +/// feed, which is off by default because the harness traces every payload. +fn settle_budget() -> Duration { + let seconds = std::env::var("BUDGET") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(30); + Duration::from_secs(seconds) +} + +fn init_tracing() { + let level = match std::env::var("LOG").as_deref() { + Ok("debug") => tracing::Level::DEBUG, + Ok("info") => tracing::Level::INFO, + Ok("warn") => tracing::Level::WARN, + _ => tracing::Level::ERROR, + }; + let _ = tracing_subscriber::fmt() + .with_max_level(level) + .with_test_writer() + .try_init(); +} + +/// The members each client reports, sorted, `None` while it has not joined. +fn rosters(h: &mut TestHarness, convo: &str) -> Vec>>> { + (0..N) + .map(|i| { + h.client_mut(i).group_members(convo).ok().map(|mut m| { + m.sort(); + m + }) + }) + .collect() +} + +/// Every member added so far reports the same roster, and it holds all of them. +fn rosters_agree(h: &mut TestHarness, convo: &str, joined: usize) -> bool { + let rosters = rosters(h, convo); + let Some(creator) = rosters[0].as_ref() else { + return false; + }; + creator.len() == joined + && rosters[..joined] + .iter() + .all(|r| r.as_ref() == Some(creator)) +} + +/// Advances the virtual clock until `ready` holds or the budget runs out. +/// [`TestHarness::process_until`] panics instead of reporting, and a settle +/// that runs out here has to come back with the group's state attached. +fn settle( + h: &mut TestHarness, + ready: impl Fn(&mut TestHarness) -> bool, +) -> bool { + let mut elapsed = Duration::ZERO; + let budget = settle_budget(); + while elapsed < budget { + if ready(h) { + return true; + } + for _ in 0..STEPS_PER_CHECK { + h.process(STEP); + elapsed += STEP; + } + } + // A step drains payloads before it advances the clock, so whatever the + // last wakeups produced is still in flight; one more drain gives the + // verdict everything the run has already generated. + h.process(Duration::ZERO); + ready(h) +} + +/// Adds members, retrying while an election in flight has the conversation +/// blocked, and reporting the last refusal when the budget runs out. +fn add_members( + h: &mut TestHarness, + convo: &str, + invited: &[&IdentId], +) -> Result<(), String> { + let mut elapsed = Duration::ZERO; + let budget = settle_budget(); + let mut refusal = String::new(); + while elapsed < budget { + match h.client_mut(0).group_add_member(convo, invited) { + Ok(()) => return Ok(()), + Err(e) => refusal = format!("{e:?}"), + } + for _ in 0..STEPS_PER_CHECK { + h.process(STEP); + elapsed += STEP; + } + } + Err(refusal) +} + +/// Joined members that have not read `content`, its sender aside. +fn unread( + h: &mut TestHarness, + convo: &str, + sender: usize, + joined: usize, + content: &[u8], +) -> Vec { + (0..joined) + .filter(|i| *i != sender && !h.client(*i).check(convo, content)) + .collect() +} + +/// Checks a post goes unread before it is posted again: an epoch that turns +/// between a post and its delivery leaves it undecryptable for members that +/// merged the commit, and only a fresh post reaches them. +const CHECKS_PER_POST: usize = 4; + +/// Posts a message and waits for every other joined member to read it. A +/// member stranded on a forked branch cannot read it whatever its roster says, +/// and a member whose conversation is frozen for an election cannot post at +/// all, so posting is retried for as long as the settle budget lasts. +fn exchange( + h: &mut TestHarness, + convo: &str, + sender: usize, + joined: usize, + content: &[u8], +) -> Result<(), String> { + let mut elapsed = Duration::ZERO; + let budget = settle_budget(); + let mut posts = 0; + let mut checks_since_post = 0; + let mut refusal = String::new(); + while elapsed < budget { + if posts > 0 { + if unread(h, convo, sender, joined, content).is_empty() { + return Ok(()); + } + checks_since_post += 1; + } + if posts == 0 || checks_since_post >= CHECKS_PER_POST { + match h.client_mut(sender).send_content(convo, content) { + Ok(_) => { + posts += 1; + checks_since_post = 0; + } + Err(e) => refusal = format!("{e:?}"), + } + } + for _ in 0..STEPS_PER_CHECK { + h.process(STEP); + elapsed += STEP; + } + } + h.process(Duration::ZERO); + if posts == 0 { + return Err(format!("member {sender} could not post: {refusal}")); + } + match unread(h, convo, sender, joined, content) { + missing if missing.is_empty() => Ok(()), + missing => Err(format!( + "members {missing:?} never read the post from member {sender}" + )), + } +} + +/// One line of failure diagnostics: how the group split, and whether anyone +/// rejected an inbound payload on the way. +fn report(h: &mut TestHarness, convo: &str) -> String { + let rosters = rosters(h, convo); + let mut sizes: BTreeMap = BTreeMap::new(); + let mut not_joined = 0; + for roster in &rosters { + match roster { + Some(members) => *sizes.entry(members.len()).or_default() += 1, + None => not_joined += 1, + } + } + let distinct: BTreeSet<_> = rosters.into_iter().flatten().collect(); + let pending = h + .client_mut(0) + .group_pending_members(convo) + .map_or("?".to_string(), |p| p.len().to_string()); + let rejections: usize = (0..N).map(|i| h.client(i).inbound_errors().len()).sum(); + let first = (0..N) + .find_map(|i| { + h.client(i) + .inbound_errors() + .first() + .map(|e| format!("member {i}: {e}")) + }) + .unwrap_or_else(|| "none".to_string()); + format!( + "rosters(size -> clients) {sizes:?} not_joined {not_joined} distinct_rosters {} creator_pending {pending} rejected_payloads {rejections} first {first}", + distinct.len() + ) +} + +fn run(batch: usize) { + init_tracing(); + + let mut harness = TestHarness::::new(|_, _| {}); + // A member on the far side of a fork rejects everything this side posts. + // Recording those instead of stopping at the first one lets the run reach + // the exchange check, which names the members that could not read. + harness.tolerate_inbound_errors(); + + let convo = harness + .client_mut(0) + .create_group_convo_v2(&[], "scale", "") + .expect("create group"); + + let mut joined = 1; + while joined < N { + let upto = (joined + batch).min(N); + let addresses: Vec = (joined..upto) + .map(|i| harness.client_mut(i).addr()) + .collect(); + let invited: Vec<&IdentId> = addresses.iter().collect(); + if let Err(refusal) = add_members(&mut harness, &convo, &invited) { + panic!( + "adding members {joined}..{upto} kept being refused: {refusal} :: {}", + report(&mut harness, &convo) + ); + } + + assert!( + settle(&mut harness, |h| rosters_agree(h, &convo, upto)), + "the group did not converge on {upto} members :: {}", + report(&mut harness, &convo) + ); + + for sender in [0, upto - 1] { + let content = format!("post from member {sender} at {upto} members"); + if let Err(split) = exchange(&mut harness, &convo, sender, upto, content.as_bytes()) { + panic!( + "the group of {upto} is no longer one group: {split} :: {}", + report(&mut harness, &convo) + ); + } + } + + joined = upto; + } + + // The clients tolerate a rejected payload so that a fork is reported by + // the exchange rather than by the first undecryptable frame. A group that + // converged all the way should have handed every client everything it was + // sent. + let mut rejected = Vec::new(); + for i in 0..N { + for error in harness.client(i).inbound_errors() { + rejected.push(format!("member {i}: {error}")); + } + } + assert!( + rejected.is_empty(), + "the group converged but clients rejected payloads on the way: {rejected:?}" + ); +} + +/// The flow a person follows: invite, let the group settle, invite the next. +#[test] +fn groupv2_grows_one_member_at_a_time() { + run::<12>(1); +} + +/// Growth past the twenty members the issue reports as the ceiling. +#[test] +fn groupv2_grows_in_batches() { + run::<26>(5); +}