xmtp_mls/subscriptions/incoming/
status.rs

1use std::sync::Arc;
2use xmtp_common::RetryableError;
3use xmtp_proto::types::{Cursor, Topic};
4
5/// Transport health; a connected wire does not imply completed processing.
6#[derive(Clone, Copy, Debug, PartialEq, Eq)]
7pub enum IncomingConnection {
8    Connecting,
9    Connected,
10    Reconnecting,
11    Failed,
12    Closed,
13}
14
15/// Whether this topic still has network interest for the current scope.
16#[derive(Clone, Copy, Debug, PartialEq, Eq)]
17pub enum IncomingRegistration {
18    Pending,
19    Active,
20    Removed,
21}
22
23/// Processing state through the fixed targets, independent of application delivery.
24#[derive(Clone, Copy, Debug, PartialEq, Eq)]
25pub enum IncomingProcessing {
26    Pending,
27    Complete,
28    Blocked,
29    Cancelled,
30}
31
32#[derive(Debug, thiserror::Error)]
33pub enum IncomingError {
34    #[error("incoming receiver ended before recovery completed")]
35    ReceiverEnded,
36    #[error(transparent)]
37    Storage(#[from] xmtp_db::StorageError),
38    #[error(transparent)]
39    Store(#[from] crate::mls_store::MlsStoreError),
40    #[error(transparent)]
41    Transport(#[from] xmtp_proto::api::NetworkError),
42    #[error(transparent)]
43    Group(#[from] crate::groups::GroupError),
44    #[error(transparent)]
45    Processing(#[from] crate::groups::mls_sync::GroupMessageProcessingError),
46    #[error(transparent)]
47    Identity(#[from] crate::identity_updates::IdentityDependencyError),
48    #[error("unsupported incoming topic")]
49    UnsupportedTopic,
50}
51
52impl RetryableError for IncomingError {
53    fn is_retryable(&self) -> bool {
54        match self {
55            // The controller repairs this refusal with an ordered read from F.
56            // Raw storage callers must still treat the omitted prefix as invalid.
57            Self::Storage(xmtp_db::StorageError::Stream(
58                xmtp_db::stream_storage::StreamStorageError::MissingPrefix { .. },
59            ))
60            | Self::Store(crate::mls_store::MlsStoreError::Storage(
61                xmtp_db::StorageError::Stream(
62                    xmtp_db::stream_storage::StreamStorageError::MissingPrefix { .. },
63                ),
64            )) => true,
65            Self::Storage(error) => error.is_retryable(),
66            Self::Store(error) => error.is_retryable(),
67            Self::Transport(error) => error.is_retryable(),
68            Self::Group(error) => error.is_retryable(),
69            Self::Processing(error) => error.is_retryable(),
70            Self::Identity(error) => error.is_retryable(),
71            Self::UnsupportedTopic => false,
72            Self::ReceiverEnded => true,
73        }
74    }
75
76    fn rate_limit_delay(&self) -> Option<xmtp_common::time::Duration> {
77        match self {
78            Self::Store(error) => error.rate_limit_delay(),
79            Self::Transport(error) => error.rate_limit_delay(),
80            Self::Group(error) => error.rate_limit_delay(),
81            Self::Processing(error) => error.rate_limit_delay(),
82            Self::Identity(error) => error.rate_limit_delay(),
83            _ => None,
84        }
85    }
86}
87
88impl crate::worker::NeedsDbReconnect for IncomingError {
89    #[deny(clippy::wildcard_enum_match_arm)]
90    fn needs_db_reconnect(&self) -> bool {
91        match self {
92            Self::Storage(error) => error.db_needs_connection(),
93            Self::Store(error) => error.needs_db_reconnect(),
94            Self::Group(error) => error.needs_db_reconnect(),
95            Self::Processing(error) => error.needs_db_reconnect(),
96            Self::Identity(error) => error.needs_db_reconnect(),
97            Self::Transport(_) | Self::UnsupportedTopic | Self::ReceiverEnded => false,
98        }
99    }
100}
101
102impl IncomingError {
103    pub fn code(&self) -> &'static str {
104        match self {
105            Self::Storage(_) => "incoming_storage",
106            Self::Store(_) => "incoming_receive",
107            Self::Transport(_) => "incoming_transport",
108            Self::Group(_) => "incoming_welcome",
109            Self::Processing(error) => error.processing_code(),
110            Self::Identity(_) => "incoming_identity",
111            Self::UnsupportedTopic => "unsupported_topic",
112            Self::ReceiverEnded => "incoming_receiver_ended",
113        }
114    }
115}
116
117/// Per-topic proof of durable receipt and processing through one fixed head.
118#[derive(Clone, Debug)]
119pub struct IncomingTopicStatus {
120    /// Topic whose progress is reported.
121    pub topic: Topic,
122    /// Scope revision that owns this target; old revisions cannot complete a new scope.
123    pub scope_generation: u64,
124    /// Registration is separate from both transport health and processing.
125    pub registration: IncomingRegistration,
126    /// Fixed head H; `None` means no target has been established yet.
127    pub target: Option<Cursor>,
128    /// Durable received prefix F, including retained pending work.
129    pub received: Cursor,
130    /// Durable processed prefix P, including terminal rejections.
131    pub processed: Cursor,
132    /// Unresolved Welcomes at or below H; later successes cannot hide them.
133    pub unresolved_welcomes: u64,
134    /// Completion predicate for this topic, not an application acknowledgement.
135    pub processing: IncomingProcessing,
136    /// Stable reason code when retained work cannot currently proceed.
137    pub blocked: Option<String>,
138    /// Latest error for this topic; sibling topics can still make progress.
139    pub error: Option<Arc<IncomingError>>,
140}
141
142/// Current scope status plus the immediately replaced scope's cancellation result.
143#[derive(Clone, Debug)]
144pub struct IncomingStatus {
145    /// Changes when the caller replaces the selected scope.
146    pub scope_generation: u64,
147    /// Changes when network receipt is opened again, without replacing fixed heads.
148    pub connection_generation: u64,
149    /// Health of the shared receiver, not proof of processing completion.
150    pub connection: IncomingConnection,
151    /// Independent fixed-target obligations for this scope.
152    pub topics: Vec<IncomingTopicStatus>,
153    /// The scope can still discover enrolled groups through its Welcome target.
154    pub discovery_pending: bool,
155    /// Aggregate processing result for the scope.
156    pub processing: IncomingProcessing,
157    /// Latest shared receiver or discovery error.
158    pub error: Option<Arc<IncomingError>>,
159    /// Only the immediately replaced generation is retained.
160    pub previous: Option<Box<IncomingStatus>>,
161}
162
163impl IncomingStatus {
164    pub(super) fn pending(generation: u64) -> Self {
165        Self {
166            scope_generation: generation,
167            connection_generation: 0,
168            connection: IncomingConnection::Connecting,
169            topics: Vec::new(),
170            discovery_pending: true,
171            processing: IncomingProcessing::Pending,
172            error: None,
173            previous: None,
174        }
175    }
176
177    pub(super) fn cancelled(generation: u64) -> Self {
178        let mut status = Self::pending(generation);
179        status.cancel();
180        status
181    }
182
183    pub(super) fn without_previous(&self) -> Self {
184        let mut status = self.clone();
185        status.previous = None;
186        status
187    }
188}
xmtp_mls/subscriptions/incoming/status.rs

xmtp_mls/subscriptions/incoming/
status.rs
