feat: surface a roster-changed event when group membership changes (#177)

A group's roster changed silently. An add merges on the steward's commit-inactivity timer, and the members that receive its commit apply it, but neither surfaced any observation: Core::wakeup returned () and the client worker discarded it, and GroupV2Convo mapped only chat messages into a ConvoOutcome, so a commit produced an empty outcome. An app could learn a group grew only by re-selecting the conversation, a manual refresh, or a later message from the new member.

de-mls already reports the change: CommitApplied (adds and removes) and WelcomeReady (adds) both fire, on every member, when a commit merges. Drain them into the observation and emit a new event.

- ConvoOutcome gains members_changed, set by GroupV2Convo when a poll cycle's drained de-mls events include CommitApplied or WelcomeReady. It rides alongside content, the same way a protocol-only frame already yields content: None.
- Convo::wakeup returns a ConvoOutcome instead of (), mirroring handle_frame, so the steward's own timer-driven commit is observable. Kinds with no timers return ConvoOutcome::empty; Core::wakeup and the worker translate it through the same events_from_inbound path inbound payloads use.
- New Event::ConversationMembersChanged { convo_id }: the app re-fetches group_members. Fires on every member a commit reaches, so the inviter sees its own add land and existing members see later joins.
This commit is contained in:
osmaczko 2026-07-15 17:21:23 +02:00 committed by GitHub
parent f5e877b6e1
commit b0de532199
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 81 additions and 49 deletions

View File

@ -34,7 +34,9 @@ pub(crate) trait Convo<S: ExternalServices>: Identified + Send {
enc: EncryptedPayload, enc: EncryptedPayload,
) -> Result<ConvoOutcome, ChatError>; ) -> Result<ConvoOutcome, ChatError>;
fn wakeup(&mut self, service_ctx: &mut ServiceContext<S>) -> Result<(), ChatError>; /// Advances any time-driven protocol work (de-mls consensus deadlines) and
/// reports what it observed, mirroring [`Self::handle_frame`].
fn wakeup(&mut self, service_ctx: &mut ServiceContext<S>) -> Result<ConvoOutcome, ChatError>;
} }
/// Group-only operations. /// Group-only operations.

View File

@ -55,7 +55,10 @@ where
self.inner_group.handle_frame(cx, enc) self.inner_group.handle_frame(cx, enc)
} }
fn wakeup(&mut self, service_ctx: &mut ServiceContext<S>) -> Result<(), ChatError> { fn wakeup(
&mut self,
service_ctx: &mut ServiceContext<S>,
) -> Result<crate::ConvoOutcome, ChatError> {
self.inner_group.wakeup(service_ctx) self.inner_group.wakeup(service_ctx)
} }
} }

View File

@ -288,11 +288,12 @@ impl<S: ExternalServices> Convo<S> for GroupV1Convo {
Ok(ConvoOutcome { Ok(ConvoOutcome {
convo_id: self.id().to_string(), convo_id: self.id().to_string(),
content, content,
members_changed: false,
}) })
} }
fn wakeup(&mut self, _: &mut ServiceContext<S>) -> Result<(), ChatError> { fn wakeup(&mut self, _: &mut ServiceContext<S>) -> Result<ConvoOutcome, ChatError> {
Ok(()) Ok(ConvoOutcome::empty(self.id().to_string()))
} }
} }

View File

