xmtp_mls/subscriptions/
stream_failure.rs

1//! Structured processing failures shared by string-based binding errors.
2
3use std::error::Error;
4
5use serde::{Deserialize, Serialize};
6use xmtp_common::{ErrorCode, RetryableError};
7
8#[cfg(not(target_arch = "wasm32"))]
9use super::catch_up::CatchUpError;
10use super::{
11    SubscribeError,
12    barrier::{BarrierCause, BarrierError, BarrierFailure, BarrierTopic},
13    incoming::IncomingError,
14};
15use crate::{
16    client::ClientError,
17    groups::{GroupError, mls_sync::GroupMessageProcessingError, summary::SyncSummary},
18};
19
20/// The suffix version is independent of the normal error message and code.
21pub const STREAM_FAILURE_MARKER: &str = "\n[XMTP_STREAM_FAILURE_V1]";
22
23/// The operation whose fixed processing obligations remain unfinished.
24#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
25#[serde(rename_all = "snake_case")]
26pub enum StreamFailureKind {
27    Barrier,
28    PublishedButUnconfirmed,
29    CatchUp,
30}
31
32/// Why the barrier stopped waiting. Durable pending work is retained.
33#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
34#[serde(rename_all = "snake_case")]
35pub enum StreamBarrierReason {
36    Blocked,
37    Deadline,
38    Cancelled,
39}
40
41/// Stable categories for one unfinished topic's cause.
42#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
43#[serde(rename_all = "snake_case")]
44pub enum StreamBarrierCauseKind {
45    TargetPending,
46    ReceiptPending,
47    ProcessingPending,
48    Blocked,
49    Storage,
50    Receiver,
51    InvalidTopic,
52}
53
54/// Safe cause metadata. Messages do not contain raw nested error data.
55#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
56#[serde(rename_all = "camelCase")]
57pub struct StreamBarrierCause {
58    pub kind: StreamBarrierCauseKind,
59    /// Stable source code when the cause has one.
60    pub code: Option<String>,
61    /// Fixed category text for display, not for control flow.
62    pub message: String,
63    pub retryable: bool,
64}
65
66/// All 64-bit values are decimal strings. Topic bytes use hexadecimal.
67#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
68#[serde(rename_all = "camelCase")]
69pub struct StreamBarrierTopic {
70    /// Complete topic bytes, encoded as hexadecimal.
71    pub topic: String,
72    /// Captured owning scope, encoded as an exact decimal u64 string.
73    #[serde(default)]
74    pub scope_generation: Option<String>,
75    /// Fixed target H. None means capture failed; "0" is a captured empty target.
76    pub target: Option<String>,
77    /// Durable receipt cursor F. This is not application delivery progress.
78    pub received: String,
79    /// Resolved processing cursor P, independent of local delivery acknowledgement D.
80    pub processed: String,
81    /// Exact unresolved Welcome sequence IDs at or below H.
82    pub unresolved_welcomes: Vec<String>,
83    pub inactive: bool,
84    pub cause: Option<StreamBarrierCause>,
85}
86
87/// One failed barrier and every topic obligation that it did not complete.
88#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
89#[serde(rename_all = "camelCase")]
90pub struct StreamBarrierFailure {
91    pub reason: StreamBarrierReason,
92    /// Keep all entries, including sibling failures from the same sync summary.
93    pub unfinished: Vec<StreamBarrierTopic>,
94}
95
96/// Partial committed catch-up counts, encoded as exact decimal u64 strings.
97#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
98#[serde(rename_all = "camelCase")]
99pub struct StreamFailureSummary {
100    pub messages: String,
101    pub conversations: String,
102    pub failed: String,
103    pub completed: bool,
104}
105
106/// Versioned error details shared by all string-based binding error carriers.
107#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
108#[serde(rename_all = "camelCase")]
109pub struct StreamFailureDetails {
110    pub kind: StreamFailureKind,
111    pub code: String,
112    pub message: String,
113    pub retryable: bool,
114    /// The sole published intent, if there is exactly one.
115    pub intent_id: Option<i32>,
116    /// All published intent IDs from nested errors, without duplicates.
117    pub published_intent_ids: Vec<i32>,
118    /// Committed catch-up progress when the error came from a catch-up run.
119    pub summary: Option<StreamFailureSummary>,
120    /// All failed barriers, not only the first source in an error chain.
121    pub barriers: Vec<StreamBarrierFailure>,
122}
123
124impl From<&BarrierCause> for StreamBarrierCause {
125    fn from(cause: &BarrierCause) -> Self {
126        use StreamBarrierCauseKind as Kind;
127        let (kind, code, message, retryable) = match cause {
128            BarrierCause::TargetPending => {
129                (Kind::TargetPending, None, "Target capture is pending", true)
130            }
131            BarrierCause::ReceiptPending => {
132                (Kind::ReceiptPending, None, "Receipt is pending", true)
133            }
134            BarrierCause::ProcessingPending => {
135                (Kind::ProcessingPending, None, "Processing is pending", true)
136            }
137            BarrierCause::Blocked(code) => (
138                Kind::Blocked,
139                Some(code.clone()),
140                "Processing is blocked",
141                false,
142            ),
143            BarrierCause::Storage(error) => (
144                Kind::Storage,
145                Some(error.error_code().to_string()),
146                "Storage failed",
147                error.is_retryable(),
148            ),
149            BarrierCause::Receiver(error) => (
150                Kind::Receiver,
151                Some(error.code().to_owned()),
152                "The receiver failed",
153                error.is_retryable(),
154            ),
155            BarrierCause::InvalidTopic => (Kind::InvalidTopic, None, "The topic is invalid", false),
156        };
157        Self {
158            kind,
159            code,
160            message: message.to_owned(),
161            retryable,
162        }
163    }
164}
165
166impl From<&BarrierTopic> for StreamBarrierTopic {
167    fn from(topic: &BarrierTopic) -> Self {
168        Self {
169            topic: hex::encode(topic.topic.cloned_vec()),
170            scope_generation: topic
171                .scope_generation
172                .map(|generation| generation.to_string()),
173            target: topic.target.map(|cursor| cursor.0.to_string()),
174            received: topic.received.0.to_string(),
175            processed: topic.processed.0.to_string(),
176            unresolved_welcomes: topic
177                .unresolved_welcomes
178                .iter()
179                .map(|cursor| cursor.0.to_string())
180                .collect(),
181            inactive: topic.inactive,
182            cause: topic.cause.as_ref().map(Into::into),
183        }
184    }
185}
186
187impl From<&BarrierError> for StreamBarrierFailure {
188    fn from(error: &BarrierError) -> Self {
189        let BarrierError::Incomplete { reason, unfinished } = error;
190        Self {
191            reason: match reason {
192                BarrierFailure::Blocked => StreamBarrierReason::Blocked,
193                BarrierFailure::Deadline => StreamBarrierReason::Deadline,
194                BarrierFailure::Cancelled => StreamBarrierReason::Cancelled,
195            },
196            unfinished: unfinished.iter().map(Into::into).collect(),
197        }
198    }
199}
200
201impl StreamFailureDetails {
202    fn barrier(error: &BarrierError) -> Self {
203        Self {
204            kind: StreamFailureKind::Barrier,
205            code: error.error_code().to_string(),
206            message: "Processing barriers did not complete".into(),
207            retryable: error.is_retryable(),
208            intent_id: None,
209            published_intent_ids: Vec::new(),
210            summary: None,
211            barriers: vec![error.into()],
212        }
213    }
214}
215
216/// Read known wrapper fields explicitly. Transparent errors can skip them in `source()`.
217/// Visit every summary branch so a first error cannot hide sibling obligations.
218pub fn stream_failure_details(error: &(dyn Error + 'static)) -> Option<StreamFailureDetails> {
219    let mut pending = vec![(error, 0usize)];
220    let mut failures = Vec::new();
221    while let Some((error, depth)) = pending.pop() {
222        if depth >= 64 {
223            continue;
224        }
225        let mut push = |error| pending.push((error, depth + 1));
226        #[cfg(not(target_arch = "wasm32"))]
227        if let Some(error) = error.downcast_ref::<CatchUpError>() {
228            match error {
229                CatchUpError::Group(group) => push(group),
230                CatchUpError::Incomplete { summary, causes } => {
231                    failures.push(StreamFailureDetails {
232                        kind: StreamFailureKind::CatchUp,
233                        code: error.error_code().to_string(),
234                        message: "Catch-up did not complete".into(),
235                        retryable: error.is_retryable(),
236                        intent_id: None,
237                        published_intent_ids: Vec::new(),
238                        summary: Some(StreamFailureSummary {
239                            messages: summary.messages.to_string(),
240                            conversations: summary.conversations.to_string(),
241                            failed: summary.failed.to_string(),
242                            completed: summary.completed,
243                        }),
244                        barriers: causes.iter().map(Into::into).collect(),
245                    })
246                }
247            }
248            continue;
249        }
250        if let Some(error) = error.downcast_ref::<GroupError>() {
251            match error {
252                GroupError::StreamBarrier(error) => {
253                    failures.push(StreamFailureDetails::barrier(error))
254                }
255                GroupError::PublishedButUnconfirmed { intent_id, cause } => {
256                    failures.push(StreamFailureDetails {
257                        kind: StreamFailureKind::PublishedButUnconfirmed,
258                        code: error.error_code().to_string(),
259                        message: "The published intent is not confirmed".into(),
260                        retryable: error.is_retryable(),
261                        intent_id: Some(*intent_id),
262                        published_intent_ids: vec![*intent_id],
263                        summary: None,
264                        barriers: cause.iter().map(|cause| cause.as_ref().into()).collect(),
265                    })
266                }
267                GroupError::Sync(summary) | GroupError::SyncFailedToWait(summary) => {
268                    push(summary.as_ref())
269                }
270                GroupError::Client(error) => push(error),
271                _ => {
272                    if let Some(source) = error.source() {
273                        push(source);
274                    }
275                }
276            }
277            continue;
278        }
279        if let Some(error) = error.downcast_ref::<BarrierError>() {
280            failures.push(StreamFailureDetails::barrier(error));
281            continue;
282        }
283        if let Some(summary) = error.downcast_ref::<SyncSummary>() {
284            for (_, error) in summary.process.errored.iter().rev() {
285                push(error);
286            }
287            for error in summary.post_commit_errors.iter().rev() {
288                push(error);
289            }
290            for error in summary.publish_errors.iter().rev() {
291                push(error);
292            }
293            if let Some(error) = summary.other.as_deref() {
294                push(error);
295            }
296            continue;
297        }
298        if let Some(ClientError::Group(error)) = error.downcast_ref::<ClientError>() {
299            push(error.as_ref());
300        } else if let Some(SubscribeError::Group(error)) = error.downcast_ref::<SubscribeError>() {
301            push(error.as_ref());
302        } else if let Some(IncomingError::Group(error)) = error.downcast_ref::<IncomingError>() {
303            push(error);
304        } else if let Some(error) = error.downcast_ref::<GroupMessageProcessingError>() {
305            match error {
306                GroupMessageProcessingError::PreparedAttempt(error) => push(error.as_ref()),
307                GroupMessageProcessingError::Client(error) => push(error),
308                _ => {
309                    if let Some(source) = error.source() {
310                        push(source);
311                    }
312                }
313            }
314        } else if let Some(source) = error.source() {
315            push(source);
316        }
317    }
318    let mut failures = failures.into_iter();
319    let mut result = failures.next()?;
320    for failure in failures {
321        result.retryable |= failure.retryable;
322        result.barriers.extend(failure.barriers);
323        result
324            .published_intent_ids
325            .extend(failure.published_intent_ids);
326    }
327    result.published_intent_ids.sort_unstable();
328    result.published_intent_ids.dedup();
329    result.intent_id = match result.published_intent_ids.as_slice() {
330        [id] => Some(*id),
331        _ => None,
332    };
333    Some(result)
334}
335
336/// Append this suffix after the existing error code and message.
337pub fn encode_stream_failure(error: &(dyn Error + 'static)) -> Option<String> {
338    let details = stream_failure_details(error)?;
339    let json = serde_json::to_string(&details).ok()?;
340    Some(format!("{STREAM_FAILURE_MARKER}{json}"))
341}
342
343/// Decode the versioned suffix. Ordinary errors have no structured details.
344pub fn decode_stream_failure(message: &str) -> Option<StreamFailureDetails> {
345    let (_, json) = message.rsplit_once(STREAM_FAILURE_MARKER)?;
346    serde_json::from_str(json).ok()
347}
348
349#[cfg(test)]
350mod tests;
xmtp_mls/subscriptions/stream_failure.rs

xmtp_mls/subscriptions/
stream_failure.rs
