fix: bump de-mls to the join-time fork fix, with a scale regression test

core/account/src/directory.rs

@@ -156,12 +156,12 @@ pub fn decode_bundle_payload(payload: &[u8]) -> Result<DecodedBundle, BundleErro
     if body.len() != count * 32 {
         return Err(BundleError::Short);
     }
-    let devices = body
-        .chunks_exact(32)
-        .map(|c| c.try_into().expect("chunks_exact(32) yields 32 bytes"))
-        .collect();
+    let (devices, _) = body.as_chunks::<32>();
 
-    Ok(DecodedBundle { lamport, devices })
+    Ok(DecodedBundle {
+        lamport,
+        devices: devices.to_vec(),
+    })
 }
 
 /// Decode `bundle`, confirm it belongs to `expected_account`, and verify the

core/conversations/Cargo.toml

@@ -19,7 +19,7 @@ storage = { workspace = true }
 alloy = "2.0"
 base64 = "0.22"
 chat-proto = { workspace = true }
-de-mls = { git = "https://github.com/vacp2p/de-mls", rev = "5cfce1b97305363466c0e68668fcd85cad4b8996" }
+de-mls = { git = "https://github.com/vacp2p/de-mls", rev = "2eef52fc134c934d4384c665eccc96112893e5b5" }
 double-ratchets = { path = "../double-ratchets" }
 hashgraph-like-consensus = "0.6.0"
 hex = "0.4.3"

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,20 +34,30 @@ pub struct ReceivedMessage<T> {
 pub struct TestClient {
     inner: ClientType,
     received_messages: Vec<ReceivedMessage<Vec<u8>>>,
+    inbound_errors: Vec<String>,
+    tolerate_inbound_errors: bool,
 }
 
 impl TestClient {
     fn init(client: ClientType) -> Self {
         Self {
             inner: client,
             received_messages: vec![],
+            inbound_errors: vec![],
+            tolerate_inbound_errors: false,
         }
     }
 
     pub fn addr(&self) -> IdentId {
         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<PayloadOutcome> {
         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<const N: usize> {
 impl<const N: usize> TestHarness<N> {
     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<const N: usize> TestHarness<N> {
             clients.push(client);
         }
 
-        dbg!(&rs);
+        debug!(?rs, "registry");
 
         Self {
             addresses,
@@ -186,13 +204,22 @@ impl<const N: usize> TestHarness<N> {
         &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}"),
         }
     }
 

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 |