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

xmtp_mls/groups/
summary.rs
