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

xmtp_mls/subscriptions/
barrier.rs
