diff --git a/core/conversations/src/conversation.rs b/core/conversations/src/conversation.rs index 6f178924..7774ff43 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 3dcf96e4..8aa74024 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 03cd8b54..d2b91d96 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,14 +260,26 @@ 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. - for member in members { + // 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 + // 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 .retrieve(member.as_str()) @@ -298,17 +293,54 @@ 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)); + } + + // 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 !roster.insert(member_id.clone()) { + 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 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 { diff --git a/core/conversations/src/core.rs b/core/conversations/src/core.rs index ca1c19bf..bab9fad6 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() } @@ -293,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()?; @@ -304,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) } @@ -362,16 +395,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 8de43407..b67ceed4 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 @@ -95,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)?; @@ -112,12 +120,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 +142,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))) } } } @@ -183,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/conversations/src/lib.rs b/core/conversations/src/lib.rs index 654630e4..9eac46c6 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 82cf1d34..3f7372d8 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 09c84efa..9b758526 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/core/integration_tests_core/tests/test_group_v2.rs b/core/integration_tests_core/tests/test_group_v2.rs index f000423c..afafd0af 100644 --- a/core/integration_tests_core/tests/test_group_v2.rs +++ b/core/integration_tests_core/tests/test_group_v2.rs @@ -176,3 +176,75 @@ 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" + ); +} + +#[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" + ); +} diff --git a/crates/generic-chat/src/builder.rs b/crates/generic-chat/src/builder.rs index 6a663948..a1c8eac1 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 d425cd32..ad6a0f8f 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}; @@ -5,8 +6,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; @@ -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. @@ -73,6 +87,7 @@ where mut transport: T, reg: R, storage: S, + group_v2: Option, ) -> Result<(Self, Receiver), ClientError> { let inbound = transport.inbound(); @@ -80,7 +95,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 +169,55 @@ 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. 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, + 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) + } + + /// 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. /// @@ -193,6 +260,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 @@ -298,59 +378,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 @@ -397,7 +547,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 @@ -567,4 +720,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 39ea635a..6d7d14f7 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; @@ -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 00000000..42e2d3d9 --- /dev/null +++ b/crates/generic-chat/tests/group_v2.rs @@ -0,0 +1,338 @@ +//! 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, + }) +} + +/// 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)); + 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); + + // 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(), + 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()) + ); + + // 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 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] +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 +/// 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(_) + )); +} diff --git a/crates/generic-chat/tests/saro_and_raya.rs b/crates/generic-chat/tests/saro_and_raya.rs index 3ad15739..eca1b0e1 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), }); diff --git a/extensions/components/src/contact_registry/http.rs b/extensions/components/src/contact_registry/http.rs index e7955d5a..0cf59878 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,70 @@ 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. 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; + +/// 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 +389,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); + } + } } diff --git a/extensions/components/src/delivery/local_broadcaster.rs b/extensions/components/src/delivery/local_broadcaster.rs index f47807c0..5034408c 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;