From 519b9982ca4244e5797db82efcbe18e13fb62b4c Mon Sep 17 00:00:00 2001 From: osmaczko <33099791+osmaczko@users.noreply.github.com> Date: Fri, 3 Jul 2026 20:21:27 +0200 Subject: [PATCH 1/9] feat: expose GroupV2 through the threaded client GroupV2 (de-mls) conversations were reachable only from Core, and every conversation ran the hardcoded millisecond timer profile in group_v2.rs (20-150 ms freeze/consensus windows), which cannot survive real network latency. Make them reachable through ChatClient (as DirectV1 already is), with timing that holds up over a real network. - ChatClient::create_group_conversation(accounts) and add_group_members(convo_id, accounts): resolve each account address to its endorsed signer ids through the client-held directory, the same resolution create_direct_conversation uses, and drive Core::create_group_convo / group_add_member. - GroupV2 timing/policy is injectable: ServiceContext carries a de_mls::ConversationConfig (re-exported as GroupV2Config) defaulting to the de-mls library defaults; Core::set_group_v2_config and the builder's group_v2_config setter override it. The creator's phase durations reach joiners inside the welcome's ConversationSync, so the group runs the creator's phase timing. - GroupV2Convo::add_member validates every member's key package before proposing any add, skips members de-mls would silently not propose (self, already in the group) instead of stranding a pending invite, and flushes opened proposals even on a mid-batch failure, so a failed batch cannot invite members behind the caller's back. - The millisecond test profile moves into the test harnesses (integration_tests_core's TestHarness, crates/client/tests/group_v2.rs). - New client-level tests: three accounts on the in-process transport create a group, a non-creator adds the third member, and messages fan out with directory-verified senders; a batch containing a member with no key package fails without inviting anyone. --- .../src/conversation/group_v2.rs | 71 +++--- core/conversations/src/core.rs | 9 +- core/conversations/src/lib.rs | 7 + core/conversations/src/service_context.rs | 6 + .../integration_tests_core/src/test_client.rs | 20 +- crates/generic-chat/src/builder.rs | 26 +- crates/generic-chat/src/client.rs | 57 ++++- crates/generic-chat/src/lib.rs | 2 +- crates/generic-chat/tests/group_v2.rs | 237 ++++++++++++++++++ 9 files changed, 394 insertions(+), 41 deletions(-) create mode 100644 crates/generic-chat/tests/group_v2.rs diff --git a/core/conversations/src/conversation/group_v2.rs b/core/conversations/src/conversation/group_v2.rs index 03cd8b5..4be46f0 100644 --- a/core/conversations/src/conversation/group_v2.rs +++ b/core/conversations/src/conversation/group_v2.rs @@ -11,8 +11,7 @@ use de_mls::protos::de_mls::messages::v1::{ AppMessage as AppMessageProto, MemberWelcome, app_message, }; use de_mls::{ - Conversation, ConversationConfig, ConversationEvent, PeerScoringService, ScoringConfig, - default_score_deltas, + Conversation, ConversationEvent, PeerScoringService, ScoringConfig, default_score_deltas, defaults::{DefaultConsensusPlugin, DefaultPeerScoring, InMemoryPeerScoreStorage}, }; use hashgraph_like_consensus::signing::EthereumConsensusSigner; @@ -22,7 +21,6 @@ use openmls::prelude::{KeyPackageIn, OpenMlsProvider as _, ProtocolVersion}; use prost::Message; use shared_traits::{IdentId, IdentIdRef}; use std::sync::Arc; -use std::time::Duration; use tracing::{info, instrument, warn}; use crate::IdentityProvider; @@ -58,21 +56,6 @@ fn make_consensus() -> DefaultConsensusPlugin { DefaultConsensusPlugin::new(EthereumConsensusSigner::new(PrivateKeySigner::random())) } -/// TEST-ONLY millisecond timers. de-mls deadlines are real wall-clock, so the -/// default 60s timers never fire under fast virtual time. Production needs a -/// real config injected from the caller, not these hardcoded values. -fn demls_config() -> ConversationConfig { - ConversationConfig { - commit_inactivity_duration: Duration::from_millis(50), - freeze_duration: Duration::from_millis(20), - voting_delay: Duration::from_millis(30), - election_voting_delay: Duration::from_millis(30), - consensus_timeout: Duration::from_millis(150), - proposal_expiration: Duration::from_millis(2000), - ..ConversationConfig::default() - } -} - pub struct GroupV2Convo { convo_id: String, conversation: Conversation, @@ -116,7 +99,7 @@ impl GroupV2Convo { &make_consensus(), make_scoring(), rand_app_id(), - demls_config(), + service_ctx.group_v2_config.clone(), )?; let convo = GroupV2Convo { convo_id, @@ -147,7 +130,7 @@ impl GroupV2Convo { &make_consensus(), make_scoring(), rand_app_id(), - demls_config(), + service_ctx.group_v2_config.clone(), )? else { return Err(ChatError::generic("welcome not addressed to this member")); @@ -277,13 +260,14 @@ where service_ctx: &mut ServiceContext, members: &[IdentIdRef], ) -> Result<(), ChatError> { - // Record who WE invited before touching the conversation: after_op - // forwards a welcome only to joiners in pending_invites. Members are - // signer ids; the de-mls member id must match the id of the - // IdentityProvider that generated the key package (its MLS leaf - // credential content — de-mls matches members by credential), so it is - // read from the fetched key package rather than assumed equal to the - // signer id. + // Fetch and validate every key package before proposing any add, so a + // member with no key package fails the call before it opens proposals + // for the others. Members are signer ids; the de-mls member id must + // match the id of the IdentityProvider that generated the key package + // (its MLS leaf credential content — de-mls matches members by + // credential), so it is read from the fetched key package rather than + // assumed equal to the signer id. + let mut invites = Vec::with_capacity(members.len()); for member in members { let kp_bytes = service_ctx .registry @@ -298,17 +282,38 @@ where .credential() .serialized_content() .to_vec(); - self.pending_invites - .push((member_id.clone(), member.to_string())); - self.conversation.add_member( + invites.push((member_id, member.to_string(), kp_bytes)); + } + + // Record who WE invited before touching the conversation: after_op + // forwards a welcome only to joiners in pending_invites. Skip a member + // de-mls would silently not propose (self, or already in the group) — + // recording it would strand a pending invite that later matches a + // re-join and fires a duplicate welcome. + let mut result = Ok(()); + for (member_id, signer_id, kp_bytes) in invites { + if member_id == self.conversation.member_id_bytes() + || self.conversation.members()?.contains(&member_id) + { + continue; + } + self.pending_invites.push((member_id.clone(), signer_id)); + if let Err(e) = self.conversation.add_member( &service_ctx.mls_provider, &service_ctx.mls_identity, &member_id, &kp_bytes, - )?; + ) { + self.pending_invites.pop(); + result = Err(e.into()); + break; + } } - self.after_op(service_ctx)?; - Ok(()) + // Flush even on a mid-loop failure: proposals already opened must be + // published and the wakeup re-armed, or they sit dormant until an + // unrelated frame drives the conversation. + let flushed = self.after_op(service_ctx).map(drop); + result.and(flushed) } // fn conversation_state(&self) -> Result { diff --git a/core/conversations/src/core.rs b/core/conversations/src/core.rs index ca1c19b..25e6fb2 100644 --- a/core/conversations/src/core.rs +++ b/core/conversations/src/core.rs @@ -3,7 +3,7 @@ use crate::conversation::{ ConversationIdRef, DirectV1Convo, GroupV1Convo, GroupV2Convo, Identified, PrivateV1Convo, }; use crate::service_context::{ExternalServices, ServiceContext}; -use crate::{DeliveryService, IdentityProvider, RegistrationService, WakeupService}; +use crate::{DeliveryService, GroupV2Config, IdentityProvider, RegistrationService, WakeupService}; use crate::{ conversation::{Convo, GroupConvo}, errors::ChatError, @@ -140,6 +140,7 @@ where causal, identity, wakeup_service, + group_v2_config: GroupV2Config::default(), }, inbox, pq_inbox, @@ -174,6 +175,12 @@ impl<'a, S: ExternalServices + 'static> Core { self.pq_inbox.register(&mut self.services) } + /// Timing/policy for GroupV2 conversations created or joined after this + /// call. Existing conversations keep the config they were built with. + pub fn set_group_v2_config(&mut self, config: GroupV2Config) { + self.services.group_v2_config = config; + } + pub fn installation_name(&self) -> &str { self.services.identity.get_name() } diff --git a/core/conversations/src/lib.rs b/core/conversations/src/lib.rs index 654630e..9eac46c 100644 --- a/core/conversations/src/lib.rs +++ b/core/conversations/src/lib.rs @@ -16,6 +16,13 @@ pub use causal_history::{Frontier, MissingMessage}; pub use chat_sqlite::ChatStorage; pub use chat_sqlite::StorageConfig; pub use core::{ConversationId, Core, Introduction}; +/// Timing/policy for GroupV2 conversations (de-mls's per-conversation config). +/// Defaults to the de-mls library defaults; inject via +/// [`Core::set_group_v2_config`]. The creator's phase durations (commit +/// inactivity, freeze, recovery, voting inactivity, proposal expiration, +/// consensus timeout) travel to joiners with the welcome and overwrite +/// theirs; vote delays and the policy fields stay local to each member. +pub use de_mls::ConversationConfig as GroupV2Config; pub use errors::ChatError; pub use outcomes::{ Content, ConversationClass, ConvoOutcome, InboxOutcome, NewConversation, PayloadOutcome, diff --git a/core/conversations/src/service_context.rs b/core/conversations/src/service_context.rs index 82cf1d3..3f7372d 100644 --- a/core/conversations/src/service_context.rs +++ b/core/conversations/src/service_context.rs @@ -44,6 +44,11 @@ pub(crate) struct ServiceContext { pub(crate) causal: CausalHistoryStore, pub(crate) identity: Identity, pub(crate) wakeup_service: S::WS, + /// Timing/policy applied to GroupV2 conversations created or joined by + /// this core. Read at conversation construction; a joiner's phase + /// durations are then overwritten by the creator's, carried with the + /// welcome (vote delays and policy fields stay local). + pub(crate) group_v2_config: de_mls::ConversationConfig, } #[cfg(test)] @@ -110,6 +115,7 @@ mod test_support { causal: CausalHistoryStore::new(), identity: Identity::new(name), wakeup_service: NoopWakeups {}, + group_v2_config: de_mls::ConversationConfig::default(), }) } } diff --git a/core/integration_tests_core/src/test_client.rs b/core/integration_tests_core/src/test_client.rs index 09c84ef..9b75852 100644 --- a/core/integration_tests_core/src/test_client.rs +++ b/core/integration_tests_core/src/test_client.rs @@ -1,5 +1,5 @@ use crate::test_ident::TestIdent; -use libchat::{ConversationId, Core, IdentityProvider, PayloadOutcome}; +use libchat::{ConversationId, Core, GroupV2Config, IdentityProvider, PayloadOutcome}; use shared_traits::IdentId; use std::collections::HashMap; use std::fmt::Debug; @@ -21,6 +21,21 @@ const RAYA: usize = 1; const PAX: usize = 2; const MIRA: usize = 3; +/// Millisecond GroupV2 timers for the harness. de-mls deadlines are real +/// wall-clock, so the library defaults (60s commit inactivity) would stall +/// `process_until`, which settles in 50ms steps. +fn fast_group_v2_config() -> GroupV2Config { + GroupV2Config { + commit_inactivity_duration: Duration::from_millis(50), + freeze_duration: Duration::from_millis(20), + voting_delay: Duration::from_millis(30), + election_voting_delay: Duration::from_millis(30), + consensus_timeout: Duration::from_millis(150), + proposal_expiration: Duration::from_millis(2000), + ..GroupV2Config::default() + } +} + // type ClientType = CoreClient; type ClientType = Core<(TestIdent, LocalBroadcaster, EphemeralRegistry, WP, MemStore)>; @@ -148,9 +163,10 @@ impl TestHarness { let ident = TestIdent::new(Self::names(i)); addresses.insert(i, ident.id().clone()); - let core_client = + let mut core_client = ClientType::new_with_name(ident, ds.clone(), rs.clone(), wp, MemStore::new()) .unwrap(); + core_client.set_group_v2_config(fast_group_v2_config()); let client = TestClient::init(core_client); diff --git a/crates/generic-chat/src/builder.rs b/crates/generic-chat/src/builder.rs index 6a66394..a1c8eac 100644 --- a/crates/generic-chat/src/builder.rs +++ b/crates/generic-chat/src/builder.rs @@ -1,6 +1,6 @@ use components::EphemeralRegistry; use crossbeam_channel::Receiver; -use libchat::{ChatError, ChatStorage, RegistrationService, StorageConfig}; +use libchat::{ChatError, ChatStorage, GroupV2Config, RegistrationService, StorageConfig}; use logos_account::AccountDirectory; use storage::ChatStore; @@ -20,6 +20,7 @@ pub struct ChatClientBuilder { transport: T, registration: R, storage: S, + group_v2: Option, } impl ChatClientBuilder { @@ -35,6 +36,7 @@ impl ChatClientBuilder { transport: Unset, registration: Unset, storage: Unset, + group_v2: None, } } } @@ -47,6 +49,7 @@ impl ChatClientBuilder { transport: self.transport, registration: self.registration, storage: self.storage, + group_v2: self.group_v2, } } @@ -57,6 +60,7 @@ impl ChatClientBuilder { transport, registration: self.registration, storage: self.storage, + group_v2: self.group_v2, } } @@ -67,6 +71,7 @@ impl ChatClientBuilder { transport: self.transport, registration, storage: self.storage, + group_v2: self.group_v2, } } @@ -77,6 +82,7 @@ impl ChatClientBuilder { transport: self.transport, registration: self.registration, storage, + group_v2: self.group_v2, } } @@ -91,8 +97,18 @@ impl ChatClientBuilder { transport: self.transport, registration: self.registration, storage, + group_v2: self.group_v2, } } + + /// Timing/policy for GroupV2 conversations this client creates or joins. + /// Defaults to the de-mls library defaults; the creator's phase durations + /// travel to joiners with the welcome and overwrite theirs (vote delays + /// and policy fields stay local). + pub fn group_v2_config(mut self, config: GroupV2Config) -> Self { + self.group_v2 = Some(config); + self + } } type Built = Result<(ChatClient, Receiver), ClientError>; @@ -111,6 +127,7 @@ where self.transport, self.registration, self.storage, + self.group_v2, ) } } @@ -124,6 +141,7 @@ impl ChatClientBuilder { self.transport, EphemeralRegistry::new(), ChatStorage::in_memory(), + self.group_v2, ) } } @@ -140,6 +158,7 @@ where self.transport, EphemeralRegistry::new(), ChatStorage::in_memory(), + self.group_v2, ) } } @@ -157,6 +176,7 @@ where self.transport, self.registration, ChatStorage::in_memory(), + self.group_v2, ) } } @@ -174,6 +194,7 @@ where self.transport, EphemeralRegistry::new(), self.storage, + self.group_v2, ) } } @@ -191,6 +212,7 @@ where self.transport, self.registration, ChatStorage::in_memory(), + self.group_v2, ) } } @@ -209,6 +231,7 @@ where self.transport, self.registration, self.storage, + self.group_v2, ) } } @@ -226,6 +249,7 @@ where self.transport, EphemeralRegistry::new(), self.storage, + self.group_v2, ) } } diff --git a/crates/generic-chat/src/client.rs b/crates/generic-chat/src/client.rs index d425cd3..dc173da 100644 --- a/crates/generic-chat/src/client.rs +++ b/crates/generic-chat/src/client.rs @@ -5,8 +5,8 @@ use components::{ThreadedWakeupService, WakeupEvent}; use crossbeam_channel::{Receiver, Sender, select}; use crypto::Ed25519VerifyingKey; use libchat::{ - ConversationId, ConvoOutcome, Core, DeliveryService, IdentId, IdentIdRef, InboxOutcome, - Introduction, PayloadOutcome, RegistrationService, + ConversationId, ConvoOutcome, Core, DeliveryService, GroupV2Config, IdentId, IdentIdRef, + InboxOutcome, Introduction, PayloadOutcome, RegistrationService, }; use logos_account::{AccountDirectory, resolve_device_ids}; use parking_lot::Mutex; @@ -73,6 +73,7 @@ where mut transport: T, reg: R, storage: S, + group_v2: Option, ) -> Result<(Self, Receiver), ClientError> { let inbound = transport.inbound(); @@ -80,7 +81,10 @@ where let wakeup_service = ThreadedWakeupService::new(wakeup_tx); let directory = reg.clone(); let ident = DelegateIdentity::new(ident, &account); - let core = Core::new_with_name(ident, transport, reg, wakeup_service, storage)?; + let mut core = Core::new_with_name(ident, transport, reg, wakeup_service, storage)?; + if let Some(config) = group_v2 { + core.set_group_v2_config(config); + } Ok(Self::spawn(core, directory, account, inbound, wakeup_rx)) } @@ -151,6 +155,40 @@ where .map_err(Into::into) } + /// Create a GroupV2 conversation with the given accounts' devices. Each + /// account resolves to the signer ids its directory bundle endorses; the + /// group invite goes to every one of them. An empty slice creates a group + /// with only this client, to grow via [`Self::add_group_members`]. + pub fn create_group_conversation( + &mut self, + accounts: &[AccountAddressRef], + ) -> Result { + let signers = self.signers_from_accounts(accounts)?; + let signer_refs: Vec = signers.iter().collect(); + + self.core + .lock() + .create_group_convo(&signer_refs) + .map_err(Into::into) + } + + /// Add accounts' devices to an existing group conversation. Membership + /// changes are agreed and committed by the group asynchronously; the + /// joiners' welcomes go out once the add commits. + pub fn add_group_members( + &mut self, + convo_id: &str, + accounts: &[AccountAddressRef], + ) -> Result<(), ClientError> { + let signers = self.signers_from_accounts(accounts)?; + let signer_refs: Vec = signers.iter().collect(); + + self.core + .lock() + .group_add_member(convo_id, &signer_refs) + .map_err(Into::into) + } + /// Parse intro bundle bytes and initiate a private conversation. Outbound /// envelopes are published by the core. Returns this side's conversation ID. /// @@ -193,6 +231,19 @@ where .map_err(|e| ClientError::AccountResolution(e.to_string()))?; Ok(device_ids.into_iter().map(IdentId::new).collect()) } + + /// Resolve each account to its signer ids and flatten them, failing on the + /// first unresolvable account. + fn signers_from_accounts( + &self, + accounts: &[AccountAddressRef], + ) -> Result, ClientError> { + let mut signers = Vec::new(); + for account in accounts { + signers.extend(self.signers_from_account(account)?); + } + Ok(signers) + } } impl Drop for ChatClient diff --git a/crates/generic-chat/src/lib.rs b/crates/generic-chat/src/lib.rs index 39ea635..2bc2c54 100644 --- a/crates/generic-chat/src/lib.rs +++ b/crates/generic-chat/src/lib.rs @@ -15,7 +15,7 @@ pub use event::{Event, MessageSender}; // Re-export types callers need to interact with ChatClient. pub use libchat::{ AddressedEnvelope, ChatStore, ConversationClass, ConversationId, DeliveryService, - IdentityProvider, RegistrationService, StorageConfig, + GroupV2Config, IdentityProvider, RegistrationService, StorageConfig, }; // The directory trait bounds ChatClient's registry parameter, so callers // writing code generic over ChatClient need it too. diff --git a/crates/generic-chat/tests/group_v2.rs b/crates/generic-chat/tests/group_v2.rs new file mode 100644 index 0000000..bfa2073 --- /dev/null +++ b/crates/generic-chat/tests/group_v2.rs @@ -0,0 +1,237 @@ +//! GroupV2 through the threaded client: three accounts on the in-process +//! transport, driven purely over the public `ChatClient` API and its event +//! channel. Group commits and welcomes are minted asynchronously by de-mls +//! (wakeup-driven), so assertions wait for events rather than expecting a +//! fixed sequence. + +use std::time::Duration; + +use components::EphemeralRegistry; +use crossbeam_channel::Receiver; +use libchat::ChatStorage; +use logos_account::TestLogosAccount; +use logos_generic_chat::{ + ChatClient, ChatClientBuilder, ConversationClass, DelegateSigner, Event, GroupV2Config, + InProcessDelivery, MessageBus, +}; + +/// Millisecond GroupV2 timers so the de-mls commit/consensus dance completes +/// in test time; the library defaults wait 60s before committing an add. +fn fast_group_v2_config() -> GroupV2Config { + GroupV2Config { + commit_inactivity_duration: Duration::from_millis(50), + freeze_duration: Duration::from_millis(20), + voting_delay: Duration::from_millis(30), + election_voting_delay: Duration::from_millis(30), + consensus_timeout: Duration::from_millis(150), + proposal_expiration: Duration::from_millis(2000), + ..GroupV2Config::default() + } +} + +type TestClient = ChatClient; + +/// A client for a fresh account: mints the account and a delegate, publishes +/// the endorsing bundle, and builds the client on the shared bus/registry with +/// the fast GroupV2 timers. Returns the account address peers invite by. +fn create_test_client( + message_bus: MessageBus, + mut reg: EphemeralRegistry, +) -> (TestClient, Receiver, String) { + let account = TestLogosAccount::new(); + let delegate = DelegateSigner::random(); + account + .add_delegate_signer(&mut reg, delegate.public_key()) + .unwrap(); + let (client, events) = ChatClientBuilder::new(account.address()) + .ident(delegate) + .transport(InProcessDelivery::new(message_bus)) + .registration(reg) + .group_v2_config(fast_group_v2_config()) + .build() + .expect("client create"); + let addr = client.addr().to_string(); + (client, events, addr) +} + +/// Wait until an event matching `f` arrives, skipping unrelated events (group +/// protocol traffic can interleave observations); panic after `timeout`. +fn wait_for_event(events: &Receiver, label: &str, timeout: Duration, mut f: F) -> T +where + F: FnMut(&Event) -> Option, +{ + let deadline = std::time::Instant::now() + timeout; + loop { + let remaining = deadline + .checked_duration_since(std::time::Instant::now()) + .unwrap_or_else(|| panic!("timed out waiting for {label}")); + match events.recv_timeout(remaining) { + Ok(event) => { + if let Some(out) = f(&event) { + return out; + } + } + Err(_) => panic!("timed out waiting for {label}"), + } + } +} + +/// Wait for the group conversation to start on a joiner and return its id. +fn wait_for_group_started(events: &Receiver, label: &str) -> String { + wait_for_event(events, label, Duration::from_secs(10), |e| match e { + Event::ConversationStarted { convo_id, class } => { + assert_eq!(*class, ConversationClass::Group); + Some(convo_id.to_string()) + } + _ => None, + }) +} + +/// Wait for `content` to arrive and return the sender's verified account. +fn wait_for_message(events: &Receiver, content: &[u8]) -> Option { + let label = format!("MessageReceived({})", String::from_utf8_lossy(content)); + wait_for_event(events, &label, Duration::from_secs(10), |e| match e { + Event::MessageReceived { + content: got, + sender, + .. + } if got == content => Some(sender.account.as_ref().map(|a| a.as_str().to_string())), + _ => None, + }) +} + +/// A three-account group: saro creates it with raya, raya (a non-creator) +/// adds pax, and a message from each member reaches both others with a +/// directory-verified sender account. +#[test] +fn group_v2_three_members() { + let bus = MessageBus::default(); + let reg = EphemeralRegistry::new(); + + let (mut saro, saro_events, saro_addr) = create_test_client(bus.clone(), reg.clone()); + let (mut raya, raya_events, raya_addr) = create_test_client(bus.clone(), reg.clone()); + let (mut pax, pax_events, pax_addr) = create_test_client(bus.clone(), reg.clone()); + + let convo_id = saro + .create_group_conversation(&[&raya_addr]) + .expect("saro create group"); + + // The invite lands once saro's steward commit finalizes (wakeup-driven); + // both sides then share the de-mls conversation id. + let raya_convo_id = wait_for_group_started(&raya_events, "raya ConversationStarted"); + assert_eq!(raya_convo_id, convo_id); + + saro.send_message(&convo_id, b"hello raya").unwrap(); + assert_eq!( + wait_for_message(&raya_events, b"hello raya").as_deref(), + Some(saro_addr.as_str()) + ); + + raya.send_message(&raya_convo_id, b"hi saro").unwrap(); + assert_eq!( + wait_for_message(&saro_events, b"hi saro").as_deref(), + Some(raya_addr.as_str()) + ); + + // A non-creator grows the group: raya proposes pax, the steward commits, + // and raya (who holds the pending invite) routes the welcome to pax. + raya.add_group_members(&raya_convo_id, &[&pax_addr]) + .expect("raya add pax"); + let pax_convo_id = wait_for_group_started(&pax_events, "pax ConversationStarted"); + assert_eq!(pax_convo_id, convo_id); + + // Everyone is at the post-add epoch: a message from the creator reaches + // both peers, and one from the newest member reaches both elders. + saro.send_message(&convo_id, b"all three?").unwrap(); + assert_eq!( + wait_for_message(&raya_events, b"all three?").as_deref(), + Some(saro_addr.as_str()) + ); + assert_eq!( + wait_for_message(&pax_events, b"all three?").as_deref(), + Some(saro_addr.as_str()) + ); + + pax.send_message(&pax_convo_id, b"pax is in").unwrap(); + assert_eq!( + wait_for_message(&saro_events, b"pax is in").as_deref(), + Some(pax_addr.as_str()) + ); + assert_eq!( + wait_for_message(&raya_events, b"pax is in").as_deref(), + Some(pax_addr.as_str()) + ); + + assert_eq!(saro.list_conversations().unwrap().len(), 1); + assert_eq!(raya.list_conversations().unwrap().len(), 1); + assert_eq!(pax.list_conversations().unwrap().len(), 1); +} + +/// A batch add is validated before any member is proposed: a member whose +/// account is endorsed in the directory but whose device registered no key +/// package fails the whole call, the resolvable member in the same batch is +/// not invited, and the group keeps working. +#[test] +fn add_batch_with_missing_key_package_invites_no_one() { + let bus = MessageBus::default(); + let mut reg = EphemeralRegistry::new(); + + let (mut saro, _saro_events, _saro_addr) = create_test_client(bus.clone(), reg.clone()); + let (_raya, raya_events, raya_addr) = create_test_client(bus.clone(), reg.clone()); + let (_pax, pax_events, pax_addr) = create_test_client(bus.clone(), reg.clone()); + + // Ghost: its account endorses a device in the directory, but that device + // never registered a key package (no client was built for it). + let ghost_account = TestLogosAccount::new(); + let ghost_delegate = DelegateSigner::random(); + ghost_account + .add_delegate_signer(&mut reg, ghost_delegate.public_key()) + .unwrap(); + + let convo_id = saro + .create_group_conversation(&[&raya_addr]) + .expect("saro create group"); + wait_for_group_started(&raya_events, "raya ConversationStarted"); + + saro.add_group_members(&convo_id, &[&ghost_account.address(), &pax_addr]) + .expect_err("ghost has no key package"); + + // Pax was in the failed batch and must not have been invited. + assert!( + pax_events.recv_timeout(Duration::from_secs(1)).is_err(), + "pax must not join from a failed batch" + ); + + // The failed add left the group functional. + saro.send_message(&convo_id, b"still alive").unwrap(); + wait_for_message(&raya_events, b"still alive"); +} + +/// Group membership is resolved through the account directory, so inviting an +/// address whose account never published a bundle fails at resolution — on +/// create and on add alike. +#[test] +fn group_invite_of_unpublished_account_is_an_error() { + let bus = MessageBus::default(); + let reg = EphemeralRegistry::new(); + + let (mut saro, _saro_events, _saro_addr) = create_test_client(bus.clone(), reg.clone()); + let unpublished = TestLogosAccount::new(); + + let err = saro + .create_group_conversation(&[&unpublished.address()]) + .expect_err("no bundle published for the account"); + assert!(matches!( + err, + logos_generic_chat::ClientError::AccountResolution(_) + )); + + let convo_id = saro.create_group_conversation(&[]).expect("empty group"); + let err = saro + .add_group_members(&convo_id, &[&unpublished.address()]) + .expect_err("no bundle published for the account"); + assert!(matches!( + err, + logos_generic_chat::ClientError::AccountResolution(_) + )); +} From 884509a4cf67df39d7dcc48c60b7f30234e8960a Mon Sep 17 00:00:00 2001 From: osmaczko <33099791+osmaczko@users.noreply.github.com> Date: Sat, 4 Jul 2026 21:30:24 +0200 Subject: [PATCH 2/9] fix: class inbound DirectV1 joins as Private, not Group The joiner of a DirectV1 (pairwise) conversation received it classed as Group, because dispatch_to_inbox2 hardcoded ConversationClass::Group for every InboxV2 join. DirectV1 welcomes (InviteType::GroupV1) and GroupV2 welcomes (InviteType::GroupV2) both arrive over InboxV2, so a plain 1:1 invite surfaced to the display layer as a group. ConversationClass is documented as stable across protocol versions of the same conversation shape, and DirectV1 is the pairwise shape, so its joiner must see Private. - InboxV2::handle_frame returns the class alongside the convo: InviteType::GroupV1 (the DirectV1 welcome carrier) yields Private, InviteType::GroupV2 yields Group. - dispatch_to_inbox2 propagates that class instead of hardcoding Group. - direct_v1_by_account_address asserts the joiner sees Private. --- core/conversations/src/core.rs | 7 ++----- core/conversations/src/inbox_v2.rs | 15 ++++++++++++--- crates/generic-chat/tests/saro_and_raya.rs | 11 ++++++++--- 3 files changed, 22 insertions(+), 11 deletions(-) diff --git a/core/conversations/src/core.rs b/core/conversations/src/core.rs index 25e6fb2..ab99cff 100644 --- a/core/conversations/src/core.rs +++ b/core/conversations/src/core.rs @@ -369,16 +369,13 @@ impl<'a, S: ExternalServices + 'static> Core { // Dispatch encrypted payload to the post-quantum inbox. fn dispatch_to_inbox2(&mut self, payload: &[u8]) -> Result { - if let Some(convo) = self.pq_inbox.handle_frame(&mut self.services, payload)? { + if let Some((convo, class)) = self.pq_inbox.handle_frame(&mut self.services, payload)? { let convo_id = convo.id().to_string(); // Cache convos created by InboxV2 self.register_convo(ConvoTypeOwned::Group(convo))?; Ok(PayloadOutcome::Inbox(InboxOutcome { - new_conversation: crate::NewConversation { - convo_id, - class: crate::ConversationClass::Group, - }, + new_conversation: crate::NewConversation { convo_id, class }, initial: None, })) } else { diff --git a/core/conversations/src/inbox_v2.rs b/core/conversations/src/inbox_v2.rs index 8de4340..4895841 100644 --- a/core/conversations/src/inbox_v2.rs +++ b/core/conversations/src/inbox_v2.rs @@ -20,6 +20,7 @@ use crate::conversation::GroupConvo; use crate::conversation::GroupV1Convo; use crate::conversation::GroupV2Convo; use crate::conversation::Identified as _; +use crate::outcomes::ConversationClass; use crate::service_context::{ExternalServices, ServiceContext}; use crate::utils::{blake2b_hex, hash_size}; use crate::{AddressedEnvelope, IdentId, IdentIdRef, IdentityProvider}; @@ -70,6 +71,10 @@ pub fn invite_user_v2( .map_err(ChatError::generic) } +/// A convo built from an InboxV2 invite, paired with the display class its +/// invite type implies. +type ClassifiedConvo = (Box>, ConversationClass); + /// A PQ focused Conversation initializer. /// InboxV2 is signer-scoped: it receives invites under this installation's /// signer id (the hex of the signer's verifying key), supporting PQ based @@ -112,12 +117,15 @@ impl InboxV2 { conversation_id_for(&self.ident_id) } + /// The convo built from an invite, paired with the display class its invite + /// type implies: `InviteType::GroupV1` carries the pairwise DirectV1 welcome, + /// so it is `Private`; `InviteType::GroupV2` is a real group. #[instrument(name = "inboxV2.handle_frame", skip_all, fields(user_id = %service_ctx.mls_identity.display_name()))] pub fn handle_frame( &self, service_ctx: &mut ServiceContext, payload_bytes: &[u8], - ) -> Result>>, ChatError> { + ) -> Result>, ChatError> { // On a broadcast transport the inbox address also receives traffic // that isn't an invite (or that prost decodes into an empty frame). // Treat anything we can't interpret as "not for us" and skip it, @@ -131,14 +139,15 @@ impl InboxV2 { match payload { InviteType::GroupV1(inv) => { - Ok(Some(Box::new(self.handle_heavy_invite(service_ctx, inv)?))) + let convo = self.handle_heavy_invite(service_ctx, inv)?; + Ok(Some((Box::new(convo), ConversationClass::Private))) } InviteType::GroupV2(welcome_bytes) => { info!("Process WelcomeMessage"); let mw = MemberWelcome::decode(welcome_bytes.as_slice()).map_err(ChatError::generic)?; let convo = GroupV2Convo::new_from_welcome(service_ctx, &mw)?; - Ok(Some(Box::new(convo))) + Ok(Some((Box::new(convo), ConversationClass::Group))) } } } diff --git a/crates/generic-chat/tests/saro_and_raya.rs b/crates/generic-chat/tests/saro_and_raya.rs index 3ad1573..eca1b0e 100644 --- a/crates/generic-chat/tests/saro_and_raya.rs +++ b/crates/generic-chat/tests/saro_and_raya.rs @@ -5,8 +5,8 @@ use crossbeam_channel::{Receiver, Sender}; use crypto::Ed25519VerifyingKey; use logos_account::TestLogosAccount; use logos_generic_chat::{ - AddressedEnvelope, ChatClient, ChatClientBuilder, DelegateSigner, DeliveryService, Event, - InProcessDelivery, MessageBus, Transport, + AddressedEnvelope, ChatClient, ChatClientBuilder, ConversationClass, DelegateSigner, + DeliveryService, Event, InProcessDelivery, MessageBus, Transport, }; /// Publish a signed device bundle endorsing `device` as a device of `account`, @@ -166,8 +166,13 @@ fn direct_v1_by_account_address() { assert_eq!(raya.addr(), raya_account_addr.as_str()); let convo_id = saro.create_direct_conversation(&raya_account_addr).unwrap(); + // DirectV1 is the pairwise shape, so the joiner sees it classed Private even + // though its welcome rides the InboxV2 (GroupV1 invite) path. let raya_convo_id = expect_event(&raya_events, "ConversationStarted", |e| match e { - Event::ConversationStarted { convo_id, .. } => Ok(convo_id), + Event::ConversationStarted { convo_id, class } => { + assert_eq!(class, ConversationClass::Private); + Ok(convo_id) + } other => Err(other), }); From 8297ec8847ca95fb017f7001965a3542280e759c Mon Sep 17 00:00:00 2001 From: osmaczko <33099791+osmaczko@users.noreply.github.com> Date: Sat, 4 Jul 2026 21:39:43 +0200 Subject: [PATCH 3/9] feat: expose a group's roster, deduped to one entry per account The display layer needs a group's membership, but nothing exposed it: de-mls holds the authoritative roster (MLS group state) with no public accessor, and members added by other members stay invisible until they send a message. Rebuilding the roster from observed messages would fork state the crypto layer owns and be wrong exactly when a group grows. - GroupConvo::members() returns each member's hex-encoded MLS leaf-credential content, self included. GroupV2Convo delegates to de-mls and guarantees self-inclusion; GroupV1Convo reads its openmls leaves. - Core::group_members(convo_id) mirrors group_add_member's dispatch: a cached group yields its members, a direct conversation is an UnsupportedFunction, otherwise the group is loaded. - ChatClient::group_members returns Vec, resolving each member's account claim through the directory. A member whose account claim is unconfirmable is listed by device with account None rather than dropped: it is cryptographically in the group, only the account claim is unproven. The credential parsing decode_sender did is factored into parse_credential and shared by both, leaving decode_sender's stricter drop semantics for message senders unchanged. - Because resolve_device_ids fans an account out to every endorsed device, an account whose devices all join surfaced once per device; group_members dedups by account, keeping the first-seen device as the account's representative. Members with no confirmed account stay individual, keyed by their unique device key. - Unit tests cover the tolerant-vs-drop split and the per-account dedup; the three-member group integration test asserts the roster converges after create and after each add, and a solo group lists only its creator. --- core/conversations/src/conversation.rs | 4 + .../src/conversation/group_v1.rs | 8 + .../src/conversation/group_v2.rs | 10 + core/conversations/src/core.rs | 22 ++ crates/generic-chat/src/client.rs | 282 +++++++++++++++--- crates/generic-chat/src/lib.rs | 2 +- crates/generic-chat/tests/group_v2.rs | 52 ++++ 7 files changed, 345 insertions(+), 35 deletions(-) diff --git a/core/conversations/src/conversation.rs b/core/conversations/src/conversation.rs index 6f17892..7774ff4 100644 --- a/core/conversations/src/conversation.rs +++ b/core/conversations/src/conversation.rs @@ -42,6 +42,10 @@ pub(crate) trait GroupConvo: Convo + std::fmt::Debug + S cx: &mut ServiceContext, members: &[IdentIdRef], ) -> Result<(), ChatError>; + + /// Each current member's MLS leaf-credential content (hex-encoded), self + /// included. + fn members(&self) -> Result>, ChatError>; } pub(crate) trait Identified { diff --git a/core/conversations/src/conversation/group_v1.rs b/core/conversations/src/conversation/group_v1.rs index 3dcf96e..8aa7402 100644 --- a/core/conversations/src/conversation/group_v1.rs +++ b/core/conversations/src/conversation/group_v1.rs @@ -341,4 +341,12 @@ impl GroupConvo for GroupV1Convo { self.send_payload(cx, commit.to_bytes()?) } + + fn members(&self) -> Result>, ChatError> { + Ok(self + .mls_group + .members() + .map(|m| m.credential.serialized_content().to_vec()) + .collect()) + } } diff --git a/core/conversations/src/conversation/group_v2.rs b/core/conversations/src/conversation/group_v2.rs index 4be46f0..dfcee72 100644 --- a/core/conversations/src/conversation/group_v2.rs +++ b/core/conversations/src/conversation/group_v2.rs @@ -316,6 +316,16 @@ where result.and(flushed) } + fn members(&self) -> Result>, ChatError> { + // Guarantee the local member is listed so callers see the full roster. + let mut members = self.conversation.members()?; + let self_id = self.conversation.member_id_bytes().to_vec(); + if !members.contains(&self_id) { + members.push(self_id); + } + Ok(members) + } + // fn conversation_state(&self) -> Result { // Ok(self // .conversation diff --git a/core/conversations/src/core.rs b/core/conversations/src/core.rs index ab99cff..1dcaf3a 100644 --- a/core/conversations/src/core.rs +++ b/core/conversations/src/core.rs @@ -300,6 +300,28 @@ impl<'a, S: ExternalServices + 'static> Core { } } + /// Each member's MLS leaf-credential content (hex-encoded); errors if + /// `convo_id` names a direct (non-group) conversation. + pub fn group_members(&mut self, convo_id: &str) -> Result>, ChatError> { + if self.cached_convos.contains_key(convo_id) { + let convo = self + .cached_convos + .get(convo_id) + .ok_or_else(|| ChatError::NoConvo(convo_id.to_string()))?; + + match convo { + ConvoTypeOwned::Group(group_convo) => group_convo.members(), + ConvoTypeOwned::Direct(convo) => Err(ChatError::UnsupportedFunction( + convo.id().into(), + "List Members".into(), + )), + } + } else { + let convo = self.load_group_convo(convo_id)?; + convo.members() + } + } + pub fn list_conversations(&self) -> Result, ChatError> { // Check Legacy load_convo store let records = self.services.store.load_conversations()?; diff --git a/crates/generic-chat/src/client.rs b/crates/generic-chat/src/client.rs index dc173da..2d2b93d 100644 --- a/crates/generic-chat/src/client.rs +++ b/crates/generic-chat/src/client.rs @@ -1,3 +1,4 @@ +use std::collections::HashSet; use std::sync::Arc; use std::thread::{self, JoinHandle}; @@ -20,6 +21,19 @@ type ClientCore = Core<(DelegateIdentity, T, R, ThreadedWakeupService, type AccountAddressRef<'a> = &'a str; type LocalSignerId = IdentId; +/// A member of a group conversation's roster. +/// +/// Shares [`MessageSender`]'s field semantics: `account` is set only when the +/// member's credential claimed an account *and* the directory confirmed this +/// device belongs to it. Unlike a message sender, an unconfirmable claim does +/// not hide the member: it is cryptographically in the group, so it is listed +/// by `local_identity` (its device) with `account: None`. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct GroupMember { + pub account: Option, + pub local_identity: IdentId, +} + /// The transport as the client sees it: a [`DeliveryService`] for outbound /// publishing plus the inbound payload stream the worker drains. One object owns /// both directions of the boundary. @@ -189,6 +203,20 @@ where .map_err(Into::into) } + /// The group's roster, one [`GroupMember`] per account (self included). An + /// account's several devices collapse to a single entry surfacing that + /// account; a member whose account claim the directory can't confirm stays + /// on the roster individually, keyed by its device. Costs one directory + /// lookup per member that claims an account, the same per-member cost a + /// received message's sender check pays. + pub fn group_members(&mut self, convo_id: &str) -> Result, ClientError> { + let credentials = self.core.lock().group_members(convo_id)?; + let members = credentials + .iter() + .filter_map(|credential| roster_member(&self.directory, credential)); + Ok(dedup_members(members)) + } + /// Parse intro bundle bytes and initiate a private conversation. Outbound /// envelopes are published by the core. Returns this side's conversation ID. /// @@ -349,59 +377,129 @@ enum SenderError { Unverified, } -/// Decode and verify a message's sender from its credential, checked against the -/// account → device directory (our account store). +/// The resolution of a credential's account claim against the directory. +enum AccountClaim { + /// The credential claimed no account. + None, + /// Confirmed: the directory lists this device under the claimed account. + Verified(IdentId), + /// An account was claimed but could not be confirmed (see [`SenderError`]). + Unverified(SenderError), +} + +/// Parse a wire credential into the device it names and the resolution of any +/// account claim, checked against the account → device directory. `Err` only +/// when no device can be attributed at all (missing or unparseable credential). /// -/// `Ok(sender)` — deliver with the sender; its `account` is set only when the -/// directory confirmed the device, so it is always verified. `Err` — drop the -/// message (including when no credential is present, since every delivered -/// message must carry an explicit sender). -fn decode_sender( +/// The account-claim policy is left to the caller: a message drops on an +/// unconfirmable claim, a roster entry keeps the device and forgoes the account. +fn parse_credential( directory: &impl AccountDirectory, encoded: &[u8], -) -> Result { - // No credential at all: there is no sender to attribute, so drop it. +) -> Result<(IdentId, AccountClaim), SenderError> { + // No credential at all: there is no device to attribute. if encoded.is_empty() { return Err(SenderError::Missing); } let Ok(data) = hex::decode(encoded) else { - tracing::warn!("sender credential is not valid hex; dropping message"); + tracing::warn!("credential is not valid hex"); return Err(SenderError::NotHex); }; - let cred = match DelegateCredential::try_from(data) { - Ok(cred) => cred, - Err(_) => { - tracing::warn!("malformed sender credential; dropping message"); - return Err(SenderError::Malformed); - } + let Ok(cred) = DelegateCredential::try_from(data) else { + tracing::warn!("malformed credential"); + return Err(SenderError::Malformed); }; - let device = hex::encode(cred.delegate_id().as_ref()); + let device = IdentId::new(hex::encode(cred.delegate_id().as_ref())); // An unassociated delegate asserts no account → device mapping. let Some(account_addr) = cred.account_addr() else { - return Ok(MessageSender { - account: None, - local_identity: IdentId::new(device), - }); + return Ok((device, AccountClaim::None)); }; let Some(account_key) = account_key_from_hex(account_addr) else { - tracing::warn!( - account_addr, - "sender account address is not a verifying key; dropping message" - ); - return Err(SenderError::AccountNotAKey); + tracing::warn!(account_addr, "account address is not a verifying key"); + return Ok(( + device, + AccountClaim::Unverified(SenderError::AccountNotAKey), + )); }; - match directory.fetch(&account_key) { - Ok(Some(set)) if set.devices.iter().any(|d| d == &device) => Ok(MessageSender { - account: Some(IdentId::new(account_addr.to_string())), - local_identity: IdentId::new(device), - }), + let claim = match directory.fetch(&account_key) { + Ok(Some(set)) if set.devices.iter().any(|d| d.as_str() == device.as_str()) => { + AccountClaim::Verified(IdentId::new(account_addr.to_string())) + } _ => { - tracing::warn!(account_addr, %device, "account → device mapping is wrong or unconfirmable; dropping message"); - Err(SenderError::Unverified) + tracing::warn!(account_addr, device = %device.as_str(), "account → device mapping is wrong or unconfirmable"); + AccountClaim::Unverified(SenderError::Unverified) } + }; + Ok((device, claim)) +} + +/// Decode and verify a message's sender from its credential, checked against the +/// account → device directory (our account store). +/// +/// `Ok(sender)` — deliver with the sender; its `account` is set only when the +/// directory confirmed the device, so it is always verified. `Err` — drop the +/// message (including when no credential is present, since every delivered +/// message must carry an explicit sender). +fn decode_sender( + directory: &impl AccountDirectory, + encoded: &[u8], +) -> Result { + let (device, claim) = parse_credential(directory, encoded)?; + match claim { + AccountClaim::None => Ok(MessageSender { + account: None, + local_identity: device, + }), + AccountClaim::Verified(account) => Ok(MessageSender { + account: Some(account), + local_identity: device, + }), + // An unconfirmable account claim drops the message: every delivered + // message must carry a verified sender. + AccountClaim::Unverified(err) => Err(err), } } +/// Map a group member's credential (as reported by MLS, in the same hex-encoded +/// form a message carries as its sender) to a roster entry, tolerating an +/// unconfirmable account claim by listing the device without an account. `None` +/// only when the credential cannot be parsed, which does not happen for a real +/// MLS leaf. +fn roster_member(directory: &impl AccountDirectory, encoded: &[u8]) -> Option { + let (device, claim) = parse_credential(directory, encoded).ok()?; + let account = match claim { + AccountClaim::Verified(account) => Some(account), + AccountClaim::None | AccountClaim::Unverified(_) => None, + }; + Some(GroupMember { + account, + local_identity: device, + }) +} + +/// The key that decides whether two roster entries are the same member: a +/// verified account, so an account's several devices count once; or, for a +/// member with no confirmed account, its device — unique per MLS leaf, so it +/// never merges with another. +fn member_key(member: &GroupMember) -> &str { + member + .account + .as_ref() + .unwrap_or(&member.local_identity) + .as_str() +} + +/// Collapse a roster to one entry per account (keeping the first-seen device as +/// the account's representative) while leaving account-less members individual, +/// order preserved. +fn dedup_members(members: impl IntoIterator) -> Vec { + let mut seen = HashSet::new(); + members + .into_iter() + .filter(|member| seen.insert(member_key(member).to_owned())) + .collect() +} + fn convo_events(outcome: ConvoOutcome, directory: &impl AccountDirectory) -> Vec { let ConvoOutcome { convo_id, content } = outcome; content @@ -448,7 +546,10 @@ mod sender_check_tests { use libchat::IdentId; use logos_account::{DeviceSet, SignedDeviceBundle}; - use super::{MessageSender, SenderError, decode_sender}; + use super::{ + GroupMember, MessageSender, SenderError, decode_sender, dedup_members, member_key, + roster_member, + }; use crate::delegate::DelegateCredential; /// In-test account → device directory. Holds device id sets keyed by the hex @@ -618,4 +719,117 @@ mod sender_check_tests { Err(SenderError::AccountNotAKey) ); } + + /// A verified account claim surfaces the member's account and device — the + /// same happy path as a message sender. + #[test] + fn roster_verified_member_surfaces_account() { + let account = key(); + let device = key(); + let dir = FakeDir::with_devices(&account, &[&device]); + let cred = DelegateCredential::associated(&device, &hex::encode(account.as_ref())); + assert_eq!( + roster_member(&dir, &encoded(cred)), + Some(GroupMember { + account: Some(local_id(&account)), + local_identity: local_id(&device), + }) + ); + } + + /// Unlike a message sender, a spoofed account claim does not hide the + /// member: the device is cryptographically in the group, so it is listed + /// with no account rather than dropped. + #[test] + fn roster_contradicted_claim_lists_device_without_account() { + let account = key(); + let endorsed = key(); + let spoofer = key(); + let dir = FakeDir::with_devices(&account, &[&endorsed]); + let cred = DelegateCredential::associated(&spoofer, &hex::encode(account.as_ref())); + assert_eq!( + roster_member(&dir, &encoded(cred)), + Some(GroupMember { + account: None, + local_identity: local_id(&spoofer), + }) + ); + } + + /// A member whose credential claims no account is listed by device only. + #[test] + fn roster_unassociated_member_lists_device_without_account() { + let dir = FakeDir::default(); + let device = key(); + let cred = DelegateCredential::unassociated(&device); + assert_eq!( + roster_member(&dir, &encoded(cred)), + Some(GroupMember { + account: None, + local_identity: local_id(&device), + }) + ); + } + + /// A directory outage leaves the account unconfirmed, but the member stays + /// on the roster by device (a message would drop here). + #[test] + fn roster_directory_outage_lists_device_without_account() { + let account = key(); + let device = key(); + let dir = FakeDir { + fail: true, + ..Default::default() + }; + let cred = DelegateCredential::associated(&device, &hex::encode(account.as_ref())); + assert_eq!( + roster_member(&dir, &encoded(cred)), + Some(GroupMember { + account: None, + local_identity: local_id(&device), + }) + ); + } + + /// A non-key account address can't be confirmed, so the member is listed by + /// device without an account. + #[test] + fn roster_non_key_account_lists_device_without_account() { + let dir = FakeDir::default(); + let device = key(); + let cred = DelegateCredential::associated(&device, "user@example.com"); + assert_eq!( + roster_member(&dir, &encoded(cred)), + Some(GroupMember { + account: None, + local_identity: local_id(&device), + }) + ); + } + + /// The roster collapses an account's several devices into one entry (keeping + /// the first device seen) while leaving account-less members individual, + /// order preserved. + #[test] + fn dedup_collapses_account_devices_and_keeps_unknowns() { + let with_account = |account: &str, device: &str| GroupMember { + account: Some(IdentId::new(account.to_string())), + local_identity: IdentId::new(device.to_string()), + }; + let device_only = |device: &str| GroupMember { + account: None, + local_identity: IdentId::new(device.to_string()), + }; + let roster = dedup_members(vec![ + with_account("alice", "alice-dev-1"), + with_account("alice", "alice-dev-2"), + device_only("orphan-x"), + with_account("bob", "bob-dev-1"), + device_only("orphan-y"), + ]); + let keys: Vec<&str> = roster.iter().map(member_key).collect(); + assert_eq!(keys, ["alice", "orphan-x", "bob", "orphan-y"]); + // Alice's collapsed entry keeps her first-seen device. + assert_eq!(roster[0].local_identity.as_str(), "alice-dev-1"); + } } diff --git a/crates/generic-chat/src/lib.rs b/crates/generic-chat/src/lib.rs index 2bc2c54..6d7d14f 100644 --- a/crates/generic-chat/src/lib.rs +++ b/crates/generic-chat/src/lib.rs @@ -6,7 +6,7 @@ mod errors; mod event; pub use builder::{ChatClientBuilder, Unset}; -pub use client::{ChatClient, Transport}; +pub use client::{ChatClient, GroupMember, Transport}; pub use delegate::DelegateSigner; pub use delivery_in_process::{InProcessDelivery, MessageBus}; pub use errors::ClientError; diff --git a/crates/generic-chat/tests/group_v2.rs b/crates/generic-chat/tests/group_v2.rs index bfa2073..dcc7287 100644 --- a/crates/generic-chat/tests/group_v2.rs +++ b/crates/generic-chat/tests/group_v2.rs @@ -87,6 +87,30 @@ fn wait_for_group_started(events: &Receiver, label: &str) -> String { }) } +/// Poll a client's roster for `convo_id` until its verified accounts equal +/// `expected` (order-independent), or panic after a timeout. The roster settles +/// asynchronously as each member applies the add commit, so it is polled rather +/// than snapshotted. +fn wait_for_members(client: &mut TestClient, convo_id: &str, expected: &[&str]) { + use std::collections::BTreeSet; + let want: BTreeSet<&str> = expected.iter().copied().collect(); + let deadline = std::time::Instant::now() + Duration::from_secs(10); + loop { + let roster = client.group_members(convo_id).expect("group_members"); + let got: BTreeSet<&str> = roster + .iter() + .filter_map(|m| m.account.as_ref().map(|a| a.as_str())) + .collect(); + if got == want { + return; + } + if std::time::Instant::now() >= deadline { + panic!("roster did not converge for {convo_id}: got {got:?}, want {want:?}"); + } + std::thread::sleep(Duration::from_millis(20)); + } +} + /// Wait for `content` to arrive and return the sender's verified account. fn wait_for_message(events: &Receiver, content: &[u8]) -> Option { let label = format!("MessageReceived({})", String::from_utf8_lossy(content)); @@ -121,6 +145,10 @@ fn group_v2_three_members() { let raya_convo_id = wait_for_group_started(&raya_events, "raya ConversationStarted"); assert_eq!(raya_convo_id, convo_id); + // Both sides see the two-account roster once the add commits. + wait_for_members(&mut saro, &convo_id, &[&saro_addr, &raya_addr]); + wait_for_members(&mut raya, &raya_convo_id, &[&saro_addr, &raya_addr]); + saro.send_message(&convo_id, b"hello raya").unwrap(); assert_eq!( wait_for_message(&raya_events, b"hello raya").as_deref(), @@ -162,11 +190,35 @@ fn group_v2_three_members() { Some(pax_addr.as_str()) ); + // All three rosters converge on the same three accounts. + let all = [saro_addr.as_str(), raya_addr.as_str(), pax_addr.as_str()]; + wait_for_members(&mut saro, &convo_id, &all); + wait_for_members(&mut raya, &raya_convo_id, &all); + wait_for_members(&mut pax, &pax_convo_id, &all); + assert_eq!(saro.list_conversations().unwrap().len(), 1); assert_eq!(raya.list_conversations().unwrap().len(), 1); assert_eq!(pax.list_conversations().unwrap().len(), 1); } +/// The creator is in its own roster from the start, with no other members: the +/// roster always includes self. +#[test] +fn group_creator_is_in_own_roster() { + let bus = MessageBus::default(); + let reg = EphemeralRegistry::new(); + + let (mut saro, _saro_events, saro_addr) = create_test_client(bus.clone(), reg.clone()); + + let convo_id = saro.create_group_conversation(&[]).expect("empty group"); + let roster = saro.group_members(&convo_id).expect("group_members"); + let accounts: Vec> = roster + .iter() + .map(|m| m.account.as_ref().map(|a| a.as_str())) + .collect(); + assert_eq!(accounts, vec![Some(saro_addr.as_str())]); +} + /// A batch add is validated before any member is proposed: a member whose /// account is endorsed in the directory but whose device registered no key /// package fails the whole call, the resolvable member in the same batch is From da4ea8259ea43f48c1d00ddc15c9d4e2cc07dd90 Mon Sep 17 00:00:00 2001 From: osmaczko <33099791+osmaczko@users.noreply.github.com> Date: Sun, 5 Jul 2026 17:24:42 +0200 Subject: [PATCH 4/9] feat: retry the registry on transient 5xx with backoff and jitter The keypackage/account registry is reliable request-by-request but sheds concurrent bursts with a 5xx, so several instances registering at once each hard-failed on init. HttpRegistry's four calls now retry network errors and 5xx/429 with exponential backoff and full jitter (the jitter decorrelates concurrent publishers so their retries don't re-collide); 4xx and success return immediately. The total retry window is bounded to a few seconds. --- .../components/src/contact_registry/http.rs | 108 +++++++++++++++++- 1 file changed, 104 insertions(+), 4 deletions(-) diff --git a/extensions/components/src/contact_registry/http.rs b/extensions/components/src/contact_registry/http.rs index e7955d5..a1cbca9 100644 --- a/extensions/components/src/contact_registry/http.rs +++ b/extensions/components/src/contact_registry/http.rs @@ -125,7 +125,7 @@ impl RegistrationService for HttpRegistry { }; let url = format!("{}/v0/keypackage", self.base_url); - let resp = self.http.post(&url).json(&req).send()?; + let resp = send_retrying(|| self.http.post(&url).json(&req))?; if !resp.status().is_success() { let status = resp.status().as_u16(); let body = resp.text().unwrap_or_default(); @@ -136,7 +136,7 @@ impl RegistrationService for HttpRegistry { fn retrieve(&self, device_id: &str) -> Result>, HttpRegistryError> { let url = format!("{}/v0/keypackage/{}", self.base_url, device_id); - let resp = self.http.get(&url).send()?; + let resp = send_retrying(|| self.http.get(&url))?; if resp.status().as_u16() == 404 { return Ok(None); } @@ -188,7 +188,7 @@ impl AccountDirectory for HttpRegistry { }; let url = format!("{}/v0/account", self.base_url); - let resp = self.http.post(&url).json(&req).send()?; + let resp = send_retrying(|| self.http.post(&url).json(&req))?; if !resp.status().is_success() { let status = resp.status().as_u16(); let body = resp.text().unwrap_or_default(); @@ -203,7 +203,7 @@ impl AccountDirectory for HttpRegistry { self.base_url, hex::encode(account.as_ref()) ); - let resp = self.http.get(&url).send()?; + let resp = send_retrying(|| self.http.get(&url))?; if resp.status().as_u16() == 404 { return Ok(None); } @@ -266,6 +266,67 @@ fn decode_payload(payload: &[u8]) -> Option<(u64, &[u8])> { Some((timestamp_ms, &payload[8..])) } +/// Retry budget for the registry's transient, load-induced 5xx/429 responses. +/// The service is reliable request-by-request but sheds concurrent bursts, so a +/// few backed-off retries let a request land once the burst clears. Sized to +/// stay within chat_module's ~20s init IPC budget even at the worst-case sum. +const MAX_RETRIES: u32 = 4; +const RETRY_BASE_MS: u64 = 200; +const RETRY_MAX_BACKOFF_MS: u64 = 2000; + +/// Send a request built by `build`, retrying transient failures — network errors +/// and 5xx/429 responses — with exponential backoff and full jitter. The +/// registry is reliable request-by-request but sheds concurrent bursts with a +/// 5xx, so a backed-off retry lands once the burst clears; a 4xx (and any other +/// final response) is returned to the caller unchanged. `build` is re-invoked per +/// attempt because sending consumes the builder. +fn send_retrying( + build: impl Fn() -> reqwest::blocking::RequestBuilder, +) -> Result { + let mut attempt = 0; + loop { + let outcome = build().send(); + let transient = match &outcome { + Err(_) => true, // network error / timeout: worth another try + Ok(resp) => is_transient_status(resp.status()), + }; + if !transient || attempt >= MAX_RETRIES { + return Ok(outcome?); + } + std::thread::sleep(backoff_with_jitter(attempt)); + attempt += 1; + } +} + +/// Whether a response status is worth retrying: 5xx (the registry sheds +/// concurrent load with these) or 429 (explicit backpressure). A 4xx is the +/// caller's fault and won't change on retry. +fn is_transient_status(status: reqwest::StatusCode) -> bool { + status.is_server_error() || status == reqwest::StatusCode::TOO_MANY_REQUESTS +} + +/// Full-jitter exponential backoff: a random delay in +/// `[0, min(RETRY_MAX_BACKOFF_MS, RETRY_BASE_MS * 2^attempt)]`. The jitter +/// decorrelates concurrent publishers so their retries don't collide into the +/// same burst that failed them. +fn backoff_with_jitter(attempt: u32) -> Duration { + let exp = RETRY_BASE_MS.saturating_mul(1u64 << attempt.min(16)); + Duration::from_millis(jitter_below(exp.min(RETRY_MAX_BACKOFF_MS))) +} + +/// A value in `[0, max]`, seeded from the wall clock's sub-second nanos — enough +/// entropy to spread retries across processes without pulling in an RNG crate. +fn jitter_below(max: u64) -> u64 { + if max == 0 { + return 0; + } + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.subsec_nanos() as u64) + .unwrap_or(0); + nanos % (max + 1) +} + #[cfg(test)] mod tests { use super::*; @@ -325,4 +386,43 @@ mod tests { .verify(&payload, &signature) .expect("recovered key must verify the register-time signature"); } + + /// Only 5xx and 429 are retried; 2xx/4xx are returned to the caller as-is. + #[test] + fn only_5xx_and_429_are_transient() { + use reqwest::StatusCode; + for s in [500u16, 502, 503, 504, 429] { + assert!( + is_transient_status(StatusCode::from_u16(s).unwrap()), + "{s} should be retried" + ); + } + for s in [200u16, 201, 400, 401, 404, 409] { + assert!( + !is_transient_status(StatusCode::from_u16(s).unwrap()), + "{s} should not be retried" + ); + } + } + + /// Backoff never exceeds the exponential ceiling for its attempt, nor the + /// absolute cap — and the exponent shift can't overflow at high attempts. + #[test] + fn backoff_stays_within_the_cap() { + for attempt in 0..40u32 { + let ceiling = RETRY_BASE_MS + .saturating_mul(1u64 << attempt.min(16)) + .min(RETRY_MAX_BACKOFF_MS); + let delay = backoff_with_jitter(attempt).as_millis() as u64; + assert!(delay <= ceiling, "attempt {attempt}: {delay} > {ceiling}"); + } + } + + #[test] + fn jitter_is_bounded() { + assert_eq!(jitter_below(0), 0); + for _ in 0..200 { + assert!(jitter_below(50) <= 50); + } + } } From 3d2ff284eba40e2d783d1fbbbfaebb08bd0ac81e Mon Sep 17 00:00:00 2001 From: osmaczko <33099791+osmaczko@users.noreply.github.com> Date: Tue, 7 Jul 2026 12:12:45 +0200 Subject: [PATCH 5/9] fix: mark InboxV2 key package last-resort so members can join multiple groups A key package's init key is one-time-use: openmls deletes it after the first welcome that consumes it. Each installation registers a single key package, so a second group inviting the same member found no matching key package and rejected the welcome with "welcome not addressed to this member", the flaky group add. Mark the InboxV2 key package as last-resort (and advertise the extension in the leaf capabilities, which key-package validation requires) so openmls retains the init key, letting one key package admit an installation to any number of groups. This reuses one init key for every join, trading per-join forward secrecy for membership that just works. A TODO at the publish site tracks the intended one-time key-package pool (the registry pops one per fetch, the client replenishes) with last-resort as the exhaustion fallback (#169). Add regression tests: a member joining two groups (core harness) and two peers invited to several groups over the threaded client. --- core/conversations/src/inbox_v2.rs | 20 ++++++-- .../tests/test_group_v2.rs | 36 ++++++++++++++ crates/generic-chat/tests/group_v2.rs | 49 +++++++++++++++++++ 3 files changed, 102 insertions(+), 3 deletions(-) diff --git a/core/conversations/src/inbox_v2.rs b/core/conversations/src/inbox_v2.rs index 4895841..b67ceed 100644 --- a/core/conversations/src/inbox_v2.rs +++ b/core/conversations/src/inbox_v2.rs @@ -100,8 +100,11 @@ impl InboxV2 { ) -> Result<(), ChatError> { let keypackage_bytes = Self::create_keypackage(cx)?.tls_serialize_detached()?; - // TODO: (P3) Each keypackage can only be used once either enable... - // "LastResort" package or publish multiple + // TODO: publishes a single key package per installation. The intended + // design is a pool of one-time key packages (the registry pops one per + // fetch, the client replenishes) with the last-resort key package as the + // exhaustion fallback rather than the primary; that needs pop/claim + // semantics in the registry service. Tracked in #169. cx.registry .register(&cx.mls_identity, keypackage_bytes) .map_err(ChatError::generic)?; @@ -192,11 +195,22 @@ impl InboxV2 { fn create_keypackage( cx: &ServiceContext, ) -> Result { + // Last-resort key package. openmls consumes (deletes) a normal key + // package's init key on the first welcome that uses it; since each + // installation publishes just one, a second group inviting it would find + // no matching key package and reject the welcome ("welcome not addressed + // to this member"). Last-resort key packages are retained, so one admits + // the installation to any number of groups. Every key-package extension + // must be advertised in the leaf capabilities, hence LastResort there. let capabilities = Capabilities::builder() .ciphersuites(vec![CIPHER_SUITE]) - .extensions(vec![ExtensionType::ApplicationId]) + .extensions(vec![ + ExtensionType::ApplicationId, + ExtensionType::LastResort, + ]) .build(); let a = KeyPackage::builder() + .mark_as_last_resort() .leaf_node_capabilities(capabilities) .build( CIPHER_SUITE, diff --git a/core/integration_tests_core/tests/test_group_v2.rs b/core/integration_tests_core/tests/test_group_v2.rs index f000423..983dd7f 100644 --- a/core/integration_tests_core/tests/test_group_v2.rs +++ b/core/integration_tests_core/tests/test_group_v2.rs @@ -176,3 +176,39 @@ fn core_client_four_members_two_epochs() { && h.mira().check(&convo_id, MSG) }); } + +#[test] +fn member_joins_two_groups() { + // The same installation is invited to two separate groups. Its single + // registered key package is consumed by the first join, so the second + // group's welcome must still admit it — otherwise it is rejected with + // "welcome not addressed to this member" and never joins. Regression for + // key-package reuse across groups. + let _ = tracing_subscriber::fmt() + .with_max_level(tracing::Level::INFO) + .with_test_writer() + .try_init(); + + let mut harness = TestHarness::<2>::new(|_, _| {}); + let raya_addr = harness.raya().addr(); + + // Group 1: Saro invites Raya. + harness + .saro() + .create_group_convo_v2(&[&raya_addr]) + .expect("saro create group 1"); + harness.process_until_label("raya joins group 1", |h| h.raya().convo_count() == 1); + + // Group 2: Saro invites Raya again, into a fresh group. + harness + .saro() + .create_group_convo_v2(&[&raya_addr]) + .expect("saro create group 2"); + harness.process_until_label("raya joins group 2", |h| h.raya().convo_count() == 2); + + assert_eq!( + harness.raya().convo_count(), + 2, + "raya did not join the second group" + ); +} diff --git a/crates/generic-chat/tests/group_v2.rs b/crates/generic-chat/tests/group_v2.rs index dcc7287..42e2d3d 100644 --- a/crates/generic-chat/tests/group_v2.rs +++ b/crates/generic-chat/tests/group_v2.rs @@ -201,6 +201,55 @@ fn group_v2_three_members() { assert_eq!(pax.list_conversations().unwrap().len(), 1); } +/// The same two peers are invited to several groups at once. Each installation +/// registers a single key package, so admitting it to more than one group only +/// works if that key package survives a join — a regression guard for the +/// multi-group "welcome not addressed to this member" flake. +#[test] +fn peers_invited_to_many_groups() { + const GROUPS: usize = 3; + + let bus = MessageBus::default(); + let reg = EphemeralRegistry::new(); + + let (mut saro, _saro_events, saro_addr) = create_test_client(bus.clone(), reg.clone()); + let (_raya, raya_events, raya_addr) = create_test_client(bus.clone(), reg.clone()); + let (_pax, pax_events, pax_addr) = create_test_client(bus.clone(), reg.clone()); + + // Saro opens several groups, each inviting both Raya and Pax; every group + // reuses Raya's and Pax's one key package. + let mut convo_ids = Vec::new(); + for _ in 0..GROUPS { + convo_ids.push( + saro.create_group_conversation(&[&raya_addr, &pax_addr]) + .expect("saro create group"), + ); + } + + // Both peers must join all of them. + for _ in 0..GROUPS { + wait_for_group_started(&raya_events, "raya joins a group"); + wait_for_group_started(&pax_events, "pax joins a group"); + } + + // Every group is live: a distinct message in each reaches both peers with + // the creator's verified account. + for (i, convo_id) in convo_ids.iter().enumerate() { + let msg = format!("hello group {i}").into_bytes(); + saro.send_message(convo_id, &msg).unwrap(); + assert_eq!( + wait_for_message(&raya_events, &msg).as_deref(), + Some(saro_addr.as_str()) + ); + assert_eq!( + wait_for_message(&pax_events, &msg).as_deref(), + Some(saro_addr.as_str()) + ); + } + + assert_eq!(saro.list_conversations().unwrap().len(), GROUPS); +} + /// The creator is in its own roster from the start, with no other members: the /// roster always includes self. #[test] From f5a823c1de46aa5455458f75fba2b8fef0c3076a Mon Sep 17 00:00:00 2001 From: osmaczko <33099791+osmaczko@users.noreply.github.com> Date: Tue, 7 Jul 2026 17:09:52 +0200 Subject: [PATCH 6/9] fix: dedup list_conversations across the store and the in-memory cache A DirectV1 join persists its conversation to the store and also caches it in memory, so list_conversations saw it twice. It deduped with Vec::dedup, which only drops consecutive repeats, over cached_convos' nondeterministic HashMap order, so the duplicate survived whenever another cached conversation fell between the two copies. list_conversations then intermittently returned a conversation twice, and a consumer counting conversations (e.g. checking that a peer joined a group while a direct chat already existed) saw a flaky count. Dedup through a set so a conversation held in both stores is listed once regardless of iteration order. Add a DirectV1-then-GroupV2 regression test, which also covers key-package reuse across conversation types. --- core/conversations/src/core.rs | 10 ++++-- .../tests/test_group_v2.rs | 36 +++++++++++++++++++ 2 files changed, 43 insertions(+), 3 deletions(-) diff --git a/core/conversations/src/core.rs b/core/conversations/src/core.rs index 1dcaf3a..bab9fad 100644 --- a/core/conversations/src/core.rs +++ b/core/conversations/src/core.rs @@ -333,9 +333,13 @@ impl<'a, S: ExternalServices + 'static> Core { convos.push(convo.to_string()); } - // Conversations may use both storage mechanisms. - // Remove duplicates - convos.dedup(); + // A conversation can live in both the store and the in-memory cache (a + // DirectV1 join persists to the store and is also cached), so drop + // duplicates across the two. `Vec::dedup` only removes *consecutive* + // repeats and `cached_convos` iterates in nondeterministic HashMap + // order, so dedup through a set instead. + let mut seen = std::collections::HashSet::new(); + convos.retain(|c| seen.insert(c.clone())); Ok(convos) } diff --git a/core/integration_tests_core/tests/test_group_v2.rs b/core/integration_tests_core/tests/test_group_v2.rs index 983dd7f..afafd0a 100644 --- a/core/integration_tests_core/tests/test_group_v2.rs +++ b/core/integration_tests_core/tests/test_group_v2.rs @@ -212,3 +212,39 @@ fn member_joins_two_groups() { "raya did not join the second group" ); } + +#[test] +fn direct_v1_then_group_v2_reuses_key_package() { + // The reported flake: a DirectV1 (pairwise) conversation is opened with a + // peer first, then the same peer is invited to a GroupV2. Both fetch the + // peer's single registered key package; the DirectV1 join consumes it, so + // without a last-resort key package the group welcome finds no key package + // and is rejected ("welcome not addressed to this member"). + let _ = tracing_subscriber::fmt() + .with_max_level(tracing::Level::INFO) + .with_test_writer() + .try_init(); + + let mut harness = TestHarness::<2>::new(|_, _| {}); + let raya_addr = harness.raya().addr(); + + // 1. DirectV1 with Raya. + harness + .saro() + .create_direct_convo_v1(&[&raya_addr]) + .expect("saro create direct"); + harness.process_until_label("raya joins direct", |h| h.raya().convo_count() == 1); + + // 2. GroupV2 inviting the same Raya. + harness + .saro() + .create_group_convo_v2(&[&raya_addr]) + .expect("saro create group"); + harness.process_until_label("raya joins group", |h| h.raya().convo_count() == 2); + + assert_eq!( + harness.raya().convo_count(), + 2, + "raya did not join the group after a direct chat" + ); +} From 0adcc50553a13e625b4ffa8209dab24a01a00c98 Mon Sep 17 00:00:00 2001 From: osmaczko <33099791+osmaczko@users.noreply.github.com> Date: Wed, 8 Jul 2026 19:02:10 +0200 Subject: [PATCH 7/9] fix: dedup the GroupV2 add batch to avoid redundant fetches and duplicate invites Both create_group_convo_v2 and group_add_member funnel through GroupV2Convo::add_member, so a duplicate signer (an account that resolves to the same signer twice, or a repeated account) cost a redundant key-package fetch and a second Add proposal. The existing guard skipped only self and already-committed members, which a within-batch duplicate escapes because add_member opens a proposal the committed roster does not yet reflect, stranding a pending_invite that can later fire a spurious duplicate welcome. Dedup the requested signers before fetching, and guard the add loop with a membership set seeded from the roster and self, hoisting the per-iteration members() call out of the loop. --- .../src/conversation/group_v2.rs | 35 ++++++++++++++----- 1 file changed, 26 insertions(+), 9 deletions(-) diff --git a/core/conversations/src/conversation/group_v2.rs b/core/conversations/src/conversation/group_v2.rs index dfcee72..d2b91d9 100644 --- a/core/conversations/src/conversation/group_v2.rs +++ b/core/conversations/src/conversation/group_v2.rs @@ -260,6 +260,17 @@ where service_ctx: &mut ServiceContext, members: &[IdentIdRef], ) -> Result<(), ChatError> { + // Dedup the requested signers up front: an account can resolve to the + // same signer twice, or a caller can repeat one, and a duplicate would + // otherwise cost a redundant key-package fetch here and a second Add + // proposal for a member already being added in this batch. + let mut seen = std::collections::HashSet::new(); + let members: Vec = members + .iter() + .copied() + .filter(|m| seen.insert(m.as_str().to_string())) + .collect(); + // Fetch and validate every key package before proposing any add, so a // member with no key package fails the call before it opens proposals // for the others. Members are signer ids; the de-mls member id must @@ -268,7 +279,7 @@ where // credential), so it is read from the fetched key package rather than // assumed equal to the signer id. let mut invites = Vec::with_capacity(members.len()); - for member in members { + for member in &members { let kp_bytes = service_ctx .registry .retrieve(member.as_str()) @@ -285,16 +296,22 @@ where invites.push((member_id, member.to_string(), kp_bytes)); } - // Record who WE invited before touching the conversation: after_op - // forwards a welcome only to joiners in pending_invites. Skip a member - // de-mls would silently not propose (self, or already in the group) — - // recording it would strand a pending invite that later matches a - // re-join and fires a duplicate welcome. + // pending_invites drives welcome delivery: after_op forwards a welcome + // only to a joiner recorded here. Record a member only if de-mls will + // actually propose its add — recording one it silently drops strands an + // entry that a later re-join can match, firing a spurious duplicate + // welcome. de-mls drops self and members already in the group; and since + // add_member only opens a proposal, the committed roster won't reflect a + // member added earlier in this same loop, so the set tracks those too. + // Seed it with the roster and self, insert as we go, and one check + // covers all three. + let mut roster: std::collections::HashSet> = + self.conversation.members()?.into_iter().collect(); + roster.insert(self.conversation.member_id_bytes().to_vec()); + let mut result = Ok(()); for (member_id, signer_id, kp_bytes) in invites { - if member_id == self.conversation.member_id_bytes() - || self.conversation.members()?.contains(&member_id) - { + if !roster.insert(member_id.clone()) { continue; } self.pending_invites.push((member_id.clone(), signer_id)); From 464029a61ddba0f53582e15c9d665fc6be9890e1 Mon Sep 17 00:00:00 2001 From: osmaczko <33099791+osmaczko@users.noreply.github.com> Date: Wed, 8 Jul 2026 19:02:10 +0200 Subject: [PATCH 8/9] docs: correct the retry-budget and group-add doc comments The retry-budget comment claimed the ~20s init IPC budget held even at the worst-case sum, but that only holds on the load-shed path where each retry returns fast; a fully unreachable registry costs up to MAX_RETRIES times the reqwest timeout, which no retry budget can rescue. State both. Reword add_group_members to name the proposal, commit, and welcome flow rather than the unexplained "once the add commits". --- crates/generic-chat/src/client.rs | 7 ++++--- extensions/components/src/contact_registry/http.rs | 7 +++++-- 2 files changed, 9 insertions(+), 5 deletions(-) diff --git a/crates/generic-chat/src/client.rs b/crates/generic-chat/src/client.rs index 2d2b93d..ad6a0f8 100644 --- a/crates/generic-chat/src/client.rs +++ b/crates/generic-chat/src/client.rs @@ -186,9 +186,10 @@ where .map_err(Into::into) } - /// Add accounts' devices to an existing group conversation. Membership - /// changes are agreed and committed by the group asynchronously; the - /// joiners' welcomes go out once the add commits. + /// Add accounts' devices to an existing group conversation. The add is + /// staged as an MLS proposal and merged by the group's next commit (driven + /// asynchronously by the wakeup loop); each joiner's welcome is sent when + /// that commit lands, not when this call returns. pub fn add_group_members( &mut self, convo_id: &str, diff --git a/extensions/components/src/contact_registry/http.rs b/extensions/components/src/contact_registry/http.rs index a1cbca9..0cf5987 100644 --- a/extensions/components/src/contact_registry/http.rs +++ b/extensions/components/src/contact_registry/http.rs @@ -268,8 +268,11 @@ fn decode_payload(payload: &[u8]) -> Option<(u64, &[u8])> { /// Retry budget for the registry's transient, load-induced 5xx/429 responses. /// The service is reliable request-by-request but sheds concurrent bursts, so a -/// few backed-off retries let a request land once the burst clears. Sized to -/// stay within chat_module's ~20s init IPC budget even at the worst-case sum. +/// few backed-off retries let a request land once the burst clears. On that path +/// each retry returns fast, so the added cost is the ~3s worst-case backoff sum, +/// well inside chat_module's ~20s init IPC budget. A fully unreachable registry +/// instead costs up to MAX_RETRIES times the reqwest timeout, which no retry +/// budget can rescue. const MAX_RETRIES: u32 = 4; const RETRY_BASE_MS: u64 = 200; const RETRY_MAX_BACKOFF_MS: u64 = 2000; From 1877e9b476f43e54d0fc133580e86005e8c33fbb Mon Sep 17 00:00:00 2001 From: osmaczko <33099791+osmaczko@users.noreply.github.com> Date: Thu, 9 Jul 2026 19:10:59 +0200 Subject: [PATCH 9/9] chore: allow clippy::question_mark in LocalBroadcaster::poll (Rust 1.97 FP) Stable rolled to 1.97, whose clippy question_mark flags poll()'s match on `self.shared.borrow().read(next)`. Its suggested `read(next)?` would drop the RefCell Ref guard and dangle the returned reference, so the lint is a false positive here. CI tracks floating stable (`rustup update stable`), so this is pre-existing code newly flagged; suppress it to keep the branch green. --- extensions/components/src/delivery/local_broadcaster.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/extensions/components/src/delivery/local_broadcaster.rs b/extensions/components/src/delivery/local_broadcaster.rs index f47807c..5034408 100644 --- a/extensions/components/src/delivery/local_broadcaster.rs +++ b/extensions/components/src/delivery/local_broadcaster.rs @@ -69,6 +69,10 @@ impl LocalBroadcaster { /// Pulls all messages this consumer has not yet seen on `address`, /// applying any registered filter. Advances the cursor so the same /// messages are not returned again. + // clippy's question_mark (1.97+) wants `self.shared.borrow().read(next)?`, but + // `read` returns a reference into the RefCell `Ref`; the `?` form drops that + // guard at the `;` and `ae` would dangle. Keep the explicit match. + #[allow(clippy::question_mark)] pub fn poll(&mut self) -> Option> { loop { let next = self.cursor;