@ -28,7 +28,7 @@ use prost::Message;
use shared_traits::{IdentId, IdentIdRef}; use shared_traits::{IdentId, IdentIdRef};
use std::sync::Arc; use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH}; use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tracing::{info, instrument, warn}; use tracing::{info, instrument};
use crate::IdentityProvider; use crate::IdentityProvider;
use crate::conversation::{ConversationIdRef, ExternalServices, ServiceContext}; use crate::conversation::{ConversationIdRef, ExternalServices, ServiceContext};
@ -275,28 +275,21 @@ where
self.conversation self.conversation
.poll(&service_ctx.mls_provider, &service_ctx.mls_identity); .poll(&service_ctx.mls_provider, &service_ctx.mls_identity);
let events = self.after_op(service_ctx)?; // route + publish + re-arm, returns events let events = self.after_op(service_ctx)?; // route + publish + re-arm, returns events
Ok(self.outcome_from_events(&events))
match self.events_to_content(&events) {
Some(o) => Ok(o),
None => {
warn!("returning None as ConvoOutcome");
Ok(ConvoOutcome::empty(self.convo_id.to_string()))
}
}
} }
#[instrument(name = "groupv2.wakeup", skip_all, fields(user_id = %ctx.mls_identity.display_name()))] #[instrument(name = "groupv2.wakeup", skip_all, fields(user_id = %ctx.mls_identity.display_name()))]
fn wakeup(&mut self, ctx: &mut ServiceContext<S>) -> Result<(), ChatError> { fn wakeup(&mut self, ctx: &mut ServiceContext<S>) -> Result<ConvoOutcome, ChatError> {
info!(convo = %self.convo_id, "Wakeup"); info!(convo = %self.convo_id, "Wakeup");
let outcome = self.conversation.poll(&ctx.mls_provider, &ctx.mls_identity); let poll_outcome = self.conversation.poll(&ctx.mls_provider, &ctx.mls_identity);
if outcome.leave_requested { if poll_outcome.leave_requested {
// Commit ejected us (or join expired). Real handling - drops // Commit ejected us (or join expired). Real handling - drops
// this convo from its map; // this convo from its map;
tracing::warn!(convo = %self.convo_id, "conversation requested teardown"); tracing::warn!(convo = %self.convo_id, "conversation requested teardown");
} }
self.after_op(ctx)?; // publish what poll produced + re-arm alarm let events = self.after_op(ctx)?; // publish what poll produced + re-arm alarm
Ok(()) Ok(self.outcome_from_events(&events))
} }
} }
@ -472,19 +465,27 @@ impl GroupV2Convo {
Ok(events) Ok(events)
} }
fn events_to_content(&self, events: &[ConversationEvent]) -> Option<ConvoOutcome> { fn outcome_from_events(&self, events: &[ConversationEvent]) -> ConvoOutcome {
events.iter().find_map(|evt| match evt { let content = events.iter().find_map(|evt| match evt {
ConversationEvent::ConversationMessage(AppMessageProto { ConversationEvent::ConversationMessage(AppMessageProto {
payload: Some(app_message::Payload::ConversationMessage(cm)), payload: Some(app_message::Payload::ConversationMessage(cm)),
}) => Some(ConvoOutcome { }) => Some(Content {
convo_id: self.convo_id.clone(), bytes: cm.message.clone(),
content: Some(Content { encoded_credential: cm.sender.clone(),
bytes: cm.message.clone(),
encoded_credential: cm.sender.clone(),
}),
}), }),
_ => None, _ => None,
}) });
let members_changed = events.iter().any(|evt| {
matches!(
evt,
ConversationEvent::CommitApplied(_) | ConversationEvent::WelcomeReady { .. }
)
});
ConvoOutcome {
convo_id: self.convo_id.clone(),
content,
members_changed,
}
} }
} }

View File

@ -276,11 +276,12 @@ impl<S: ExternalServices> Convo<S> for PrivateV1Convo {
Ok(ConvoOutcome { Ok(ConvoOutcome {
convo_id: self.id().to_string(), convo_id: self.id().to_string(),
content, content,
members_changed: false,
}) })
} }
fn wakeup(&mut self, _: &mut ServiceContext<S>) -> Result<(), ChatError> { fn wakeup(&mut self, _: &mut ServiceContext<S>) -> Result<ConvoOutcome, ChatError> {
Ok(()) Ok(ConvoOutcome::empty(self.id().to_string()))
} }
} }

View File

