slim_datapath/
errors.rs

1// Copyright AGNTCY Contributors (https://github.com/agntcy)
2// SPDX-License-Identifier: Apache-2.0
3
4use std::string;
5
6use crate::api::ProtoName;
7use crate::api::ProtoSessionMessageType;
8use crate::api::proto::dataplane::v1::Message;
9use crate::messages::utils::MessageError;
10use agntcy_slim_proto::NameError;
11#[cfg(not(target_arch = "wasm32"))]
12use slim_config::errors::ConfigError;
13use thiserror::Error;
14
15/// DataPath and subscription table errors merged into a single enum.
16#[derive(Error, Debug)]
17pub enum DataPathError {
18    // Connection lifecycle
19    #[error("connection error")]
20    ConnectionError,
21    #[error("disconnection error")]
22    DisconnectionError(u64),
23    #[cfg(not(target_arch = "wasm32"))]
24    #[error("grpc error")]
25    GrpcError(#[from] tonic::Status),
26    #[error("link negotiation error {0}")]
27    NegotiationError(String),
28
29    // Message classification / validation
30    #[error("unknown message type")]
31    UnknownMsgType,
32    #[error("invalid message: {0}")]
33    InvalidMessage(MessageError),
34    #[error("invalid name id format: {0}")]
35    InvalidNameIdFormat(String),
36    #[error("invalid name format: {0}")]
37    InvalidNameFormat(String),
38
39    // Subscription / matching
40    #[error("no matching found for [{:x}, {:x}, {:x}, {}]", .0, .1, .2, .3)]
41    NoMatchEncoded(u64, u64, u64, String),
42    #[error("subscription not found")]
43    SubscriptionNotFound(ProtoName),
44    #[error("subscription id not found: {0}")]
45    SubscriptionIdNotFound(u64),
46    #[error("id not found: {0}")]
47    IdNotFound(string::String),
48
49    // Connection lookup
50    #[error("connection not found: {0}")]
51    ConnectionNotFound(u64),
52    #[error("connection id not found: {0}")]
53    ConnectionIdNotFound(u64),
54
55    // Processing
56    #[error("malformed message")]
57    MalformedMessage(#[from] MessageError),
58    #[error("message processing error: {0}")]
59    ProcessingError(MessageError),
60    #[error("connection send error")]
61    ConnectionSendError,
62    #[error("error adding connection to connection table")]
63    ConnectionTableAddError,
64    #[error("message processing error: {source}")]
65    MessageProcessingError {
66        #[source]
67        source: Box<DataPathError>,
68        msg: Box<Message>,
69    },
70
71    // Configuration error
72    #[cfg(not(target_arch = "wasm32"))]
73    #[error("configuration error")]
74    ConfigurationError(#[from] ConfigError),
75
76    // Remote subscription ACK errors
77    #[error("remote subscription ack timed out after {0} retries")]
78    RemoteSubscriptionAckTimeout(u32),
79
80    #[error("remote subscription ack error")]
81    RemoteSubscriptionAckError(String),
82
83    // Shutdown errors
84    #[error("data path is already closed")]
85    AlreadyClosedError,
86    #[error("data plane is shutting down")]
87    ShuttingDownError,
88    #[error("timeout during shutdown")]
89    ShutdownTimeoutError,
90
91    #[cfg(not(target_arch = "wasm32"))]
92    #[error("SLIM header integrity: {0}")]
93    HeaderIntegrity(#[from] crate::header_mac::HeaderMacError),
94
95    #[error("header MAC requires completed link negotiation on connection {0}")]
96    HeaderMacAwaitingLinkNegotiation(u64),
97
98    #[error("inter-node ephemeral key generation failed")]
99    LinkKeyGeneration,
100
101    #[error("message TTL expired")]
102    TtlExpired,
103}
104
105#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
106pub struct MessageContext {
107    pub message_id: u32,
108    pub session_id: u32,
109    pub session_message_type: i32,
110}
111
112impl From<NameError> for DataPathError {
113    fn from(value: NameError) -> Self {
114        match value {
115            NameError::InvalidNameIdFormat(name) => Self::InvalidNameIdFormat(name),
116            NameError::InvalidNameFormat(name) => Self::InvalidNameFormat(name),
117        }
118    }
119}
120
121impl MessageContext {
122    pub fn from_msg(msg: &Message) -> Option<Self> {
123        msg.try_get_session_header().map(|header| Self {
124            message_id: header.get_message_id(),
125            session_id: header.get_session_id(),
126            session_message_type: header.session_message_type().into(),
127        })
128    }
129
130    pub fn get_session_message_type(&self) -> ProtoSessionMessageType {
131        self.session_message_type
132            .try_into()
133            .unwrap_or(ProtoSessionMessageType::Unspecified)
134    }
135}
136
137/// A unified error payload that includes an error message and optional session context.
138/// This type is used to serialize/deserialize errors sent over gRPC with consistent JSON structure.
139#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
140pub struct ErrorPayload {
141    pub error: String,
142    #[serde(skip_serializing_if = "Option::is_none")]
143    pub session_context: Option<MessageContext>,
144}
145
146impl std::fmt::Display for ErrorPayload {
147    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
148        write!(f, "ErrorPayload: {}", self.error)?;
149        match &self.session_context {
150            Some(ctx) => write!(
151                f,
152                " (session_id={}, message_id={}, session_message_type={:?})",
153                ctx.session_id,
154                ctx.message_id,
155                ctx.get_session_message_type()
156            ),
157            None => Ok(()),
158        }
159    }
160}
161
162impl ErrorPayload {
163    /// Create a new error payload
164    pub fn new(error: String, session_context: Option<MessageContext>) -> Self {
165        Self {
166            error,
167            session_context,
168        }
169    }
170
171    /// Convert to JSON string for transmission
172    pub fn to_json_string(&self) -> String {
173        serde_json::to_string(self).expect("ErrorPayload should be serializable")
174    }
175
176    /// Parse from JSON string
177    pub fn from_json_str(s: &str) -> Option<Self> {
178        serde_json::from_str(s).ok()
179    }
180}
slim_datapath/errors.rs

slim_datapath/
errors.rs
