xmtp_mls/subscriptions/
barrier.rs

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

xmtp_mls/subscriptions/
barrier.rs
