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    /// 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            target: topic.target.map(|cursor| cursor.0.to_string()),
168            received: topic.received.0.to_string(),
169            processed: topic.processed.0.to_string(),
170            unresolved_welcomes: topic
171                .unresolved_welcomes
172                .iter()
173                .map(|cursor| cursor.0.to_string())
174                .collect(),
175            inactive: topic.inactive,
176            cause: topic.cause.as_ref().map(Into::into),
177        }
178    }
179}
180
181impl From<&BarrierError> for StreamBarrierFailure {
182    fn from(error: &BarrierError) -> Self {
183        let BarrierError::Incomplete { reason, unfinished } = error;
184        Self {
185            reason: match reason {
186                BarrierFailure::Blocked => StreamBarrierReason::Blocked,
187                BarrierFailure::Deadline => StreamBarrierReason::Deadline,
188                BarrierFailure::Cancelled => StreamBarrierReason::Cancelled,
189            },
190            unfinished: unfinished.iter().map(Into::into).collect(),
191        }
192    }
193}
194
195impl StreamFailureDetails {
196    fn barrier(error: &BarrierError) -> Self {
197        Self {
198            kind: StreamFailureKind::Barrier,
199            code: error.error_code().to_string(),
200            message: "Processing barriers did not complete".into(),
201            retryable: error.is_retryable(),
202            intent_id: None,
203            published_intent_ids: Vec::new(),
204            summary: None,
205            barriers: vec![error.into()],
206        }
207    }
208}
209
210/// Read known wrapper fields explicitly. Transparent errors can skip them in `source()`.
211/// Visit every summary branch so a first error cannot hide sibling obligations.
212pub fn stream_failure_details(error: &(dyn Error + 'static)) -> Option<StreamFailureDetails> {
213    let mut pending = vec![(error, 0usize)];
214    let mut failures = Vec::new();
215    while let Some((error, depth)) = pending.pop() {
216        if depth >= 64 {
217            continue;
218        }
219        let mut push = |error| pending.push((error, depth + 1));
220        #[cfg(not(target_arch = "wasm32"))]
221        if let Some(error) = error.downcast_ref::<CatchUpError>() {
222            match error {
223                CatchUpError::Group(group) => push(group),
224                CatchUpError::Incomplete { summary, causes } => {
225                    failures.push(StreamFailureDetails {
226                        kind: StreamFailureKind::CatchUp,
227                        code: error.error_code().to_string(),
228                        message: "Catch-up did not complete".into(),
229                        retryable: error.is_retryable(),
230                        intent_id: None,
231                        published_intent_ids: Vec::new(),
232                        summary: Some(StreamFailureSummary {
233                            messages: summary.messages.to_string(),
234                            conversations: summary.conversations.to_string(),
235                            failed: summary.failed.to_string(),
236                            completed: summary.completed,
237                        }),
238                        barriers: causes.iter().map(Into::into).collect(),
239                    })
240                }
241            }
242            continue;
243        }
244        if let Some(error) = error.downcast_ref::<GroupError>() {
245            match error {
246                GroupError::StreamBarrier(error) => {
247                    failures.push(StreamFailureDetails::barrier(error))
248                }
249                GroupError::PublishedButUnconfirmed { intent_id, cause } => {
250                    failures.push(StreamFailureDetails {
251                        kind: StreamFailureKind::PublishedButUnconfirmed,
252                        code: error.error_code().to_string(),
253                        message: "The published intent is not confirmed".into(),
254                        retryable: error.is_retryable(),
255                        intent_id: Some(*intent_id),
256                        published_intent_ids: vec![*intent_id],
257                        summary: None,
258                        barriers: cause.iter().map(|cause| cause.as_ref().into()).collect(),
259                    })
260                }
261                GroupError::Sync(summary) | GroupError::SyncFailedToWait(summary) => {
262                    push(summary.as_ref())
263                }
264                GroupError::Client(error) => push(error),
265                _ => {
266                    if let Some(source) = error.source() {
267                        push(source);
268                    }
269                }
270            }
271            continue;
272        }
273        if let Some(error) = error.downcast_ref::<BarrierError>() {
274            failures.push(StreamFailureDetails::barrier(error));
275            continue;
276        }
277        if let Some(summary) = error.downcast_ref::<SyncSummary>() {
278            for (_, error) in summary.process.errored.iter().rev() {
279                push(error);
280            }
281            for error in summary.post_commit_errors.iter().rev() {
282                push(error);
283            }
284            for error in summary.publish_errors.iter().rev() {
285                push(error);
286            }
287            if let Some(error) = summary.other.as_deref() {
288                push(error);
289            }
290            continue;
291        }
292        if let Some(ClientError::Group(error)) = error.downcast_ref::<ClientError>() {
293            push(error.as_ref());
294        } else if let Some(SubscribeError::Group(error)) = error.downcast_ref::<SubscribeError>() {
295            push(error.as_ref());
296        } else if let Some(IncomingError::Group(error)) = error.downcast_ref::<IncomingError>() {
297            push(error);
298        } else if let Some(error) = error.downcast_ref::<GroupMessageProcessingError>() {
299            match error {
300                GroupMessageProcessingError::PreparedAttempt(error) => push(error.as_ref()),
301                GroupMessageProcessingError::Client(error) => push(error),
302                _ => {
303                    if let Some(source) = error.source() {
304                        push(source);
305                    }
306                }
307            }
308        } else if let Some(source) = error.source() {
309            push(source);
310        }
311    }
312    let mut failures = failures.into_iter();
313    let mut result = failures.next()?;
314    for failure in failures {
315        result.retryable |= failure.retryable;
316        result.barriers.extend(failure.barriers);
317        result
318            .published_intent_ids
319            .extend(failure.published_intent_ids);
320    }
321    result.published_intent_ids.sort_unstable();
322    result.published_intent_ids.dedup();
323    result.intent_id = match result.published_intent_ids.as_slice() {
324        [id] => Some(*id),
325        _ => None,
326    };
327    Some(result)
328}
329
330/// Append this suffix after the existing error code and message.
331pub fn encode_stream_failure(error: &(dyn Error + 'static)) -> Option<String> {
332    let details = stream_failure_details(error)?;
333    let json = serde_json::to_string(&details).ok()?;
334    Some(format!("{STREAM_FAILURE_MARKER}{json}"))
335}
336
337/// Decode the versioned suffix. Ordinary errors have no structured details.
338pub fn decode_stream_failure(message: &str) -> Option<StreamFailureDetails> {
339    let (_, json) = message.rsplit_once(STREAM_FAILURE_MARKER)?;
340    serde_json::from_str(json).ok()
341}
342
343#[cfg(test)]
344mod tests;
xmtp_mls/subscriptions/stream_failure.rs

xmtp_mls/subscriptions/
stream_failure.rs
