xmtp_mls/subscriptions/
catch_up.rs

1//! Bounded catch-up through the shared receiver and fixed processing barriers.
2
3use super::barrier::{BarrierError, receive_with_welcomes_until};
4use crate::{Client, context::XmtpSharedContext, groups::GroupError};
5use std::collections::{HashMap, HashSet};
6use xmtp_common::{
7    ErrorCode, RetryableError,
8    time::{Duration, Instant, now_ns},
9};
10use xmtp_db::{
11    delivery::{DeliveryScope, QueryDelivery},
12    group::{ConversationType, GroupQueryArgs},
13    incoming_envelope::{QueryIncomingEnvelope, StreamTopic},
14    prelude::*,
15};
16
17/// Committed catch-up observations; fixed processing predicates determine completion.
18#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
19pub struct CatchUpSummary {
20    /// Newly deliverable retained rows observed in enrolled groups during the run.
21    pub messages: u64,
22    /// Newly discovered non-virtual conversations within the enrolled scope.
23    pub conversations: u64,
24    /// Enrolled groups with a newly recorded terminal rejection.
25    pub failed: u64,
26    /// True only when every fixed processing obligation completed.
27    pub completed: bool,
28}
29
30#[derive(Debug, thiserror::Error, ErrorCode)]
31pub enum CatchUpError {
32    /// Target discovery or storage failed. May be retryable.
33    #[error(transparent)]
34    #[error_code(inherit)]
35    Group(#[from] GroupError),
36    /// Fixed targets remain unfinished. Partial committed progress is retained. May be retryable.
37    #[error("Catch-up did not complete: {summary:?}")]
38    Incomplete {
39        /// Partial progress remains committed even when the barrier fails.
40        summary: CatchUpSummary,
41        /// Every unfinished fixed-target obligation and its typed cause.
42        causes: Vec<BarrierError>,
43    },
44}
45
46impl RetryableError for CatchUpError {
47    fn is_retryable(&self) -> bool {
48        match self {
49            Self::Group(error) => error.is_retryable(),
50            Self::Incomplete { causes, .. } => causes.iter().any(RetryableError::is_retryable),
51        }
52    }
53}
54
55impl<C: XmtpSharedContext + 'static> Client<C> {
56    /// Process fixed starting targets and only the discoveries from the enrolled Welcomes.
57    pub async fn catch_up_to_live(
58        &self,
59        timeout: Option<Duration>,
60    ) -> Result<CatchUpSummary, CatchUpError> {
61        let timeout = timeout.unwrap_or(self.context.incoming_runtime().policy().barrier_timeout);
62        let started = Instant::now();
63        let query = || GroupQueryArgs {
64            include_sync_groups: true,
65            include_duplicate_dms: true,
66            ..Default::default()
67        };
68        let db = self.context.db();
69        let mut position = db.current_delivery_cursor().map_err(GroupError::from)?;
70        let before = db.find_groups(query()).map_err(GroupError::from)?;
71        let before_ids: HashSet<_> = before.iter().map(|group| group.id).collect();
72        let previous_rejections: HashMap<_, _> = before
73            .iter()
74            .map(|group| {
75                Ok((
76                    group.id,
77                    db.read_last_rejection(&StreamTopic::group(group.id))?
78                        .map(|entry| entry.sequence_id)
79                        .unwrap_or_default(),
80                ))
81            })
82            .collect::<Result<_, xmtp_db::StorageError>>()
83            .map_err(GroupError::from)?;
84        let run = receive_with_welcomes_until(
85            &self.context,
86            before_ids.iter().copied().collect(),
87            None,
88            started + timeout,
89        )
90        .await;
91        let causes: Vec<_> = run.result.err().into_iter().collect();
92        let enrolled: HashSet<_> = run.group_ids.iter().copied().collect();
93        let groups: Vec<_> = db
94            .find_groups(query())
95            .map_err(GroupError::from)?
96            .into_iter()
97            .filter(|group| enrolled.contains(&group.id))
98            .collect();
99        let upper = db.current_delivery_cursor().map_err(GroupError::from)?;
100        let settings = self.context.incoming_runtime().policy();
101        let mut summary = CatchUpSummary {
102            conversations: groups
103                .iter()
104                .filter(|group| {
105                    !before_ids.contains(&group.id)
106                        && !ConversationType::virtual_types().contains(&group.conversation_type)
107                })
108                .count() as u64,
109            completed: causes.is_empty(),
110            ..Default::default()
111        };
112        for group in &groups {
113            if let Some(rejection) = db
114                .read_last_rejection(&StreamTopic::group(group.id))
115                .map_err(GroupError::from)?
116                && rejection.sequence_id
117                    > previous_rejections
118                        .get(&group.id)
119                        .copied()
120                        .unwrap_or_default()
121            {
122                summary.failed += 1;
123            }
124        }
125        loop {
126            let rows = db
127                .replay_delivery_messages_bounded(
128                    position,
129                    &DeliveryScope::Groups(run.group_ids.clone()),
130                    now_ns(),
131                    settings.max_local_read_rows,
132                    settings.max_local_read_bytes,
133                )
134                .map_err(GroupError::from)?;
135            if rows.is_empty() {
136                break;
137            }
138            let mut reached_upper = false;
139            for row in rows {
140                if row.cursor.delivery_sequence > upper.delivery_sequence {
141                    reached_upper = true;
142                    break;
143                }
144                position = row.cursor;
145                summary.messages += 1;
146            }
147            if reached_upper || position.delivery_sequence >= upper.delivery_sequence {
148                break;
149            }
150        }
151        if !causes.is_empty() {
152            return Err(CatchUpError::Incomplete { summary, causes });
153        }
154        Ok(summary)
155    }
156}
157
158#[cfg(test)]
159mod tests {
160    use super::CatchUpSummary;
161    use crate::{tester, utils::MlsGroupExt};
162    use xmtp_db::group_message::MsgQueryArgs;
163    /// A client that has never streamed or synced catches up: the pending
164    /// welcome joins the group (discovery adds an update on the same
165    /// stream) and the group's history lands in the store.
166    #[xmtp_common::test(unwrap_try = true)]
167    async fn catch_up_joins_pending_groups_and_stores_history() {
168        tester!(alix);
169        tester!(bo);
170
171        let group = bo.create_group(None, None)?;
172        group.invite(&alix).await?;
173        group.send_msg(b"while you were out").await;
174        group.send_msg(b"still out").await;
175
176        let summary = alix.catch_up_to_live(None).await?;
177
178        let alix_group = alix.group(&group.group_id)?;
179        let bodies: Vec<Vec<u8>> = alix_group
180            .find_messages(&MsgQueryArgs::default())?
181            .into_iter()
182            .map(|m| m.decrypted_message_bytes)
183            .collect();
184        assert!(bodies.contains(&b"while you were out".to_vec()));
185        assert!(bodies.contains(&b"still out".to_vec()));
186        assert_eq!(summary.conversations, 1, "the joined group must be counted");
187        assert!(
188            summary.messages >= 2,
189            "the stored history must be counted (got {})",
190            summary.messages
191        );
192    }
193
194    /// A member with a stale durable cursor gets the missed tail replayed
195    /// and stored — and a second run finds nothing new to do.
196    #[xmtp_common::test(unwrap_try = true)]
197    async fn catch_up_replays_the_missed_tail_idempotently() {
198        tester!(alix);
199        tester!(bo);
200
201        let group = alix.create_group(None, None)?;
202        group.invite(&bo).await?;
203        bo.sync_welcomes().await?;
204        let bo_group = bo.group(&group.group_id)?;
205        bo_group.sync().await?; // durable cursor at "now"
206
207        group.send_msg(b"missed one").await;
208        group.send_msg(b"missed two").await;
209
210        let first = bo.catch_up_to_live(None).await?;
211        let count_after_first = bo_group.find_messages(&MsgQueryArgs::default())?.len();
212        let bodies: Vec<Vec<u8>> = bo_group
213            .find_messages(&MsgQueryArgs::default())?
214            .into_iter()
215            .map(|m| m.decrypted_message_bytes)
216            .collect();
217        assert!(bodies.contains(&b"missed one".to_vec()));
218        assert!(bodies.contains(&b"missed two".to_vec()));
219        assert!(
220            first.messages >= 2,
221            "the replayed tail must be counted (got {})",
222            first.messages
223        );
224        assert_eq!(first.conversations, 0, "no new group was joined");
225
226        // The honesty proof for the counter: the replay of already-stored
227        // history persists nothing, so the summary must say so.
228        let second = bo.catch_up_to_live(None).await?;
229        let count_after_second = bo_group.find_messages(&MsgQueryArgs::default())?.len();
230        assert_eq!(count_after_first, count_after_second);
231        assert_eq!(
232            second,
233            CatchUpSummary {
234                completed: true,
235                ..Default::default()
236            },
237            "a second run still reaches live, but persists nothing, so its counts are zero"
238        );
239    }
240
241    /// Nothing owed at all — a fresh client's run is just the welcome-topic
242    /// update acknowledgement and the empty target set.
243    #[xmtp_common::test(unwrap_try = true)]
244    async fn catch_up_with_nothing_owed_completes() {
245        tester!(alix);
246        let summary = alix.catch_up_to_live(None).await?;
247        assert_eq!(
248            summary,
249            CatchUpSummary {
250                completed: true,
251                ..Default::default()
252            },
253            "nothing owed still completes, with zero counts"
254        );
255    }
256}
xmtp_mls/subscriptions/catch_up.rs

xmtp_mls/subscriptions/
catch_up.rs
