diff --git a/core/conversations/src/conversation/group_v2.rs b/core/conversations/src/conversation/group_v2.rs index d37b2be..95951bb 100644 --- a/core/conversations/src/conversation/group_v2.rs +++ b/core/conversations/src/conversation/group_v2.rs @@ -5,6 +5,7 @@ use crate::conversation::mls_extensions::{ ConvoMetaInfo, GROUP_METADATA_EXTENSION_TYPE, capabilities_with_group_metadata, }; +use crate::group_v2_status::GroupV2StatusKind; use crate::types::{AddressedEncryptedPayload, ConvoMetadata}; use crate::{Content, WakeupService}; use alloy::signers::local::PrivateKeySigner; @@ -469,7 +470,28 @@ impl GroupV2Convo { } } - // 2. Publish + // 2. Record what the conversation said about running itself, so a + // client can surface a commit round that is missing candidates or a + // step that did not go through. + for evt in &events { + let kind = match evt { + ConversationEvent::PhaseChange(state) => GroupV2StatusKind::Phase(*state), + ConversationEvent::CommitRoundProgress { received, expected } => { + GroupV2StatusKind::CommitRound { + received: *received, + expected: *expected, + } + } + ConversationEvent::Error { operation, message } => GroupV2StatusKind::Failed { + operation: operation.clone(), + message: message.clone(), + }, + _ => continue, + }; + service_ctx.group_v2_status.record(&self.convo_id, kind); + } + + // 3. Publish for out in outbound { let frame = GroupV2Frame { payload: Some(GroupV2Payload::DeMlsWrapper(out.payload.into())), @@ -489,7 +511,7 @@ impl GroupV2Convo { .map_err(ChatError::generic)?; } - // 3. Re-arm the alarm with the conversation's earliest deadline. + // 4. Re-arm the alarm with the conversation's earliest deadline. if let Some(d) = wakeup { service_ctx .wakeup_service diff --git a/core/conversations/src/core.rs b/core/conversations/src/core.rs index f6003f1..c1cea99 100644 --- a/core/conversations/src/core.rs +++ b/core/conversations/src/core.rs @@ -2,6 +2,7 @@ use crate::causal_history::{CausalHistoryStore, DeliveryAck, MissingMessage}; use crate::conversation::{ ConversationIdRef, DirectV1Convo, GroupV1Convo, GroupV2Convo, Identified, MessageId, }; +use crate::group_v2_status::{GroupV2Status, GroupV2StatusStore}; use crate::service_context::{ExternalServices, ServiceContext}; use crate::types::ConvoMetadata; use crate::{ @@ -146,6 +147,7 @@ where mls_identity, mls_provider, causal, + group_v2_status: GroupV2StatusStore::default(), identity, wakeup_service, demls_clock: GroupV2Clock::default(), @@ -335,6 +337,13 @@ impl<'a, S: ExternalServices + 'static> Core { self.services.causal.take_acks() } + /// Drain what the GroupV2 conversations reported about running themselves + /// since the last call: phase changes, commit-round progress, and steps of + /// their own that did not go through. + pub fn take_group_v2_status(&self) -> Vec { + self.services.group_v2_status.take() + } + /// Encrypt and publish `content` to an existing conversation, returning the /// id assigned to the message so later acknowledgements can be matched to /// it. diff --git a/core/conversations/src/group_v2_status.rs b/core/conversations/src/group_v2_status.rs new file mode 100644 index 0000000..bbf2388 --- /dev/null +++ b/core/conversations/src/group_v2_status.rs @@ -0,0 +1,55 @@ +//! What a GroupV2 conversation reports about running itself. +//! +//! de-mls narrates its own commit-and-recovery cycle alongside the messages it +//! decrypts. None of it is content and none of it needs acting on, but it is +//! the only account of why a group is or is not moving, so it is buffered here +//! and drained by the client the same way causal-history observations are. + +use std::cell::RefCell; + +use crate::core::ConversationId; + +/// The phase of a GroupV2 conversation's commit-and-recovery cycle. +pub use de_mls::ConversationState as GroupV2Phase; + +/// One report from a GroupV2 conversation. +#[derive(Debug, Clone)] +pub struct GroupV2Status { + pub convo_id: ConversationId, + pub kind: GroupV2StatusKind, +} + +#[derive(Debug, Clone)] +pub enum GroupV2StatusKind { + /// The conversation entered a new phase. + Phase(GroupV2Phase), + /// `received` of `expected` stewards' commit candidates have arrived for + /// the round in progress; reported again whenever the count changes. A + /// round that ends with fewer than it expected is one where the members + /// chose from different sets of candidates. + CommitRound { received: usize, expected: usize }, + /// A step the conversation was carrying out on its own, such as submitting + /// a vote, did not go through. The conversation stays usable. + Failed { operation: String, message: String }, +} + +/// Session-scoped buffer for [`GroupV2Status`], shared through +/// `ServiceContext` because conversations are rebuilt from storage on every +/// inbound payload and cannot hold it themselves. +#[derive(Debug, Default)] +pub(crate) struct GroupV2StatusStore { + reports: RefCell>, +} + +impl GroupV2StatusStore { + pub(crate) fn record(&self, convo_id: &str, kind: GroupV2StatusKind) { + self.reports.borrow_mut().push(GroupV2Status { + convo_id: convo_id.to_owned(), + kind, + }); + } + + pub(crate) fn take(&self) -> Vec { + std::mem::take(&mut self.reports.borrow_mut()) + } +} diff --git a/core/conversations/src/lib.rs b/core/conversations/src/lib.rs index ca319fb..273228d 100644 --- a/core/conversations/src/lib.rs +++ b/core/conversations/src/lib.rs @@ -2,6 +2,7 @@ mod causal_history; mod conversation; mod core; mod errors; +mod group_v2_status; mod inbox_v2; mod outcomes; mod proto; @@ -24,6 +25,7 @@ pub use core::{ConversationId, Core}; pub use de_mls::ConversationConfig as GroupV2Config; pub use de_mls::MockClock; pub use errors::ChatError; +pub use group_v2_status::{GroupV2Phase, GroupV2Status, GroupV2StatusKind}; 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 055d4a5..8976453 100644 --- a/core/conversations/src/service_context.rs +++ b/core/conversations/src/service_context.rs @@ -6,6 +6,7 @@ use storage::ChatStore; use crate::IdentityProvider; use crate::causal_history::CausalHistoryStore; use crate::conversation::GroupV2Clock; +use crate::group_v2_status::GroupV2StatusStore; use crate::inbox_v2::{MlsEphemeralPqProvider, MlsIdentityProvider}; use crate::service_traits::WakeupService; use crate::{DeliveryService, RegistrationService}; @@ -43,6 +44,7 @@ pub(crate) struct ServiceContext { pub(crate) mls_identity: MlsIdentityProvider, pub(crate) mls_provider: MlsEphemeralPqProvider, pub(crate) causal: CausalHistoryStore, + pub(crate) group_v2_status: GroupV2StatusStore, pub(crate) identity: Identity, pub(crate) wakeup_service: S::WS, /// Time source for GroupV2 (de-mls) conversations. diff --git a/crates/generic-chat/src/client.rs b/crates/generic-chat/src/client.rs index 27006ad..6931585 100644 --- a/crates/generic-chat/src/client.rs +++ b/crates/generic-chat/src/client.rs @@ -7,8 +7,8 @@ use crossbeam_channel::{Receiver, Sender, select}; use crypto::Ed25519VerifyingKey; use libchat::{ ConversationId, ConvoMetadata, ConvoOutcome, Core, DeliveryAck, DeliveryService, GroupV2Config, - IdentId, IdentIdRef, InboxOutcome, MessageId, MissingMessage, PayloadOutcome, - RegistrationService, + GroupV2Status, GroupV2StatusKind, IdentId, IdentIdRef, InboxOutcome, MessageId, MissingMessage, + PayloadOutcome, RegistrationService, }; use logos_account::{AccountDirectory, resolve_device_ids}; use parking_lot::Mutex; @@ -366,6 +366,7 @@ fn worker_loop( }; events.extend(delivery_ack_events(core.take_acks(), &directory)); events.extend(missing_events(core.take_missing_messages(), &directory)); + events.extend(group_v2_status_events(core.take_group_v2_status())); events }; for event in events { @@ -390,6 +391,7 @@ fn worker_loop( }; events.extend(delivery_ack_events(core.take_acks(), &directory)); events.extend(missing_events(core.take_missing_messages(), &directory)); + events.extend(group_v2_status_events(core.take_group_v2_status())); events }; for event in events { @@ -448,6 +450,38 @@ fn missing_events(missing: Vec, directory: &impl AccountDirector .collect() } +/// Map what the GroupV2 conversations reported about running themselves onto +/// [`Event::ConversationPhaseChanged`], [`Event::CommitRoundProgress`] and +/// [`Event::ConversationError`]. +/// +/// Drained after each drive of the core, so these narrate the drive that +/// produced the batch they arrive with. +fn group_v2_status_events(reports: Vec) -> Vec { + reports + .into_iter() + .map(|report| { + let convo_id = Arc::from(report.convo_id); + match report.kind { + GroupV2StatusKind::Phase(phase) => { + Event::ConversationPhaseChanged { convo_id, phase } + } + GroupV2StatusKind::CommitRound { received, expected } => { + Event::CommitRoundProgress { + convo_id, + received, + expected, + } + } + GroupV2StatusKind::Failed { operation, message } => Event::ConversationError { + convo_id, + operation, + message, + }, + } + }) + .collect() +} + /// Resolve a participant a causal-history observation named — the author of a /// message we never saw, or the peer acknowledging one of ours. /// diff --git a/crates/generic-chat/src/event.rs b/crates/generic-chat/src/event.rs index 5e2e4a8..9fa548b 100644 --- a/crates/generic-chat/src/event.rs +++ b/crates/generic-chat/src/event.rs @@ -8,7 +8,7 @@ use std::sync::Arc; -use libchat::{ConversationClass, IdentId}; +use libchat::{ConversationClass, GroupV2Phase, IdentId}; /// The sender of a received message, recovered from its credential. /// @@ -72,6 +72,29 @@ pub enum Event { ConversationMembersChanged { convo_id: Arc, }, + /// A GroupV2 conversation entered a new phase of its commit-and-recovery + /// cycle. Nothing needs acting on, but a conversation parked outside + /// `Working` is one that is accepting neither messages nor members. + ConversationPhaseChanged { + convo_id: Arc, + phase: GroupV2Phase, + }, + /// `received` of `expected` stewards' commit candidates have arrived for + /// the commit round in progress, reported again whenever the count + /// changes. A round that ends short of `expected` is one where members + /// chose from different sets of candidates. + CommitRoundProgress { + convo_id: Arc, + received: usize, + expected: usize, + }, + /// A step a GroupV2 conversation was carrying out on its own, such as + /// submitting a vote, did not go through. The conversation stays usable. + ConversationError { + convo_id: Arc, + operation: String, + message: String, + }, InboundError { message: String, }, diff --git a/crates/generic-chat/src/lib.rs b/crates/generic-chat/src/lib.rs index 7156ca7..6171532 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, ConvoMetadata, - DeliveryService, GroupV2Config, IdentityProvider, MessageId, RegistrationService, + DeliveryService, GroupV2Config, GroupV2Phase, IdentityProvider, MessageId, RegistrationService, StorageConfig, }; // The directory trait bounds ChatClient's registry parameter, so callers diff --git a/crates/generic-chat/tests/group_v2.rs b/crates/generic-chat/tests/group_v2.rs index 447f70a..cd7c60d 100644 --- a/crates/generic-chat/tests/group_v2.rs +++ b/crates/generic-chat/tests/group_v2.rs @@ -12,7 +12,7 @@ use libchat::ChatStorage; use logos_account::TestLogosAccount; use logos_generic_chat::{ ChatClient, ChatClientBuilder, ConversationClass, DelegateSigner, Event, GroupMetadata, - GroupV2Config, InProcessDelivery, MessageBus, + GroupV2Config, GroupV2Phase, InProcessDelivery, MessageBus, }; /// Metadata for a group these tests create without a name or description. @@ -539,3 +539,51 @@ fn a_sent_message_is_acknowledged_by_the_peers_that_reply() { expected.sort(); assert_eq!(holders, expected, "both replying peers should be listed"); } + +/// The commit-and-recovery cycle reaches the application. +/// +/// A group that stops moving is otherwise silent: the roster keeps reporting +/// whatever it last committed, and the account of why sits in de-mls's own log. +/// Adding a member takes the creator through a freeze and a selection, so its +/// channel has to carry them. +#[test] +fn group_v2_phase_changes_reach_the_application() { + 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()); + + // An empty group and then an add, rather than a group created around its + // members: only the add runs a commit round. + let convo_id = saro + .create_group_conversation(&[], unnamed_group()) + .expect("saro create group"); + saro.add_group_members(&convo_id, &[&raya_addr]) + .expect("saro add raya"); + + // A conversation opens in `Working`, so the phases worth seeing are the + // ones the commit goes through: the freeze that collects candidates, and + // the selection that picks one. + let mut seen = Vec::new(); + wait_for_event( + &saro_events, + "saro selecting a commit candidate", + Duration::from_secs(10), + |e| match e { + Event::ConversationPhaseChanged { + convo_id: id, + phase, + } if id.as_ref() == convo_id => { + seen.push(*phase); + (*phase == GroupV2Phase::Selection).then_some(()) + } + _ => None, + }, + ); + + assert!( + seen.contains(&GroupV2Phase::Freezing), + "the freeze that minted the commit went unreported :: {seen:?}" + ); +}