xmtp_mls/subscriptions/incoming/
stream_api.rs

1//! The receiver operations that application streams use.
2//!
3//! A stream acquires its lease with [`IncomingCoordinator::acquire_stream`],
4//! which gives that lease its own recovery budget. While it waits, the stream
5//! rechecks the budget with [`IncomingLease::check_recovery`] on a timer as
6//! well as on lease changes, and ends with the [`RecoveryFailure`] it returns.
7//! A stream reports its last status with [`IncomingStatus::cancel`] applied
8//! when it closes, and a conversation stream skips rediscovery with a
9//! [`GroupSetGate`]. The order and timing of these checks belong to each stream.
10//!
11//! On native targets, the `shared_wires` child module owns the shared Bidi
12//! transports and their suspension, which the app's suspend and resume calls
13//! drive.
14
15use super::{
16    IncomingConnection, IncomingCoordinator, IncomingError, IncomingLease, IncomingProcessing,
17    IncomingRegistration, IncomingScope, IncomingStatus,
18};
19use std::sync::Arc;
20use xmtp_common::time::{Duration, Instant};
21
22#[cfg(not(target_arch = "wasm32"))]
23pub mod shared_wires;
24
25/// How often a waiting message reader rechecks its recovery budget without
26/// another wake, so a budget that runs out by time alone still ends the stream.
27pub const RECOVERY_POLL: Duration = Duration::from_secs(1);
28
29/// Why a stream's recovery budget ended the stream.
30#[derive(Clone, Debug)]
31pub enum RecoveryFailure {
32    /// The stream failed too often, or stayed disconnected too long, to retry.
33    Exhausted {
34        /// Failed attempts counted against this stream's budget.
35        attempts: u32,
36        /// The latest failure, or `None` when the deadline passed without one.
37        source: Option<Arc<IncomingError>>,
38    },
39    /// A failure that retrying cannot fix ended the stream at once.
40    Terminal(Arc<IncomingError>),
41}
42
43/// The group-set generation (`QueryGroup::group_set_generation`) of an
44/// operation's last successful discovery read, so a rediscovery is skipped
45/// while the stored group set is unchanged.
46///
47/// Read the generation before the discovery query and record it only after
48/// the query succeeds. A write that commits between the two reads then changes
49/// the generation the next pass sees, so the next pass reads again. Reading the
50/// generation after the query could cache an old set under a new generation.
51#[derive(Debug, Default)]
52pub struct GroupSetGate(Option<i64>);
53
54impl GroupSetGate {
55    /// Whether `current` differs from the recorded generation, or none is
56    /// recorded, so the operation must read again.
57    pub fn changed(&self, current: i64) -> bool {
58        self.0 != Some(current)
59    }
60
61    /// Record the generation read before a discovery read that succeeded.
62    pub fn record(&mut self, current: i64) {
63        self.0 = Some(current);
64    }
65}
66
67impl IncomingCoordinator {
68    /// Keep this scope active for an application stream until the returned
69    /// lease closes or drops.
70    ///
71    /// Application streams own a finite outage budget. Internal workers and
72    /// bounded sync operations keep their existing lifetime and deadline rules.
73    pub fn acquire_stream(self: &Arc<Self>, scope: IncomingScope) -> IncomingLease {
74        self.acquire_with_recovery(scope, true)
75    }
76}
77
78impl IncomingLease {
79    /// The failure that ended this lease's recovery budget, if one did.
80    ///
81    /// A lease from [`IncomingCoordinator::acquire`] has no budget, and a
82    /// closed lease no longer reports one, so both return `Ok`.
83    pub fn check_recovery(&self) -> Result<(), RecoveryFailure> {
84        let mut consumers = self.coordinator.state.consumer_recovery.lock();
85        match consumers.get_mut(&self.id) {
86            Some(state) => state.check(Instant::now()),
87            None => Ok(()),
88        }
89    }
90}
91
92impl IncomingStatus {
93    /// Report this status as ended: the connection closed, processing and
94    /// every topic cancelled, every topic registration removed, and no
95    /// discovery pending.
96    pub fn cancel(&mut self) {
97        self.connection = IncomingConnection::Closed;
98        self.processing = IncomingProcessing::Cancelled;
99        self.discovery_pending = false;
100        for topic in &mut self.topics {
101            topic.registration = IncomingRegistration::Removed;
102            topic.processing = IncomingProcessing::Cancelled;
103        }
104    }
105}
xmtp_mls/subscriptions/incoming/stream_api.rs

xmtp_mls/subscriptions/incoming/
stream_api.rs
