offline_protocol/
mls_observability.rs

1//! MLS lifecycle observability primitives.
2
3use crate::telemetry::CategorySampler;
4use offline_protocol_mls::MlsError;
5use serde::Serialize;
6
7/// Operation context for MLS lifecycle events.
8#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
9#[serde(rename_all = "snake_case")]
10pub enum MlsOperationContext {
11    /// Outbound send path.
12    Send,
13    /// Inbound receive path.
14    Receive,
15    /// MLS manager initialization path.
16    Initialize,
17    /// Session lookup or resolution path.
18    SessionLookup,
19    /// Welcome processing path.
20    Welcome,
21}
22
23/// Typed error category for MLS lifecycle events.
24#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
25#[serde(rename_all = "snake_case")]
26pub enum MlsErrorCategory {
27    /// Session or group state is missing.
28    SessionStateMissing,
29    /// MLS manager is not initialized.
30    NotInitialized,
31    /// Ciphertext/message shape is invalid.
32    InvalidCiphertext,
33    /// Sender identity or signature verification failed.
34    IdentityMismatch,
35    /// Crypto engine operation failed.
36    CryptoFailure,
37    /// Unknown or uncategorized failure.
38    Unknown,
39}
40
41/// Fine-grained decryption failure classification.
42#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
43#[serde(rename_all = "snake_case")]
44pub enum DecryptionFailureKind {
45    /// Session or group state was not found.
46    SessionNotFound,
47    /// MLS manager is not initialized.
48    NotInitialized,
49    /// Message bytes cannot be decrypted/parsed as valid MLS ciphertext.
50    InvalidCiphertext,
51    /// Signature/identity verification failed.
52    IdentityMismatch,
53    /// Other cryptographic failure.
54    CryptoFailure,
55    /// Fallback class for unsupported variants.
56    Unknown,
57}
58
59impl DecryptionFailureKind {
60    /// Classifies an MLS error into a decryption failure kind.
61    pub fn from_mls_error(error: &MlsError) -> Self {
62        match error {
63            MlsError::SessionNotFound(_) | MlsError::GroupNotFound(_) => Self::SessionNotFound,
64            MlsError::NotInitialized => Self::NotInitialized,
65            MlsError::InvalidMessage(_)
66            | MlsError::Deserialization(_)
67            | MlsError::Decryption(_) => Self::InvalidCiphertext,
68            MlsError::VerificationFailed(_) | MlsError::InvalidPublicKey(_) => {
69                Self::IdentityMismatch
70            }
71            MlsError::Encryption(_)
72            | MlsError::OpenMls(_)
73            | MlsError::Signing(_)
74            | MlsError::Storage(_)
75            | MlsError::CryptoGeneration(_)
76            // An epoch desync only reaches here on the recovery-disabled
77            // fall-through (recovery-enabled desyncs never take the drop path);
78            // classify it as a crypto failure rather than the opaque `Unknown`.
79            | MlsError::SessionDesync(_) => Self::CryptoFailure,
80            _ => Self::Unknown,
81        }
82    }
83
84    /// Maps a decryption failure kind to its broader telemetry category.
85    pub fn error_category(self) -> MlsErrorCategory {
86        match self {
87            Self::SessionNotFound => MlsErrorCategory::SessionStateMissing,
88            Self::NotInitialized => MlsErrorCategory::NotInitialized,
89            Self::InvalidCiphertext => MlsErrorCategory::InvalidCiphertext,
90            Self::IdentityMismatch => MlsErrorCategory::IdentityMismatch,
91            Self::CryptoFailure => MlsErrorCategory::CryptoFailure,
92            Self::Unknown => MlsErrorCategory::Unknown,
93        }
94    }
95}
96
97/// Typed MLS lifecycle events for operational observability.
98#[allow(missing_docs)]
99#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
100#[serde(tag = "type", rename_all = "snake_case")]
101pub enum MlsLifecycleEvent {
102    /// MLS subsystem has been initialized and is ready for use.
103    Initialized {
104        timestamp_ms: i64,
105        session_id: String,
106        group_id: Option<String>,
107        peer_id: Option<String>,
108        context: MlsOperationContext,
109        error_category: Option<MlsErrorCategory>,
110    },
111    /// Encryption path was successfully used.
112    EncryptionUsed {
113        timestamp_ms: i64,
114        session_id: String,
115        group_id: Option<String>,
116        peer_id: Option<String>,
117        context: MlsOperationContext,
118        error_category: Option<MlsErrorCategory>,
119    },
120    /// Decryption failed with a typed failure class.
121    DecryptionFailed {
122        timestamp_ms: i64,
123        session_id: String,
124        group_id: Option<String>,
125        peer_id: Option<String>,
126        context: MlsOperationContext,
127        error_category: Option<MlsErrorCategory>,
128        failure_kind: DecryptionFailureKind,
129    },
130    /// Session lookup failed or session/group state is missing.
131    SessionMissing {
132        timestamp_ms: i64,
133        session_id: String,
134        group_id: Option<String>,
135        peer_id: Option<String>,
136        context: MlsOperationContext,
137        error_category: Option<MlsErrorCategory>,
138    },
139    /// A handshake with this peer has begun: no session existed, the peer's
140    /// key package was imported and a Welcome was built for them.
141    ///
142    /// This is the *start* of the span `mls_session_ready_latency_p50_ms`
143    /// measures, and the only start that fires on the path a device usually
144    /// takes. [`Self::SessionMissing`] marks a handshake that could not
145    /// begin (no key package on hand, or MLS not initialized); with
146    /// `auto_key_exchange` on, which is the default, a peer's key package
147    /// normally arrives before the first send and the ordinary handshake
148    /// raises no miss at all. Pairing on misses alone samples the slow tail
149    /// and calls it the median.
150    SessionEstablishing {
151        timestamp_ms: i64,
152        session_id: String,
153        group_id: Option<String>,
154        peer_id: Option<String>,
155        context: MlsOperationContext,
156        error_category: Option<MlsErrorCategory>,
157    },
158    /// Session lifecycle has reached a ready/usable state.
159    SessionReady {
160        timestamp_ms: i64,
161        session_id: String,
162        group_id: Option<String>,
163        peer_id: Option<String>,
164        context: MlsOperationContext,
165        error_category: Option<MlsErrorCategory>,
166    },
167}
168
169impl MlsLifecycleEvent {
170    /// Returns the stable `snake.dot.case` telemetry name for this event.
171    ///
172    /// Names are compatible with OpenTelemetry naming conventions and are
173    /// considered stable across minor versions of the SDK.
174    pub fn telemetry_name(&self) -> &'static str {
175        match self {
176            Self::Initialized { .. } => "mls.initialized",
177            Self::EncryptionUsed { .. } => "mls.encryption_used",
178            Self::DecryptionFailed { .. } => "mls.decryption_failed",
179            Self::SessionMissing { .. } => "mls.session_missing",
180            Self::SessionEstablishing { .. } => "mls.session_establishing",
181            Self::SessionReady { .. } => "mls.session_ready",
182        }
183    }
184}
185
186/// Sink abstraction for MLS lifecycle events.
187pub trait MlsEventEmitter: Send + Sync {
188    /// Emits a structured MLS lifecycle event.
189    fn emit(&self, event: MlsLifecycleEvent);
190}
191
192/// Default no-op emitter used when observability is not configured.
193#[derive(Debug, Default)]
194pub struct NoopMlsEventEmitter;
195
196impl MlsEventEmitter for NoopMlsEventEmitter {
197    fn emit(&self, _event: MlsLifecycleEvent) {}
198}
199
200/// Best-effort fixed-window limiter to reduce high-volume failure event floods.
201///
202/// Thin wrapper around [`CategorySampler`] that maps MLS lifecycle events to
203/// rate-limit keys.
204#[derive(Debug, Default)]
205pub struct MlsEventRateLimiter {
206    sampler: CategorySampler,
207}
208
209impl MlsEventRateLimiter {
210    /// Returns true if the event passes rate limiting.
211    pub fn should_emit(&self, event: &MlsLifecycleEvent) -> bool {
212        let now_ms = timestamp_now_ms();
213        match event {
214            MlsLifecycleEvent::DecryptionFailed {
215                peer_id,
216                failure_kind,
217                ..
218            } => {
219                let key = format!(
220                    "decryption_failed:{}:{:?}",
221                    peer_id.as_deref().unwrap_or("none"),
222                    failure_kind
223                );
224                self.sampler.allow(&key, now_ms)
225            }
226            MlsLifecycleEvent::SessionMissing { peer_id, .. } => {
227                let key = format!("session_missing:{}", peer_id.as_deref().unwrap_or("none"));
228                self.sampler.allow(&key, now_ms)
229            }
230            _ => true,
231        }
232    }
233}
234
235/// Returns current wall-clock time in Unix milliseconds.
236pub fn timestamp_now_ms() -> i64 {
237    chrono::Utc::now().timestamp_millis()
238}
239
240#[cfg(test)]
241mod tests {
242    use super::*;
243
244    #[test]
245    fn decryption_failure_classification_is_typed() {
246        assert_eq!(
247            DecryptionFailureKind::from_mls_error(&MlsError::SessionNotFound("p".to_string())),
248            DecryptionFailureKind::SessionNotFound
249        );
250        assert_eq!(
251            DecryptionFailureKind::from_mls_error(&MlsError::NotInitialized),
252            DecryptionFailureKind::NotInitialized
253        );
254        assert_eq!(
255            DecryptionFailureKind::from_mls_error(&MlsError::InvalidMessage("bad".to_string())),
256            DecryptionFailureKind::InvalidCiphertext
257        );
258        assert_eq!(
259            DecryptionFailureKind::from_mls_error(&MlsError::VerificationFailed("sig".to_string())),
260            DecryptionFailureKind::IdentityMismatch
261        );
262    }
263
264    #[test]
265    fn limiter_suppresses_flood() {
266        let limiter = MlsEventRateLimiter::default();
267        let base_ts = 1000_i64;
268        // First 10 DecryptionFailed events for the same peer+kind pass.
269        for i in 0..10 {
270            let event = MlsLifecycleEvent::DecryptionFailed {
271                timestamp_ms: base_ts + i,
272                session_id: "s".to_string(),
273                group_id: None,
274                peer_id: Some("peer-a".to_string()),
275                context: MlsOperationContext::Receive,
276                error_category: None,
277                failure_kind: DecryptionFailureKind::InvalidCiphertext,
278            };
279            assert!(limiter.should_emit(&event));
280        }
281        // 11th is suppressed.
282        let event = MlsLifecycleEvent::DecryptionFailed {
283            timestamp_ms: base_ts + 10,
284            session_id: "s".to_string(),
285            group_id: None,
286            peer_id: Some("peer-a".to_string()),
287            context: MlsOperationContext::Receive,
288            error_category: None,
289            failure_kind: DecryptionFailureKind::InvalidCiphertext,
290        };
291        assert!(!limiter.should_emit(&event));
292    }
293}
offline_protocol/mls_observability.rs

offline_protocol/
mls_observability.rs
