test: cover a member that loses one commit candidate

Every steward mints a commit over the same batch and broadcasts it, and every member applies the best of the candidates it holds once its freeze window closes, with no retry and no minimum. Two commits over one batch are not interchangeable, since each carries its committer's key material, so a member that never receives one of them applies a different commit and its MLS state diverges from the group's for good: no layer retransmits the frame, the delivery node keeps no history to replay, and a ConversationSync carries the steward list rather than MLS state.

The test grows a group of four and loses one candidate on its way to one member, each member taking its turn. It is red on purpose: every client reports the same five members while the one that lost a candidate sits on its own branch, unable to read anything the group posts.

A test client can now be given an inbound filter, and the frames it rejects are discarded unread. It recognises a candidate the way its receiver does, so the de-mls pin moves to the workspace manifest for the test crate to share.

Cargo.lock

@@ -3143,10 +3143,13 @@ dependencies = [
 name = "integration_tests_core"
 version = "0.1.0"
 dependencies = [
+ "chat-proto",
  "chat-sqlite",
  "components",
  "crypto",
+ "de-mls",
  "libchat",
+ "prost",
  "shared-traits",
  "storage",
  "tempfile",

Cargo.toml

@@ -52,6 +52,7 @@ chat-proto = { git = "https://github.com/logos-messaging/chat_proto", rev = "948
 # External Workspace dependency declarations (sorted)
 blake2 = "0.10"
 crossbeam-channel = "0.5"
+de-mls = { git = "https://github.com/vacp2p/de-mls", rev = "2eef52fc134c934d4384c665eccc96112893e5b5" }
 
 # Panicking across FFI boundaries is UB; chat-cli registers Rust callbacks
 # that liblogosdelivery invokes, so abort instead of unwinding.

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 = "2eef52fc134c934d4384c665eccc96112893e5b5" }
+de-mls = { workspace = true }
 double-ratchets = { path = "../double-ratchets" }
 hashgraph-like-consensus = "0.6.0"
 hex = "0.4.3"

core/conversations/src/conversation.rs

@@ -10,7 +10,7 @@ use crate::service_context::{ExternalServices, ServiceContext};
 use crate::types::ConvoMetadata;
 pub use direct_v1::DirectV1Convo;
 pub use group_v1::GroupV1Convo;
-pub use group_v2::{GroupV2Clock, GroupV2Convo};
+pub use group_v2::{GroupV2Clock, GroupV2Convo, GroupV2Frame, GroupV2Payload};
 use shared_traits::IdentIdRef;
 
 pub type ConversationId = String;

core/conversations/src/lib.rs

@@ -13,7 +13,7 @@ mod utils;
 pub use causal_history::{DeliveryAck, Frontier, MissingMessage};
 pub use chat_sqlite::ChatStorage;
 pub use chat_sqlite::StorageConfig;
-pub use conversation::{GroupV2Clock, MessageId};
+pub use conversation::{GroupV2Clock, GroupV2Frame, GroupV2Payload, MessageId};
 pub use core::{ConversationId, Core};
 /// Timing/policy for GroupV2 conversations (de-mls's per-conversation config).
 /// Defaults to the de-mls library defaults; inject via

core/integration_tests_core/Cargo.toml

@@ -19,10 +19,13 @@ shared-traits = { workspace = true }
 tracing = "0.1"
 
 [dev-dependencies]
+chat-proto = { workspace = true }
 chat-sqlite = { workspace = true }
+de-mls = { workspace = true }
 storage = { workspace = true }
 
 # External dependencies (sorted)
+prost = "0.14.1"
 tempfile = "3"
 tracing = "0.1.44"
 tracing-subscriber = "0.3"

core/integration_tests_core/src/test_client.rs

@@ -13,6 +13,7 @@ use components::{EphemeralRegistry, LocalBroadcaster, MemStore};
 use crate::wakeup::{TestWakeupProvider, TestWakeupService, WakeupRecord};
 
 type OnMessageCallback = dyn Fn(&TestClient, PayloadOutcome);
+type InboundFilter = dyn FnMut(&[u8]) -> bool;
 
 type WS = TestWakeupService;
 type WP = TestWakeupProvider;
@@ -36,6 +37,8 @@ pub struct TestClient {
     received_messages: Vec<ReceivedMessage<Vec<u8>>>,
     inbound_errors: Vec<String>,
     tolerate_inbound_errors: bool,
+    inbound_filter: Option<Box<InboundFilter>>,
+    dropped: usize,
 }
 
 impl TestClient {
@@ -45,6 +48,8 @@ impl TestClient {
             received_messages: vec![],
             inbound_errors: vec![],
             tolerate_inbound_errors: false,
+            inbound_filter: None,
+            dropped: 0,
         }
     }
 
@@ -58,9 +63,28 @@ impl TestClient {
         &self.inbound_errors
     }
 
+    /// Frames this client never receives: `filter` sees every payload the
+    /// transport hands over, and the ones it rejects are discarded unread.
+    /// Nothing in the stack retransmits them, so a rejected frame is one this
+    /// client never learns of.
+    pub fn set_inbound_filter(&mut self, filter: impl FnMut(&[u8]) -> bool + 'static) {
+        self.inbound_filter = Some(Box::new(filter));
+    }
+
+    /// How many inbound frames the filter has dropped.
+    pub fn dropped(&self) -> usize {
+        self.dropped
+    }
+
     fn drain_outcomes(&mut self) -> Vec<PayloadOutcome> {
         let mut messages = vec![];
         while let Some(data) = self.inner.ds().poll() {
+            if let Some(filter) = self.inbound_filter.as_mut()
+                && !filter(&data)
+            {
+                self.dropped += 1;
+                continue;
+            }
             messages.push(data);
         }
 

core/integration_tests_core/tests/README.md

@@ -11,13 +11,14 @@ have called that group converged. Covers libchat#199.
 |---|---|---|
 | `groupv2_grows_one_member_at_a_time` | 12 members | one per add |
 | `groupv2_grows_in_batches` | 26 members | five per add |
+| `groupv2_survives_a_lost_commit_candidate` | 5 members | one per add, one commit candidate lost on the last |
 
-The clock is virtual, and the two together take well under a minute.
+The clock is virtual, and the three together take well under a minute.
 
 ## Run them
 
 ```sh
-# both (needs protoc, as the rest of the workspace does: apt-get install protobuf-compiler)
+# all three (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
@@ -41,9 +42,36 @@ rejected_payloads 16 first member 4: DeMlsError(Mls(ProcessMessage(ValidationErr
   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
+## The lost-candidate test is red on purpose
 
-Point de-mls at the commit before the fix in `core/conversations/Cargo.toml`:
+`groupv2_survives_a_lost_commit_candidate` grows a group of four and loses one commit candidate on
+its way to one member, in the round that adds the fifth.
+
+Both stewards mint a commit over the same batch and broadcast it, and every member applies the best
+of the candidates it holds once its freeze window closes, with no retry and no minimum. Two commits
+over the same batch are not interchangeable, since each carries its committer's key material, so
+the member that receives one of the two applies a different commit and its MLS state diverges from
+the group's. Nothing brings it back: no layer retransmits the frame, the delivery node keeps no
+history to replay, and a `ConversationSync` carries the steward list rather than MLS state.
+
+The frame is singled out by `TestClient::set_inbound_filter`, which discards it unread, and it is
+recognised the way its receiver recognises it: a GroupV2 frame wrapping a plaintext de-mls
+`AppMessage` that carries a `CommitCandidate`. Each member takes its turn short a candidate, since
+a steward is left holding the one it minted itself and commits a round nobody else has, where a
+plain member holds the one that reached it.
+
+```
+member 0: the group of 5 is no longer one group: members [1, 2, 3, 4] never read the post from member 0 ::
+rosters(size -> clients) {5: 5} not_joined 0 distinct_rosters 1 creator_pending 0
+rejected_payloads 74 first member 0: DeMlsError(Mls(ProcessMessage(ValidationError(UnableToDecrypt(AeadError)))))
+```
+
+Every client reports the same five members, and the one that lost a candidate is alone on its own
+branch of the group.
+
+## Watching the bug the growth tests cover
+
+Point de-mls at the commit before the fix in the workspace `Cargo.toml`:
 
 ```toml
 de-mls = { git = "https://github.com/vacp2p/de-mls", rev = "5cfce1b97305363466c0e68668fcd85cad4b8996" }

core/integration_tests_core/tests/test_group_v2_scale.rs

@@ -6,8 +6,16 @@
 //! 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.
+//!
+//! The last test grows the same group with one commit candidate lost in
+//! flight, which is what the broadcaster's loss-free delivery otherwise hides.
 
+use chat_proto::logoschat::encryption::{EncryptedPayload, encrypted_payload};
+use chat_proto::logoschat::envelope::EnvelopeV1;
+use de_mls::protos::de_mls::messages::v1::{AppMessage, app_message};
 use integration_tests_core::TestHarness;
+use libchat::{GroupV2Frame, GroupV2Payload};
+use prost::Message;
 use shared_traits::IdentId;
 use std::collections::{BTreeMap, BTreeSet};
 use std::time::Duration;
@@ -287,3 +295,129 @@ fn groupv2_grows_one_member_at_a_time() {
 fn groupv2_grows_in_batches() {
     run::<26>(5);
 }
+
+/// Members the group has before the last add. It elects two stewards
+/// (`sn_max` is 2), so the round that follows has two commit candidates in
+/// flight: a plain member sees both on the wire, and a steward, minting its
+/// own, sees one.
+const GROUP: usize = 4;
+
+/// Grows a group of [`GROUP`] members and loses one commit candidate on its way
+/// to `receiver`, in the round that adds the last member.
+///
+/// Every steward mints a commit over the same batch and broadcasts it, and
+/// every member applies the best of the candidates it holds once its freeze
+/// window closes. Two commits over the same batch are not interchangeable,
+/// since each carries its committer's key material, so a member that selects
+/// from a strict subset applies a different commit and its MLS state diverges
+/// from the group's. Nothing brings it back: no layer retransmits the frame,
+/// the delivery node keeps no history to replay, and a `ConversationSync`
+/// carries the steward list rather than MLS state.
+fn lose_a_commit_candidate(receiver: usize) {
+    init_tracing();
+    const N: usize = GROUP + 1;
+
+    let mut harness = TestHarness::<N>::new(|_, _| {});
+    harness.tolerate_inbound_errors();
+
+    let convo = harness
+        .client_mut(0)
+        .create_group_convo_v2(&[], "lost-candidate", "")
+        .expect("create group");
+
+    for joined in 1..GROUP {
+        let next = harness.client_mut(joined).addr();
+        if let Err(refusal) = add_members(&mut harness, &convo, &[&next]) {
+            panic!("adding member {joined} kept being refused: {refusal}");
+        }
+        assert!(
+            settle(&mut harness, |h| rosters_agree(h, &convo, joined + 1)),
+            "the group did not converge on {} members :: {}",
+            joined + 1,
+            report(&mut harness, &convo)
+        );
+    }
+
+    harness
+        .client_mut(receiver)
+        .set_inbound_filter(drop_first_commit_candidate());
+
+    let last = harness.client_mut(N - 1).addr();
+    if let Err(refusal) = add_members(&mut harness, &convo, &[&last]) {
+        panic!("member {receiver}: adding the last member kept being refused: {refusal}");
+    }
+
+    assert!(
+        settle(&mut harness, |h| rosters_agree(h, &convo, N)),
+        "member {receiver}: the group did not converge on {N} members :: {}",
+        report(&mut harness, &convo)
+    );
+
+    // Guards the setup rather than the behaviour: a round that delivered every
+    // candidate would let the run pass for the wrong reason.
+    assert_eq!(
+        harness.client(receiver).dropped(),
+        1,
+        "member {receiver} lost no commit candidate"
+    );
+
+    for sender in [0, receiver, N - 1] {
+        let content = format!("post from member {sender}, member {receiver} short a candidate");
+        if let Err(split) = exchange(&mut harness, &convo, sender, N, content.as_bytes()) {
+            panic!(
+                "member {receiver}: the group of {N} is no longer one group: {split} :: {}",
+                report(&mut harness, &convo)
+            );
+        }
+    }
+}
+
+/// Drops the first commit candidate the client it filters would receive, and
+/// passes every other frame.
+fn drop_first_commit_candidate() -> impl FnMut(&[u8]) -> bool {
+    let mut lost = false;
+    move |payload| {
+        if lost || !is_commit_candidate(payload) {
+            return true;
+        }
+        lost = true;
+        false
+    }
+}
+
+/// Whether a frame carries a commit candidate, read the way its receiver reads
+/// it: a GroupV2 frame wrapping a plaintext de-mls `AppMessage`.
+fn is_commit_candidate(payload: &[u8]) -> bool {
+    let Ok(envelope) = EnvelopeV1::decode(payload) else {
+        return false;
+    };
+    let Ok(encrypted) = EncryptedPayload::decode(envelope.payload.as_ref()) else {
+        return false;
+    };
+    let Some(encrypted_payload::Encryption::Plaintext(plaintext)) = encrypted.encryption else {
+        return false;
+    };
+    let Ok(frame) = GroupV2Frame::decode(plaintext.payload.as_ref()) else {
+        return false;
+    };
+    let Some(GroupV2Payload::DeMlsWrapper(inner)) = frame.payload else {
+        return false;
+    };
+    match AppMessage::decode(inner.as_ref()) {
+        Ok(message) => matches!(
+            message.payload,
+            Some(app_message::Payload::CommitCandidate(_))
+        ),
+        Err(_) => false,
+    }
+}
+
+/// Every member in turn, because a steward that loses the other steward's
+/// candidate holds one it minted itself and commits a round nobody else has,
+/// where a plain member holds the one candidate that reached it.
+#[test]
+fn groupv2_survives_a_lost_commit_candidate() {
+    for receiver in 0..GROUP {
+        lose_a_commit_candidate(receiver);
+    }
+}