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

xmtp_mls/subscriptions/incoming/
status.rs
