xmtp_mls/subscriptions/
barrier.rs

1//! Fixed processing targets over the shared durable receiver.
2
3#[cfg(test)]
4mod tests;
5
6use futures::{StreamExt, stream};
7use std::{collections::HashSet, sync::Arc};
8use xmtp_common::{
9    ErrorCode, RetryableError,
10    time::{Duration, Instant, sleep},
11};
12use xmtp_db::{
13    StorageError,
14    consent_record::ConsentState,
15    group::GroupQueryArgs,
16    incoming_envelope::{NetworkEntityKind, QueryIncomingEnvelope, StreamTopic, TopicProgress},
17    prelude::*,
18};
19use xmtp_proto::types::{Cursor, GroupId, Topic, TopicCursor, TopicKind};
20
21use super::incoming::{
22    IncomingCoordinator, IncomingError, IncomingProcessing, IncomingReceivePolicy, IncomingScope,
23};
24use crate::{
25    context::XmtpSharedContext,
26    groups::{GroupError, MlsGroup},
27};
28
29/// Why one fixed processing obligation is unfinished.
30#[derive(Debug, Clone)]
31pub enum BarrierCause {
32    /// The backend head has not been captured; zero must not be assumed.
33    TargetPending,
34    /// The durable received prefix has not reached the target.
35    ReceiptPending,
36    /// Receipt is complete, but ordered processing is still pending.
37    ProcessingPending,
38    Blocked(String),
39    Storage(Arc<StorageError>),
40    Receiver(Arc<IncomingError>),
41    InvalidTopic,
42}
43
44/// One fixed target and its latest durable progress (STR-042 and STR-043).
45#[derive(Debug, Clone)]
46pub struct BarrierTopic {
47    /// The group, Welcome, or identity topic covered by this obligation.
48    pub topic: Topic,
49    /// Fixed backend head H. `None` means target capture did not succeed.
50    pub target: Option<Cursor>,
51    /// Durable received prefix F, including retained pending envelopes.
52    pub received: Cursor,
53    /// Durable processed prefix P; independent of application delivery.
54    pub processed: Cursor,
55    /// Unresolved Welcome sequence IDs at or below H, even after later successes.
56    pub unresolved_welcomes: Vec<Cursor>,
57    /// Removal completed this obligation without advancing P further.
58    pub inactive: bool,
59    /// `None` only when this obligation is complete.
60    pub cause: Option<BarrierCause>,
61}
62
63impl BarrierTopic {
64    /// Receipt alone does not complete a processing obligation.
65    pub fn complete(&self) -> bool {
66        self.cause.is_none()
67    }
68
69    fn blocked(&self) -> bool {
70        matches!(
71            self.cause,
72            Some(BarrierCause::Blocked(_) | BarrierCause::InvalidTopic)
73        ) || matches!(&self.cause, Some(BarrierCause::Storage(error)) if !error.is_retryable())
74            || matches!(&self.cause, Some(BarrierCause::Receiver(error)) if !error.is_retryable())
75    }
76}
77
78/// Settled obligations from one bounded sync run.
79#[derive(Debug, Clone)]
80pub struct BarrierSnapshot {
81    /// Each topic keeps its own fixed target and progress.
82    pub topics: Vec<BarrierTopic>,
83}
84
85/// Why a bounded sync stopped before every obligation completed.
86#[derive(Debug, Clone, Copy, PartialEq, Eq)]
87pub enum BarrierFailure {
88    Blocked,
89    Deadline,
90    Cancelled,
91}
92
93#[derive(Debug, thiserror::Error, ErrorCode)]
94pub enum BarrierError {
95    /// Processing did not meet the fixed targets. Pending work remains durable. May be retryable.
96    #[error("Processing barrier did not complete: {reason:?}; {} unfinished topics", unfinished.len())]
97    Incomplete {
98        /// The run-level stop reason; individual causes remain below.
99        reason: BarrierFailure,
100        /// Every unfinished obligation, not only the first failure.
101        unfinished: Vec<BarrierTopic>,
102    },
103}
104
105impl RetryableError for BarrierError {
106    fn is_retryable(&self) -> bool {
107        matches!(
108            self,
109            Self::Incomplete {
110                reason: BarrierFailure::Deadline,
111                ..
112            }
113        )
114    }
115}
116
117impl crate::worker::NeedsDbReconnect for BarrierError {
118    fn needs_db_reconnect(&self) -> bool {
119        let Self::Incomplete { unfinished, .. } = self;
120        unfinished.iter().any(|topic| match &topic.cause {
121            Some(BarrierCause::Storage(error)) => error.db_needs_connection(),
122            Some(BarrierCause::Receiver(error)) => error.needs_db_reconnect(),
123            _ => false,
124        })
125    }
126}
127
128/// Test durable completion without assuming transport coverage or an active group.
129/// Welcome progress requires every pending parent through the fixed target to finish.
130pub(crate) fn durable_complete(
131    kind: NetworkEntityKind,
132    target: Option<Cursor>,
133    progress: TopicProgress,
134    pending_count: usize,
135) -> bool {
136    target.is_some_and(|target| {
137        progress.received >= target
138            && match kind {
139                NetworkEntityKind::Welcome => pending_count == 0,
140                NetworkEntityKind::Group | NetworkEntityKind::Identity => {
141                    progress.processed >= target
142                }
143            }
144    })
145}
146
147fn read_topic<C: XmtpSharedContext>(context: &C, topic: &Topic, target: Cursor) -> BarrierTopic {
148    let mut status = BarrierTopic {
149        topic: topic.clone(),
150        target: Some(target),
151        received: Cursor::default(),
152        processed: Cursor::default(),
153        unresolved_welcomes: Vec::new(),
154        inactive: false,
155        cause: Some(BarrierCause::ReceiptPending),
156    };
157    let kind = match topic.kind() {
158        TopicKind::GroupMessagesV1 => NetworkEntityKind::Group,
159        TopicKind::WelcomeMessagesV1 => NetworkEntityKind::Welcome,
160        TopicKind::IdentityUpdatesV1 => NetworkEntityKind::Identity,
161        _ => {
162            status.cause = Some(BarrierCause::InvalidTopic);
163            return status;
164        }
165    };
166    let db_topic = StreamTopic {
167        entity_id: topic.identifier().to_vec(),
168        kind,
169    };
170    let db = context.db();
171    let progress = match db.topic_progress(&db_topic) {
172        Ok(progress) => progress,
173        Err(error) => {
174            status.cause = Some(BarrierCause::Storage(Arc::new(error)));
175            return status;
176        }
177    };
178    status.received = progress.received;
179    status.processed = progress.processed;
180    if kind == NetworkEntityKind::Group {
181        match GroupId::try_from(topic.identifier()) {
182            Ok(id) => match MlsGroup::new_cached(context.context_ref().clone(), &id)
183                .map_err(GroupError::from)
184                .and_then(|(group, _)| group.is_active())
185            {
186                Ok(false) => {
187                    status.inactive = true;
188                    status.cause = None;
189                    return status;
190                }
191                Ok(true) => {}
192                Err(error) => {
193                    status.cause = Some(BarrierCause::Receiver(Arc::new(IncomingError::Group(
194                        error,
195                    ))));
196                    return status;
197                }
198            },
199            Err(_) => {
200                status.cause = Some(BarrierCause::InvalidTopic);
201                return status;
202            }
203        }
204    }
205    let pending = match db.pending_states_through(&db_topic, target) {
206        Ok(pending) => pending,
207        Err(error) => {
208            status.cause = Some(BarrierCause::Storage(Arc::new(error)));
209            return status;
210        }
211    };
212    if kind == NetworkEntityKind::Welcome {
213        status.unresolved_welcomes = pending.iter().map(|row| row.sequence_id).collect();
214    }
215    status.cause = if progress.received < target {
216        Some(BarrierCause::ReceiptPending)
217    } else if durable_complete(kind, Some(target), progress, pending.len()) {
218        None
219    } else if !pending.is_empty()
220        && (kind != NetworkEntityKind::Welcome && pending[0].blocked
221            || kind == NetworkEntityKind::Welcome && pending.iter().all(|row| row.blocked))
222    {
223        Some(BarrierCause::Blocked(
224            pending[0]
225                .error_code
226                .clone()
227                .unwrap_or_else(|| "processing_blocked".into()),
228        ))
229    } else {
230        Some(BarrierCause::ProcessingPending)
231    };
232    status
233}
234
235/// Sample targets once and query missing receipt without first waiting for a live stream.
236/// Traffic after these targets cannot extend the call.
237pub async fn receive_through_current<C: XmtpSharedContext>(
238    context: &C,
239    topics: Vec<Topic>,
240) -> Result<BarrierSnapshot, GroupError> {
241    let deadline = Instant::now() + context.incoming_runtime().policy().barrier_timeout;
242    Ok(receive_through_current_until(context, topics, deadline).await?)
243}
244
245/// Target sampling and processing use one deadline.
246pub async fn receive_through_current_until<C: XmtpSharedContext>(
247    context: &C,
248    topics: Vec<Topic>,
249    deadline: Instant,
250) -> Result<BarrierSnapshot, BarrierError> {
251    let coordinator = IncomingCoordinator::for_context(context);
252    let _receipt = coordinator.acquire(IncomingScope::Topics(topics.clone()));
253    let (targets, unavailable) = capture_targets(context, topics, deadline).await;
254    wait_for_targets(
255        context,
256        targets,
257        unavailable,
258        None,
259        IncomingReceivePolicy::ImmediateQuery,
260        deadline,
261        &mut HashSet::new(),
262    )
263    .await
264}
265
266/// Fixed starting groups plus discoveries attributed to Welcomes at or below the fixed H.
267pub struct GroupBarrierRun {
268    /// Exact enrolled scope; excludes unrelated groups created while the call runs.
269    pub group_ids: Vec<GroupId>,
270    /// Successful targets or every unfinished obligation on failure.
271    pub result: Result<BarrierSnapshot, BarrierError>,
272}
273
274/// Freezes which Welcome discoveries may expand an all-groups sync run.
275struct WelcomeDiscovery {
276    topic: Topic,
277    target: Cursor,
278    consent_states: Option<Vec<ConsentState>>,
279}
280
281/// Process starting groups and enrolled Welcome discoveries under one deadline.
282/// Later local groups and Welcomes above the sampled head cannot extend the run.
283pub async fn receive_with_welcomes_until<C: XmtpSharedContext>(
284    context: &C,
285    starting_groups: Vec<GroupId>,
286    consent_states: Option<Vec<ConsentState>>,
287    deadline: Instant,
288) -> GroupBarrierRun {
289    let mut groups: HashSet<_> = starting_groups.into_iter().collect();
290    let welcome_topic = Topic::new_welcome_message(context.installation_id());
291    let mut topics: Vec<_> = groups.iter().map(Topic::new_group_message).collect();
292    topics.push(welcome_topic.clone());
293    let coordinator = IncomingCoordinator::for_context(context);
294    let _receipt = coordinator.acquire(IncomingScope::Topics(topics.clone()));
295    let (targets, unavailable) = capture_targets(context, topics, deadline).await;
296    let discovery = targets
297        .get(&welcome_topic)
298        .copied()
299        .map(|target| WelcomeDiscovery {
300            topic: welcome_topic,
301            target,
302            consent_states,
303        });
304    let result = wait_for_targets(
305        context,
306        targets,
307        unavailable,
308        discovery,
309        IncomingReceivePolicy::ImmediateQuery,
310        deadline,
311        &mut groups,
312    )
313    .await;
314    let mut group_ids: Vec<_> = groups.into_iter().collect();
315    group_ids.sort();
316    GroupBarrierRun { group_ids, result }
317}
318
319/// Settle all bounded head queries; keep successful targets when another query fails.
320async fn capture_targets<C: XmtpSharedContext>(
321    context: &C,
322    topics: Vec<Topic>,
323    deadline: Instant,
324) -> (TopicCursor, Vec<BarrierTopic>) {
325    let mut topics = topics;
326    topics.sort_by_key(Topic::cloned_vec);
327    topics.dedup();
328    let chunks: Vec<_> = topics
329        .chunks(context.api().limits().max_query_topics)
330        .map(<[Topic]>::to_vec)
331        .collect();
332    let results = stream::iter(chunks)
333        .map(|topics| async move {
334            let remaining = deadline.saturating_duration_since(Instant::now());
335            let result = xmtp_common::time::timeout(
336                remaining,
337                context.api().newest_topic_cursors(topics.clone()),
338            )
339            .await;
340            (topics, result)
341        })
342        .buffer_unordered(context.incoming_runtime().policy().max_dependency_requests)
343        .collect::<Vec<_>>()
344        .await;
345    let mut targets = TopicCursor::new();
346    let mut unavailable = Vec::new();
347    for (topics, result) in results {
348        let cause = match result {
349            Ok(Ok(captured)) => {
350                targets.extend(captured);
351                continue;
352            }
353            Ok(Err(error)) => BarrierCause::Receiver(Arc::new(IncomingError::Transport(
354                xmtp_proto::api::NetworkError::new(error),
355            ))),
356            Err(_) => BarrierCause::TargetPending,
357        };
358        for topic in topics {
359            let mut status = read_topic(context, &topic, Cursor::default());
360            status.target = None;
361            status.cause = Some(cause.clone());
362            unavailable.push(status);
363        }
364    }
365    (targets, unavailable)
366}
367
368/// Wait for all independent obligations before reporting blocked work.
369/// Prefer a healthy receiver for caller-supplied targets, with bounded Query fallback.
370pub async fn wait_through<C: XmtpSharedContext>(
371    context: &C,
372    targets: TopicCursor,
373    timeout: Option<Duration>,
374) -> Result<BarrierSnapshot, BarrierError> {
375    let deadline =
376        Instant::now() + timeout.unwrap_or(context.incoming_runtime().policy().barrier_timeout);
377    wait_for_targets(
378        context,
379        targets,
380        Vec::new(),
381        None,
382        IncomingReceivePolicy::StreamFirst,
383        deadline,
384        &mut HashSet::new(),
385    )
386    .await
387}
388
389async fn wait_for_targets<C: XmtpSharedContext>(
390    context: &C,
391    mut targets: TopicCursor,
392    mut unavailable: Vec<BarrierTopic>,
393    discovery: Option<WelcomeDiscovery>,
394    receive_policy: IncomingReceivePolicy,
395    deadline: Instant,
396    groups: &mut HashSet<GroupId>,
397) -> Result<BarrierSnapshot, BarrierError> {
398    let coordinator = IncomingCoordinator::for_context(context);
399    let lease = coordinator.acquire(IncomingScope::Barrier {
400        targets: targets.clone(),
401        deadline,
402        receive_policy,
403    });
404    loop {
405        #[cfg(test)]
406        tests::before_progress_read();
407        // Read completion before discovery. A completed Welcome transaction also
408        // records its groups, so the later discovery scan must include them.
409        let mut topics: Vec<_> = targets
410            .iter()
411            .map(|(topic, target)| read_topic(context, topic, *target))
412            .collect();
413        let mut discovery_failure = None;
414        if let Some(discovery) = &discovery {
415            let discovered = (|| -> Result<Vec<GroupId>, StorageError> {
416                let db = context.db();
417                let eligible = db.group_ids_discovered_through(discovery.target)?;
418                let selected: HashSet<_> = db
419                    .find_groups(GroupQueryArgs {
420                        consent_states: discovery.consent_states.clone(),
421                        include_sync_groups: true,
422                        include_duplicate_dms: true,
423                        ..Default::default()
424                    })?
425                    .into_iter()
426                    .map(|group| group.id)
427                    .collect();
428                Ok(eligible
429                    .into_iter()
430                    .filter(|id| selected.contains(id))
431                    .collect())
432            })();
433            match discovered {
434                Ok(discovered) => {
435                    let new_topics: Vec<_> = discovered
436                        .into_iter()
437                        .filter(|id| groups.insert(*id))
438                        .map(Topic::new_group_message)
439                        .collect();
440                    if !new_topics.is_empty() {
441                        let (captured, failures) =
442                            capture_targets(context, new_topics, deadline).await;
443                        targets.extend(captured);
444                        unavailable.extend(failures);
445                        lease.replace_scope(IncomingScope::Barrier {
446                            targets: targets.clone(),
447                            deadline,
448                            receive_policy,
449                        });
450                        // The prior snapshot did not include these obligations.
451                        continue;
452                    }
453                }
454                Err(error) => {
455                    let mut status = read_topic(context, &discovery.topic, discovery.target);
456                    status.cause = Some(BarrierCause::Storage(Arc::new(error)));
457                    discovery_failure = Some(status);
458                }
459            }
460        }
461        let receiver = lease.snapshot();
462        for topic in &mut topics {
463            if topic.complete() {
464                continue;
465            }
466            if let Some(incoming) = receiver
467                .topics
468                .iter()
469                .find(|entry| entry.topic == topic.topic)
470                && incoming.processing == IncomingProcessing::Blocked
471                && let Some(error) = incoming.error.as_ref()
472            {
473                topic.cause = Some(BarrierCause::Receiver(error.clone()));
474            }
475        }
476        for status in &unavailable {
477            let mut fresh = read_topic(context, &status.topic, Cursor::default());
478            fresh.target = None;
479            fresh.cause = status.cause.clone();
480            topics.push(fresh);
481        }
482        if let Some(failure) = discovery_failure {
483            if let Some(status) = topics
484                .iter_mut()
485                .find(|status| status.topic == failure.topic)
486            {
487                *status = failure;
488            } else {
489                topics.push(failure);
490            }
491        }
492        if topics.iter().all(BarrierTopic::complete) {
493            return Ok(BarrierSnapshot { topics });
494        }
495        let reason = if context.is_closed() {
496            Some(BarrierFailure::Cancelled)
497        } else if topics.iter().all(|topic| {
498            topic.complete()
499                || topic.blocked()
500                    && (topic.target.is_none()
501                        || receiver
502                            .topics
503                            .iter()
504                            .any(|entry| entry.topic == topic.topic))
505        }) {
506            Some(BarrierFailure::Blocked)
507        } else if Instant::now() >= deadline {
508            Some(BarrierFailure::Deadline)
509        } else {
510            None
511        };
512        if let Some(reason) = reason {
513            return Err(BarrierError::Incomplete {
514                reason,
515                unfinished: topics
516                    .into_iter()
517                    .filter(|topic| !topic.complete())
518                    .collect(),
519            });
520        }
521        tokio::select! {
522            _ = context.cancellation_token().cancelled() => {},
523            _ = lease.changed() => {},
524            _ = sleep(context.incoming_runtime().policy().active_database_poll_interval.min(deadline.saturating_duration_since(Instant::now()))) => {},
525        }
526    }
527}
xmtp_mls/subscriptions/barrier.rs

xmtp_mls/subscriptions/
barrier.rs
