xmtp_mls/groups/
summary.rs

1use chrono::Utc;
2use derive_builder::Builder;
3use openmls::group::GroupContext;
4use std::collections::{HashMap, HashSet};
5use xmtp_common::RetryableError;
6use xmtp_db::group_intent::IntentKind;
7use xmtp_proto::types::Cursor;
8
9use super::{GroupError, change_callbacks::AppDataChange, mls_sync::GroupMessageProcessingError};
10use xmtp_proto::types::GroupId;
11
12#[derive(Default)]
13pub struct SyncSummary {
14    pub(crate) publish_errors: Vec<GroupError>,
15    pub(crate) process: ProcessSummary,
16    pub(crate) post_commit_errors: Vec<GroupError>,
17    /// an error outside of the sync occurred
18    pub(crate) other: Option<Box<GroupError>>,
19}
20
21impl RetryableError for SyncSummary {
22    fn is_retryable(&self) -> bool {
23        self.publish_errors.iter().any(|e| e.is_retryable())
24            || self.post_commit_errors.iter().any(|e| e.is_retryable())
25            || self
26                .other
27                .as_ref()
28                .map(|s| s.is_retryable())
29                .unwrap_or(false)
30    }
31
32    /// The longest delay any rate-limited request of the sync asked for.
33    fn rate_limit_delay(&self) -> Option<xmtp_common::time::Duration> {
34        self.publish_errors
35            .iter()
36            .chain(self.post_commit_errors.iter())
37            .chain(self.other.as_deref())
38            .filter_map(RetryableError::rate_limit_delay)
39            .max()
40    }
41}
42
43impl crate::worker::NeedsDbReconnect for SyncSummary {
44    fn needs_db_reconnect(&self) -> bool {
45        self.publish_errors
46            .iter()
47            .chain(self.post_commit_errors.iter())
48            .chain(self.other.as_deref())
49            .any(crate::worker::NeedsDbReconnect::needs_db_reconnect)
50            || self
51                .process
52                .errored
53                .iter()
54                .any(|(_, error)| error.needs_db_reconnect())
55    }
56}
57
58impl SyncSummary {
59    /// synced a single message successfully
60    pub fn single(msg: MessageIdentifier) -> Self {
61        let mut process = ProcessSummary::default();
62        process.add(msg);
63        SyncSummary {
64            process,
65            ..Default::default()
66        }
67    }
68
69    /// Try to get a newly processed message by its cursor ID
70    pub fn new_message_by_id(&self, id: Cursor) -> Option<&MessageIdentifier> {
71        self.process.new_messages.iter().find(|m| m.cursor == id)
72    }
73
74    /// Every error in this summary: the error outside the sync, then the
75    /// publish, post-commit and message-processing errors, each in the order
76    /// they were recorded.
77    pub fn errors(&self) -> impl DoubleEndedIterator<Item = &(dyn std::error::Error + 'static)> {
78        self.other
79            .as_deref()
80            .into_iter()
81            .map(|error| error as &(dyn std::error::Error + 'static))
82            .chain(self.publish_errors.iter().map(|error| error as _))
83            .chain(self.post_commit_errors.iter().map(|error| error as _))
84            .chain(self.process.errored.iter().map(|(_, error)| error as _))
85    }
86
87    /// Whether the sync *operation* failed (publish, post-commit, or an
88    /// unrelated error). This is the single predicate that decides whether a
89    /// summary may sit in the `Err` position and which Display branch is used —
90    /// keep the two in lockstep.
91    ///
92    /// Note: per-message processing failures (`process.errored`) are
93    /// deliberately *not* counted here. A sync that decrypts some messages and
94    /// fails others is still a successful sync of the group as a whole; those
95    /// failures are reported inside the summary, not by flipping it to an error.
96    pub fn is_errored(&self) -> bool {
97        self.other.is_some()
98            || !self.publish_errors.is_empty()
99            || !self.post_commit_errors.is_empty()
100    }
101
102    pub fn add_publish_err(&mut self, e: GroupError) {
103        self.publish_errors.push(e);
104    }
105
106    pub fn add_post_commit_err(&mut self, e: GroupError) {
107        self.post_commit_errors.push(e);
108    }
109
110    pub fn add_process(&mut self, process: ProcessSummary) {
111        self.process = process;
112    }
113
114    pub fn extend(&mut self, other: SyncSummary) {
115        self.publish_errors.extend(other.publish_errors);
116        self.process.extend(other.process);
117        self.post_commit_errors.extend(other.post_commit_errors);
118        // Preserve the first non-None cause. `extend` is called once per retry
119        // round in `sync_until_intent_resolved_inner`; overwriting here would let
120        // a later clean round clobber an earlier round's `other` error, losing
121        // the cause on the timeout (SyncFailedToWait) path.
122        if self.other.is_none() {
123            self.other = other.other;
124        }
125    }
126
127    /// Construct a sync which failed with an unrelated error.
128    pub fn other(err: GroupError) -> Self {
129        Self {
130            other: Some(Box::new(err)),
131            ..Default::default()
132        }
133    }
134
135    pub fn add_other(&mut self, err: GroupError) {
136        self.other = Some(Box::new(err));
137    }
138}
139
140impl std::error::Error for SyncSummary {
141    /// Surface the first underlying cause so the error chain (e.g. the FFI
142    /// `Caused by:` walk) reaches the real failure instead of dead-ending at
143    /// the summary's flattened Display string. Prefer the sync-operation
144    /// errors, falling back to the first per-message processing error.
145    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
146        if let Some(other) = self.other.as_deref() {
147            return Some(other);
148        }
149        if let Some(err) = self.publish_errors.first() {
150            return Some(err);
151        }
152        if let Some(err) = self.post_commit_errors.first() {
153            return Some(err);
154        }
155        self.process
156            .errored
157            .first()
158            .map(|(_, e)| e as &dyn std::error::Error)
159    }
160}
161
162impl std::fmt::Debug for SyncSummary {
163    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
164        writeln!(f, "{}", self)
165    }
166}
167
168impl std::fmt::Display for SyncSummary {
169    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
170        if !self.is_errored() {
171            let first_new = self
172                .process
173                .new_messages
174                .iter()
175                .min_by_key(|k| k.cursor)
176                .map(|m| m.cursor);
177            write!(
178                f,
179                "synced {} messages, {} failed {} succeeded from cursor {:?}",
180                self.process.total_messages.len(),
181                self.process.errored.len(),
182                self.process.new_messages.len(),
183                first_new
184            )?;
185            if !self.process.errored.is_empty() {
186                write!(f, "{}", self.process.unique_errors())?;
187            }
188        } else {
189            writeln!(
190                f,
191                "================================= Errors Occurred During Sync ==========================="
192            )?;
193            if !self.publish_errors.is_empty() {
194                writeln!(f, "{} errors publishing intents", self.publish_errors.len())?;
195            }
196            if !self.post_commit_errors.is_empty() {
197                writeln!(f, "{} errors post commit", self.post_commit_errors.len())?;
198                // A call that ended a membership change's follow-up reports it
199                // only here, so each error names its outcome.
200                for error in &self.post_commit_errors {
201                    writeln!(f, "  {error}")?;
202                }
203            }
204            if let Some(e) = &self.other {
205                writeln!(f, "{}", e)?;
206            }
207            writeln!(f, "{}", self.process)?;
208            writeln!(
209                f,
210                "========================================================================================"
211            )?;
212        }
213        Ok(())
214    }
215}
216
217#[derive(Clone, PartialEq, Eq, Builder)]
218#[builder(setter(into), build_fn(error = "GroupMessageProcessingError"))]
219pub struct MessageIdentifier {
220    /// the cursor of the message as it exists on the network
221    pub cursor: Cursor,
222    pub group_id: GroupId,
223    pub created_ns: chrono::DateTime<Utc>,
224    /// true if the message has been processed previously
225    #[builder(default = false)]
226    pub previously_processed: bool,
227    /// the id of the message in the local database
228    #[builder(default = None)]
229    pub internal_id: Option<Vec<u8>>,
230    /// The context of the MLS Group from this message
231    /// Indicates that the message is a commit
232    #[builder(default = None)]
233    pub group_context: Option<GroupContext>,
234    /// The kind of intent processed, if this was our own message
235    #[builder(default = None)]
236    pub intent_kind: Option<IntentKind>,
237}
238
239impl MessageIdentifier {
240    pub fn builder() -> MessageIdentifierBuilder {
241        Default::default()
242    }
243}
244
245impl std::fmt::Debug for MessageIdentifier {
246    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
247        f.debug_struct("MessageIdentifier")
248            .field("cursor", &self.cursor)
249            .field("group_id", &xmtp_common::fmt::debug_hex(self.group_id))
250            .field("created_ns", &self.created_ns)
251            .field("internal_id", &self.internal_id)
252            .field("context", &self.group_context.as_ref().map(|g| g.epoch()))
253            .field("intent", &self.intent_kind)
254            .finish()
255    }
256}
257
258impl From<&xmtp_proto::types::GroupMessage> for MessageIdentifierBuilder {
259    fn from(value: &xmtp_proto::types::GroupMessage) -> Self {
260        MessageIdentifierBuilder {
261            cursor: Some(value.cursor),
262            group_id: Some(value.group_id),
263            created_ns: Some(value.created_ns),
264            internal_id: None,
265            group_context: None,
266            intent_kind: None,
267            previously_processed: Some(false),
268        }
269    }
270}
271
272impl From<&xmtp_proto::types::GroupMessage> for MessageIdentifier {
273    fn from(value: &xmtp_proto::types::GroupMessage) -> Self {
274        MessageIdentifier {
275            cursor: value.cursor,
276            group_id: value.group_id,
277            created_ns: value.created_ns,
278            internal_id: None,
279            group_context: None,
280            intent_kind: None,
281            previously_processed: false,
282        }
283    }
284}
285
286impl PartialOrd for MessageIdentifier {
287    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
288        self.cursor.partial_cmp(&other.cursor)
289    }
290}
291
292/// Information about which messages could be synced,
293/// And which messages could not be synced.
294#[derive(Default)]
295pub struct ProcessSummary {
296    pub total_messages: HashSet<Cursor>,
297    pub new_messages: Vec<MessageIdentifier>,
298    pub errored: Vec<(Cursor, GroupMessageProcessingError)>,
299    /// App-data changes seen while processing, in the order they were observed.
300    ///
301    /// Carried rather than dispatched because processing runs under the
302    /// per-group sync mutex; the caller fires them once it holds no group
303    /// locks. See `MlsGroup::dispatch_app_data_changes`.
304    pub app_data_changes: Vec<AppDataChange>,
305}
306
307impl std::fmt::Debug for ProcessSummary {
308    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
309        writeln!(f, "{}", self)
310    }
311}
312
313pub struct ErrorSet {
314    // Hashmap of message ids and the error they failed with
315    unique: HashMap<String, Vec<Cursor>>,
316    /// sorted vector of all failed ids
317    sorted_ids: Vec<(Cursor, String)>,
318}
319
320impl std::fmt::Display for ErrorSet {
321    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
322        for (error_msg, ids) in &self.unique {
323            let mut sorted_ids = ids.clone();
324            sorted_ids.sort();
325            write!(f, "\n\t┝━> {:?} {error_msg}", sorted_ids)?;
326        }
327        Ok(())
328    }
329}
330
331impl ErrorSet {
332    pub fn unique(&self) -> &HashMap<String, Vec<Cursor>> {
333        &self.unique
334    }
335
336    pub fn sorted(&self) -> &[(Cursor, String)] {
337        &self.sorted_ids
338    }
339}
340
341impl ProcessSummary {
342    pub fn add_id(&mut self, cursor: Cursor) {
343        self.total_messages.insert(cursor);
344    }
345
346    pub fn add(&mut self, message: MessageIdentifier) {
347        self.total_messages.insert(message.cursor);
348        self.new_messages.push(message);
349    }
350
351    /// the last message processed
352    pub fn last(&self) -> Option<Cursor> {
353        self.total_messages.iter().max().copied()
354    }
355
356    /// the first messages processed
357    pub fn first(&self) -> Option<Cursor> {
358        self.total_messages.iter().min().copied()
359    }
360
361    /// Total messages processed
362    pub fn total(&self) -> usize {
363        self.total_messages.len()
364    }
365
366    /// number of new messages processed
367    pub fn new_message(&self) -> usize {
368        self.new_messages.len()
369    }
370
371    /// the cursor of the first decryptable message
372    pub fn first_new(&self) -> Option<Cursor> {
373        self.new_messages.iter().map(|m| m.cursor).min()
374    }
375
376    /// the latest message that failed
377    pub fn last_errored(&self) -> Option<Cursor> {
378        self.errored.iter().map(|(i, _)| *i).max()
379    }
380
381    pub fn errored(&mut self, cursor: Cursor, error: GroupMessageProcessingError) {
382        self.errored.push((cursor, error));
383    }
384
385    pub fn unique_errors(&self) -> ErrorSet {
386        let mut sorted = self
387            .errored
388            .iter()
389            .map(|(m, e)| (*m, e.to_string()))
390            .collect::<Vec<(_, String)>>();
391        sorted.sort_by_key(|(m, _)| *m);
392        let mut error_set: HashMap<String, Vec<Cursor>> = HashMap::new();
393        for (id, err) in sorted.iter().cloned() {
394            error_set.entry(err).or_default().push(id);
395        }
396        ErrorSet {
397            unique: error_set,
398            sorted_ids: sorted,
399        }
400    }
401
402    pub fn extend(&mut self, other: ProcessSummary) {
403        self.total_messages.extend(other.total_messages);
404        self.new_messages.extend(other.new_messages);
405        self.errored.extend(other.errored)
406    }
407
408    pub fn is_errored(&self) -> bool {
409        !self.errored.is_empty()
410    }
411
412    /// detailed printout of the messages processed
413    fn detailed(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
414        let error_set = self.unique_errors();
415        writeln!(
416            f,
417            "\n=========================== Processed Messages Summary  ====================="
418        )?;
419        writeln!(
420            f,
421            "Processed {} total messages in cursor range [{:?} ... {:?}]",
422            self.total_messages.len(),
423            error_set.sorted_ids.first().map(|(m, _)| m),
424            error_set.sorted_ids.last().map(|(m, _)| m)
425        )?;
426        if !self.errored.is_empty() {
427            let error_ids = error_set.unique.values().flatten();
428            let min = error_ids.clone().min();
429            let max = error_ids.clone().max();
430
431            writeln!(
432                f,
433                "Failed to process [{}] messages in cursor range [{:?} ... {:?}]\n\
434                [{}] unique errors:",
435                self.errored.len(),
436                min,
437                max,
438                error_set.unique.len(),
439            )?;
440            for (err, ids) in error_set.unique.iter() {
441                writeln!(f, "{} ids errored with [{}]", ids.len(), err)?;
442            }
443        } else {
444            writeln!(f, "no errors encountered processing messages.")?;
445        }
446        let success_range = self.new_messages.iter().map(|m| m.cursor);
447        let min = success_range.clone().min();
448        let max = success_range.clone().max();
449        writeln!(
450            f,
451            "Successfully processed {} messages in range {:?} ... {:?}",
452            self.new_messages.len(),
453            min,
454            max
455        )?;
456        write!(
457            f,
458            "=============================================================================",
459        )?;
460
461        Ok(())
462    }
463}
464
465impl std::fmt::Display for ProcessSummary {
466    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
467        if self.total_messages.len() > 1 {
468            self.detailed(f)?;
469        } else {
470            write!(
471                f,
472                "processed {} total messages, ",
473                self.total_messages.len()
474            )?;
475            if !self.errored.is_empty() {
476                for (cursor, message) in self.errored.iter() {
477                    write!(f, "message with cursor {cursor} failed with {}", message)?;
478                }
479            }
480            if !self.new_messages.is_empty() {
481                write!(
482                    f,
483                    "{} new decryptable message(s) received",
484                    self.new_messages.len()
485                )?;
486            }
487        }
488        Ok(())
489    }
490}
491
492#[cfg(test)]
493mod extend_tests {
494    use super::*;
495
496    // `extend` is called once per retry round in
497    // `sync_until_intent_resolved_inner`. A later clean round must not clobber an
498    // earlier round's `other` cause — otherwise the timeout (SyncFailedToWait)
499    // path loses the real failure.
500    #[xmtp_common::test]
501    fn extend_preserves_first_other_cause() {
502        let mut acc = SyncSummary::default();
503        acc.extend(SyncSummary::other(GroupError::GroupInactive)); // round 1: cause
504        acc.extend(SyncSummary::default()); // round 2: clean — must not clobber
505
506        let other = acc.other.as_ref().expect("first cause must survive");
507        assert_eq!(other.to_string(), GroupError::GroupInactive.to_string());
508    }
509
510    #[xmtp_common::test]
511    fn extend_takes_other_when_none_yet() {
512        let mut acc = SyncSummary::default();
513        acc.extend(SyncSummary::default()); // round 1: clean
514        acc.extend(SyncSummary::other(GroupError::GroupInactive)); // round 2: cause appears
515
516        assert!(acc.other.is_some(), "a later cause is still captured");
517    }
518}
519
520#[cfg(test)]
521mod tests {
522    use super::*;
523    use crate::groups::mls_sync::GroupMessageProcessingError;
524    use std::error::Error;
525
526    // The Display "success" banner must fire iff is_errored() is false, and a
527    // summary may only be in the Err position when is_errored() is true. These
528    // two had drifted (Display used all-three-empty; is_errored() used && for
529    // publish/post_commit), so a publish-only failure printed the error banner
530    // while still reporting as not-errored. Lock the predicates together.
531    fn errored_banner(s: &SyncSummary) -> bool {
532        s.to_string().contains("Errors Occurred During Sync")
533    }
534
535    #[xmtp_common::test]
536    fn clean_summary_is_not_errored() {
537        let summary = SyncSummary::default();
538        assert!(!summary.is_errored());
539        assert!(!errored_banner(&summary));
540        assert!(summary.source().is_none());
541    }
542
543    #[xmtp_common::test]
544    fn publish_only_error_is_errored() {
545        let mut summary = SyncSummary::default();
546        summary.add_publish_err(GroupError::GroupInactive);
547        // Regression: with `&&` this returned false while Display showed the banner.
548        assert!(summary.is_errored());
549        assert!(errored_banner(&summary));
550    }
551
552    #[xmtp_common::test]
553    fn post_commit_only_error_is_errored() {
554        let mut summary = SyncSummary::default();
555        let error = GroupError::from(
556            crate::groups::mls_sync::publish::OutgoingPreparationError::WelcomesUnsendable,
557        );
558        let outcome = error.to_string();
559        summary.add_post_commit_err(error);
560        assert!(summary.is_errored());
561        assert!(errored_banner(&summary));
562        // A sync that ended a follow-up batch tells the app what happened.
563        assert!(GroupError::from(summary).to_string().contains(&outcome));
564    }
565
566    #[xmtp_common::test]
567    fn other_error_is_errored_and_is_source() {
568        let mut summary = SyncSummary::default();
569        summary.add_other(GroupError::GroupInactive);
570        assert!(summary.is_errored());
571        // `other` is the highest-priority cause surfaced through the chain.
572        let source = summary.source().expect("source should be present");
573        assert_eq!(source.to_string(), GroupError::GroupInactive.to_string());
574    }
575
576    #[xmtp_common::test]
577    fn per_message_failures_do_not_flip_errored() {
578        // A sync that fails to process some messages is still a successful sync
579        // of the group as a whole — it stays Ok and prints the success line.
580        let mut process = ProcessSummary::default();
581        process.errored(Cursor(7u64), GroupMessageProcessingError::InvalidPayload);
582        let mut summary = SyncSummary::default();
583        summary.add_process(process);
584
585        assert!(!summary.is_errored());
586        assert!(!errored_banner(&summary));
587        // ...but the failure is still reachable as the cause of last resort.
588        let source = summary
589            .source()
590            .expect("per-message error is the fallback source");
591        assert_eq!(
592            source.to_string(),
593            GroupMessageProcessingError::InvalidPayload.to_string()
594        );
595    }
596
597    #[xmtp_common::test]
598    fn source_prefers_other_over_per_message_error() {
599        let mut process = ProcessSummary::default();
600        process.errored(Cursor(7u64), GroupMessageProcessingError::InvalidPayload);
601        let mut summary = SyncSummary::default();
602        summary.add_process(process);
603        summary.add_publish_err(GroupError::GroupInactive);
604
605        let source = summary.source().expect("source should be present");
606        assert_eq!(source.to_string(), GroupError::GroupInactive.to_string());
607    }
608}
xmtp_mls/groups/summary.rs

xmtp_mls/groups/
summary.rs
