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

slim_datapath/
errors.rs