@ -442,18 +442,18 @@ impl<'a, S: ExternalServices + 'static> Core<S> {
} }
} }
pub fn wakeup(&mut self, convo_id: ConversationIdRef) -> Result<(), ChatError> { pub fn wakeup(&mut self, convo_id: ConversationIdRef) -> Result<PayloadOutcome, ChatError> {
info!(convos = ?self.cached_convos.keys().collect::<Vec<_>>(), id = ?self.services.mls_identity.id(), "Cached Convos"); info!(convos = ?self.cached_convos.keys().collect::<Vec<_>>(), id = ?self.services.mls_identity.id(), "Cached Convos");
match convo_id { match convo_id {
c if c == self.pq_inbox.id() => todo!(), c if c == self.pq_inbox.id() => todo!(),
c if self.cached_convos.contains_key(c) => self.wakeup_convo(c), c if self.cached_convos.contains_key(c) => self.wakeup_convo(c).map(Into::into),
_ => Ok(()), _ => Ok(PayloadOutcome::Empty),
} }
} }
// Dispatch encrypted payload to its corresponding conversation // Dispatch encrypted payload to its corresponding conversation
fn wakeup_convo(&mut self, convo_id: ConversationIdRef) -> Result<(), ChatError> { fn wakeup_convo(&mut self, convo_id: ConversationIdRef) -> Result<ConvoOutcome, ChatError> {
let Some(convo) = self.cached_convos.get_mut(convo_id) else { let Some(convo) = self.cached_convos.get_mut(convo_id) else {
return Err(ChatError::generic("No Convo Found")); return Err(ChatError::generic("No Convo Found"));
}; };
@ -592,7 +592,7 @@ impl<S: ExternalServices> Convo<S> for ConvoTypeOwned<S> {
} }
} }
fn wakeup(&mut self, service_ctx: &mut ServiceContext<S>) -> Result<(), ChatError> { fn wakeup(&mut self, service_ctx: &mut ServiceContext<S>) -> Result<ConvoOutcome, ChatError> {
match self { match self {
ConvoTypeOwned::Group(group_convo) => group_convo.wakeup(service_ctx), ConvoTypeOwned::Group(group_convo) => group_convo.wakeup(service_ctx),
ConvoTypeOwned::Direct(convo) => convo.wakeup(service_ctx), ConvoTypeOwned::Direct(convo) => convo.wakeup(service_ctx),

View File

@ -1,7 +1,7 @@
//! Observations a single inbound payload produces. //! Observations a single inbound payload produces.
//! //!
//! - [`ConvoOutcome`] — an optional [`Content`] on a single existing //! - [`ConvoOutcome`] — an optional [`Content`] on a single existing
//! conversation. //! conversation, plus whether a commit changed its membership.
//! - [`InboxOutcome`] — a newly observed conversation, optionally with an //! - [`InboxOutcome`] — a newly observed conversation, optionally with an
//! initial [`ConvoOutcome`]. //! initial [`ConvoOutcome`].
//! - [`PayloadOutcome`] — the union of the above, plus `Empty`. //! - [`PayloadOutcome`] — the union of the above, plus `Empty`.
@ -22,6 +22,7 @@ pub struct Content {
pub struct ConvoOutcome { pub struct ConvoOutcome {
pub convo_id: ConversationId, pub convo_id: ConversationId,
pub content: Option<Content>, pub content: Option<Content>,
pub members_changed: bool,
} }
impl ConvoOutcome { impl ConvoOutcome {
@ -29,6 +30,7 @@ impl ConvoOutcome {
Self { Self {
convo_id, convo_id,
content: None, content: None,
members_changed: false,
} }
} }
} }

View File

@ -333,8 +333,18 @@ fn worker_loop<T, R, S: ChatStore + 'static>(
let Ok(WakeupEvent { convo_id }) = msg else { let Ok(WakeupEvent { convo_id }) = msg else {
return; // wakeup service's sender dropped return; // wakeup service's sender dropped
}; };
if let Err(e) = core.lock().wakeup(&convo_id) { // A wakeup can drive the steward's own commit, so it yields events too.
tracing::warn!("wakeup failed: {e:?}"); let events = match core.lock().wakeup(&convo_id) {
Ok(outcome) => events_from_inbound(outcome, &directory),
Err(e) => {
tracing::warn!("wakeup failed: {e:?}");
Vec::new()
}
};
for event in events {
if event_tx.send(event).is_err() {
return; // application dropped the receiver
}
} }
} }
recv(shutdown) -> _ => return, recv(shutdown) -> _ => return,
@ -502,18 +512,26 @@ fn dedup_members(members: impl IntoIterator<Item = GroupMember>) -> Vec<GroupMem
} }
fn convo_events(outcome: ConvoOutcome, directory: &impl AccountDirectory) -> Vec<Event> { fn convo_events(outcome: ConvoOutcome, directory: &impl AccountDirectory) -> Vec<Event> {
let ConvoOutcome { convo_id, content } = outcome; let ConvoOutcome {
content convo_id,
.and_then(|c| { content,
let sender = decode_sender(directory, &c.encoded_credential).ok()?; members_changed,
Some(Event::MessageReceived { } = outcome;
convo_id: Arc::from(convo_id), let convo_id: Arc<str> = Arc::from(convo_id);
content: c.bytes, let mut events = Vec::new();
sender, if let Some(c) = content
}) && let Ok(sender) = decode_sender(directory, &c.encoded_credential)
}) {
.into_iter() events.push(Event::MessageReceived {
.collect() convo_id: Arc::clone(&convo_id),
content: c.bytes,
sender,
});
}
if members_changed {
events.push(Event::ConversationMembersChanged { convo_id });
}
events
} }
fn inbox_events(outcome: InboxOutcome, directory: &impl AccountDirectory) -> Vec<Event> { fn inbox_events(outcome: InboxOutcome, directory: &impl AccountDirectory) -> Vec<Event> {

View File

@ -38,6 +38,10 @@ pub enum Event {
content: Vec<u8>, content: Vec<u8>,
sender: MessageSender, sender: MessageSender,
}, },
/// A commit changed a conversation's membership.
ConversationMembersChanged {
convo_id: Arc<str>,
},
InboundError { InboundError {
message: String, message: String,
}, },