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