diff --git a/core/conversations/src/conversation.rs b/core/conversations/src/conversation.rs index 4fa07a7..34e568c 100644 --- a/core/conversations/src/conversation.rs +++ b/core/conversations/src/conversation.rs @@ -34,7 +34,9 @@ pub(crate) trait Convo: Identified + Send { enc: EncryptedPayload, ) -> Result; - fn wakeup(&mut self, service_ctx: &mut ServiceContext) -> 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) -> Result; } /// Group-only operations. diff --git a/core/conversations/src/conversation/direct_v1.rs b/core/conversations/src/conversation/direct_v1.rs index f75bdb0..8a6cb01 100644 --- a/core/conversations/src/conversation/direct_v1.rs +++ b/core/conversations/src/conversation/direct_v1.rs @@ -55,7 +55,10 @@ where self.inner_group.handle_frame(cx, enc) } - fn wakeup(&mut self, service_ctx: &mut ServiceContext) -> Result<(), ChatError> { + fn wakeup( + &mut self, + service_ctx: &mut ServiceContext, + ) -> Result { self.inner_group.wakeup(service_ctx) } } diff --git a/core/conversations/src/conversation/group_v1.rs b/core/conversations/src/conversation/group_v1.rs index dbea42a..6d6b64a 100644 --- a/core/conversations/src/conversation/group_v1.rs +++ b/core/conversations/src/conversation/group_v1.rs @@ -288,11 +288,12 @@ impl Convo for GroupV1Convo { Ok(ConvoOutcome { convo_id: self.id().to_string(), content, + members_changed: false, }) } - fn wakeup(&mut self, _: &mut ServiceContext) -> Result<(), ChatError> { - Ok(()) + fn wakeup(&mut self, _: &mut ServiceContext) -> Result { + Ok(ConvoOutcome::empty(self.id().to_string())) } } diff --git a/core/conversations/src/conversation/group_v2.rs b/core/conversations/src/conversation/group_v2.rs index 8636167..c547e5a 100644 --- a/core/conversations/src/conversation/group_v2.rs +++ b/core/conversations/src/conversation/group_v2.rs @@ -28,7 +28,7 @@ use prost::Message; use shared_traits::{IdentId, IdentIdRef}; use std::sync::Arc; use std::time::{Duration, SystemTime, UNIX_EPOCH}; -use tracing::{info, instrument, warn}; +use tracing::{info, instrument}; use crate::IdentityProvider; use crate::conversation::{ConversationIdRef, ExternalServices, ServiceContext}; @@ -275,28 +275,21 @@ where self.conversation .poll(&service_ctx.mls_provider, &service_ctx.mls_identity); let events = self.after_op(service_ctx)?; // route + publish + re-arm, returns 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())) - } - } + Ok(self.outcome_from_events(&events)) } #[instrument(name = "groupv2.wakeup", skip_all, fields(user_id = %ctx.mls_identity.display_name()))] - fn wakeup(&mut self, ctx: &mut ServiceContext) -> Result<(), ChatError> { + fn wakeup(&mut self, ctx: &mut ServiceContext) -> Result { info!(convo = %self.convo_id, "Wakeup"); - let outcome = self.conversation.poll(&ctx.mls_provider, &ctx.mls_identity); - if outcome.leave_requested { + let poll_outcome = self.conversation.poll(&ctx.mls_provider, &ctx.mls_identity); + if poll_outcome.leave_requested { // Commit ejected us (or join expired). Real handling - drops // this convo from its map; tracing::warn!(convo = %self.convo_id, "conversation requested teardown"); } - self.after_op(ctx)?; // publish what poll produced + re-arm alarm - Ok(()) + let events = self.after_op(ctx)?; // publish what poll produced + re-arm alarm + Ok(self.outcome_from_events(&events)) } } @@ -472,19 +465,27 @@ impl GroupV2Convo { Ok(events) } - fn events_to_content(&self, events: &[ConversationEvent]) -> Option { - events.iter().find_map(|evt| match evt { + fn outcome_from_events(&self, events: &[ConversationEvent]) -> ConvoOutcome { + let content = events.iter().find_map(|evt| match evt { ConversationEvent::ConversationMessage(AppMessageProto { payload: Some(app_message::Payload::ConversationMessage(cm)), - }) => Some(ConvoOutcome { - convo_id: self.convo_id.clone(), - content: Some(Content { - bytes: cm.message.clone(), - encoded_credential: cm.sender.clone(), - }), + }) => Some(Content { + bytes: cm.message.clone(), + encoded_credential: cm.sender.clone(), }), _ => None, - }) + }); + let members_changed = events.iter().any(|evt| { + matches!( + evt, + ConversationEvent::CommitApplied(_) | ConversationEvent::WelcomeReady { .. } + ) + }); + ConvoOutcome { + convo_id: self.convo_id.clone(), + content, + members_changed, + } } } diff --git a/core/conversations/src/conversation/privatev1.rs b/core/conversations/src/conversation/privatev1.rs index a8a9054..e8dea97 100644 --- a/core/conversations/src/conversation/privatev1.rs +++ b/core/conversations/src/conversation/privatev1.rs @@ -276,11 +276,12 @@ impl Convo for PrivateV1Convo { Ok(ConvoOutcome { convo_id: self.id().to_string(), content, + members_changed: false, }) } - fn wakeup(&mut self, _: &mut ServiceContext) -> Result<(), ChatError> { - Ok(()) + fn wakeup(&mut self, _: &mut ServiceContext) -> Result { + Ok(ConvoOutcome::empty(self.id().to_string())) } } diff --git a/core/conversations/src/core.rs b/core/conversations/src/core.rs index 0934f99..414e703 100644 --- a/core/conversations/src/core.rs +++ b/core/conversations/src/core.rs @@ -442,18 +442,18 @@ impl<'a, S: ExternalServices + 'static> Core { } } - pub fn wakeup(&mut self, convo_id: ConversationIdRef) -> Result<(), ChatError> { + pub fn wakeup(&mut self, convo_id: ConversationIdRef) -> Result { info!(convos = ?self.cached_convos.keys().collect::>(), id = ?self.services.mls_identity.id(), "Cached Convos"); match convo_id { c if c == self.pq_inbox.id() => todo!(), - c if self.cached_convos.contains_key(c) => self.wakeup_convo(c), - _ => Ok(()), + c if self.cached_convos.contains_key(c) => self.wakeup_convo(c).map(Into::into), + _ => Ok(PayloadOutcome::Empty), } } // 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 { let Some(convo) = self.cached_convos.get_mut(convo_id) else { return Err(ChatError::generic("No Convo Found")); }; @@ -592,7 +592,7 @@ impl Convo for ConvoTypeOwned { } } - fn wakeup(&mut self, service_ctx: &mut ServiceContext) -> Result<(), ChatError> { + fn wakeup(&mut self, service_ctx: &mut ServiceContext) -> Result { match self { ConvoTypeOwned::Group(group_convo) => group_convo.wakeup(service_ctx), ConvoTypeOwned::Direct(convo) => convo.wakeup(service_ctx), diff --git a/core/conversations/src/outcomes.rs b/core/conversations/src/outcomes.rs index 3f77294..afa6e6f 100644 --- a/core/conversations/src/outcomes.rs +++ b/core/conversations/src/outcomes.rs @@ -1,7 +1,7 @@ //! Observations a single inbound payload produces. //! //! - [`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 //! initial [`ConvoOutcome`]. //! - [`PayloadOutcome`] — the union of the above, plus `Empty`. @@ -22,6 +22,7 @@ pub struct Content { pub struct ConvoOutcome { pub convo_id: ConversationId, pub content: Option, + pub members_changed: bool, } impl ConvoOutcome { @@ -29,6 +30,7 @@ impl ConvoOutcome { Self { convo_id, content: None, + members_changed: false, } } } diff --git a/crates/generic-chat/src/client.rs b/crates/generic-chat/src/client.rs index ad6a0f8..213fff5 100644 --- a/crates/generic-chat/src/client.rs +++ b/crates/generic-chat/src/client.rs @@ -333,8 +333,18 @@ fn worker_loop( let Ok(WakeupEvent { convo_id }) = msg else { return; // wakeup service's sender dropped }; - if let Err(e) = core.lock().wakeup(&convo_id) { - tracing::warn!("wakeup failed: {e:?}"); + // A wakeup can drive the steward's own commit, so it yields events too. + 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, @@ -502,18 +512,26 @@ fn dedup_members(members: impl IntoIterator) -> Vec Vec { - let ConvoOutcome { convo_id, content } = outcome; - content - .and_then(|c| { - let sender = decode_sender(directory, &c.encoded_credential).ok()?; - Some(Event::MessageReceived { - convo_id: Arc::from(convo_id), - content: c.bytes, - sender, - }) - }) - .into_iter() - .collect() + let ConvoOutcome { + convo_id, + content, + members_changed, + } = outcome; + let convo_id: Arc = Arc::from(convo_id); + let mut events = Vec::new(); + if let Some(c) = content + && let Ok(sender) = decode_sender(directory, &c.encoded_credential) + { + events.push(Event::MessageReceived { + 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 { diff --git a/crates/generic-chat/src/event.rs b/crates/generic-chat/src/event.rs index eaf5dbe..4618e02 100644 --- a/crates/generic-chat/src/event.rs +++ b/crates/generic-chat/src/event.rs @@ -38,6 +38,10 @@ pub enum Event { content: Vec, sender: MessageSender, }, + /// A commit changed a conversation's membership. + ConversationMembersChanged { + convo_id: Arc, + }, InboundError { message: String, },