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
77impl crate::worker::NeedsDbReconnect for IncomingError {
78    fn needs_db_reconnect(&self) -> bool {
79        match self {
80            Self::Storage(error) => error.db_needs_connection(),
81            Self::Store(error) => error.needs_db_reconnect(),
82            Self::Group(error) => error.needs_db_reconnect(),
83            Self::Processing(error) => error.needs_db_reconnect(),
84            Self::Identity(error) => error.needs_db_reconnect(),
85            Self::Transport(_) | Self::UnsupportedTopic | Self::ReceiverEnded => false,
86        }
87    }
88}
89
90impl IncomingError {
91    pub fn code(&self) -> &'static str {
92        match self {
93            Self::Storage(_) => "incoming_storage",
94            Self::Store(_) => "incoming_receive",
95            Self::Transport(_) => "incoming_transport",
96            Self::Group(_) => "incoming_welcome",
97            Self::Processing(error) => error.processing_code(),
98            Self::Identity(_) => "incoming_identity",
99            Self::UnsupportedTopic => "unsupported_topic",
100            Self::ReceiverEnded => "incoming_receiver_ended",
101        }
102    }
103}
104
105/// Per-topic proof of durable receipt and processing through one fixed head.
106#[derive(Clone, Debug)]
107pub struct IncomingTopicStatus {
108    /// Topic whose progress is reported.
109    pub topic: Topic,
110    /// Scope revision that owns this target; old revisions cannot complete a new scope.
111    pub scope_generation: u64,
112    /// Registration is separate from both transport health and processing.
113    pub registration: IncomingRegistration,
114    /// Fixed head H; `None` means no target has been established yet.
115    pub target: Option<Cursor>,
116    /// Durable received prefix F, including retained pending work.
117    pub received: Cursor,
118    /// Durable processed prefix P, including terminal rejections.
119    pub processed: Cursor,
120    /// Unresolved Welcomes at or below H; later successes cannot hide them.
121    pub unresolved_welcomes: u64,
122    /// Completion predicate for this topic, not an application acknowledgement.
123    pub processing: IncomingProcessing,
124    /// Stable reason code when retained work cannot currently proceed.
125    pub blocked: Option<String>,
126    /// Latest error for this topic; sibling topics can still make progress.
127    pub error: Option<Arc<IncomingError>>,
128}
129
130/// Current scope status plus the immediately replaced scope's cancellation result.
131#[derive(Clone, Debug)]
132pub struct IncomingStatus {
133    /// Changes when the caller replaces the selected scope.
134    pub scope_generation: u64,
135    /// Changes when network receipt is opened again, without replacing fixed heads.
136    pub connection_generation: u64,
137    /// Health of the shared receiver, not proof of processing completion.
138    pub connection: IncomingConnection,
139    /// Independent fixed-target obligations for this scope.
140    pub topics: Vec<IncomingTopicStatus>,
141    /// The scope can still discover enrolled groups through its Welcome target.
142    pub discovery_pending: bool,
143    /// Aggregate processing result for the scope.
144    pub processing: IncomingProcessing,
145    /// Latest shared receiver or discovery error.
146    pub error: Option<Arc<IncomingError>>,
147    /// Only the immediately replaced generation is retained.
148    pub previous: Option<Box<IncomingStatus>>,
149}
150
151impl IncomingStatus {
152    pub(super) fn pending(generation: u64) -> Self {
153        Self {
154            scope_generation: generation,
155            connection_generation: 0,
156            connection: IncomingConnection::Connecting,
157            topics: Vec::new(),
158            discovery_pending: true,
159            processing: IncomingProcessing::Pending,
160            error: None,
161            previous: None,
162        }
163    }
164
165    pub(super) fn cancelled(generation: u64) -> Self {
166        let mut status = Self::pending(generation);
167        status.cancel();
168        status
169    }
170
171    pub(crate) fn cancel(&mut self) {
172        self.connection = IncomingConnection::Closed;
173        self.processing = IncomingProcessing::Cancelled;
174        self.discovery_pending = false;
175        for topic in &mut self.topics {
176            topic.registration = IncomingRegistration::Removed;
177            topic.processing = IncomingProcessing::Cancelled;
178        }
179    }
180
181    pub(super) fn without_previous(&self) -> Self {
182        let mut status = self.clone();
183        status.previous = None;
184        status
185    }
186}
xmtp_mls/subscriptions/incoming/status.rs

xmtp_mls/subscriptions/incoming/
status.rs
