Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 24 additions & 2 deletions core/conversations/src/conversation/group_v2.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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())),
Expand All @@ -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
Expand Down
9 changes: 9 additions & 0 deletions core/conversations/src/core.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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::{
Expand Down Expand Up @@ -146,6 +147,7 @@ where
mls_identity,
mls_provider,
causal,
group_v2_status: GroupV2StatusStore::default(),
identity,
wakeup_service,
demls_clock: GroupV2Clock::default(),
Expand Down Expand Up @@ -335,6 +337,13 @@ impl<'a, S: ExternalServices + 'static> Core<S> {
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<GroupV2Status> {
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.
Expand Down
55 changes: 55 additions & 0 deletions core/conversations/src/group_v2_status.rs
Original file line number Diff line number Diff line change
@@ -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<Vec<GroupV2Status>>,
}

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<GroupV2Status> {
std::mem::take(&mut self.reports.borrow_mut())
}
}
2 changes: 2 additions & 0 deletions core/conversations/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ mod causal_history;
mod conversation;
mod core;
mod errors;
mod group_v2_status;
mod inbox_v2;
mod outcomes;
mod proto;
Expand All @@ -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,
};
Expand Down
2 changes: 2 additions & 0 deletions core/conversations/src/service_context.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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};
Expand Down Expand Up @@ -43,6 +44,7 @@ pub(crate) struct ServiceContext<S: ExternalServices> {
pub(crate) mls_identity: MlsIdentityProvider<S::IP>,
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.
Expand Down
38 changes: 36 additions & 2 deletions crates/generic-chat/src/client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -366,6 +366,7 @@ fn worker_loop<T, R, S: ChatStore + 'static>(
};
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 {
Expand All @@ -390,6 +391,7 @@ fn worker_loop<T, R, S: ChatStore + 'static>(
};
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 {
Expand Down Expand Up @@ -448,6 +450,38 @@ fn missing_events(missing: Vec<MissingMessage>, 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<GroupV2Status>) -> Vec<Event> {
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.
///
Expand Down
25 changes: 24 additions & 1 deletion crates/generic-chat/src/event.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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.
///
Expand Down Expand Up @@ -72,6 +72,29 @@ pub enum Event {
ConversationMembersChanged {
convo_id: Arc<str>,
},
/// 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<str>,
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<str>,
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<str>,
operation: String,
message: String,
},
InboundError {
message: String,
},
Expand Down
2 changes: 1 addition & 1 deletion crates/generic-chat/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
50 changes: 49 additions & 1 deletion crates/generic-chat/tests/group_v2.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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:?}"
);
}