offline_protocol/telemetry/
sink.rs

1//! Cross-cutting telemetry sink trait.
2//!
3//! [`TelemetrySink`] is the single observer interface for all telemetry the
4//! SDK emits. Apps implement it (or install [`NoopTelemetrySink`]) and the SDK
5//! pushes every [`TelemetryRecord`] through it.
6//!
7//! Emit wiring lives in follow-up work; this module ships the types only.
8
9use super::record::TelemetryRecord;
10use crate::events::Event;
11
12/// Sink for structured telemetry records.
13///
14/// Implementations must be cheap when called from hot paths — the SDK does not
15/// buffer or rate-limit before dispatch beyond its own sampling layer, so
16/// expensive work (I/O, serialization to external formats) should be
17/// deferred to a background task owned by the implementor.
18///
19/// # Reentrancy and locking contract
20///
21/// The SDK may invoke [`TelemetrySink::emit`] while holding internal locks
22/// (most notably the shared-state mutex on the `Protocol` fan-out path).
23/// Implementations MUST NOT:
24///
25/// - block on slow I/O (network, disk, logging sinks that fsync), or
26/// - reenter SDK methods on the same `OfflineProtocol` instance.
27///
28/// A blocking `emit` stalls every protocol operation that needs the lock.
29/// Sinks that need to do real work should push the record onto a bounded
30/// channel and return immediately, leaving the heavy lifting to a
31/// background task owned by the implementor.
32///
33/// Implementations *should* also not panic, but the SDK defends against
34/// this: every internal dispatch to `emit` runs under `catch_unwind`, so a
35/// panicking sink is logged at `error!`, the offending record is dropped,
36/// and the protocol continues. A panicking sink therefore loses telemetry
37/// and degrades observability but does not corrupt SDK state. Panic
38/// isolation only applies under `panic = unwind`, which every library the
39/// SDK ships uses (ADR 0029). A host built with `panic = abort` aborts the
40/// process on any panic, including one from a sink.
41///
42/// # Record passing
43///
44/// The record is passed by reference so implementations that only read fields
45/// do not incur a clone. Implementations that need ownership (for example to
46/// forward the record to another thread) should call [`Clone::clone`]
47/// explicitly.
48pub trait TelemetrySink: Send + Sync {
49    /// Dispatches a single telemetry record to the sink.
50    fn emit(&self, record: &TelemetryRecord);
51
52    /// Offers a protocol event by reference, before it is scrubbed or
53    /// cloned into a [`TelemetryRecord`].
54    ///
55    /// A sink that returns `true` has consumed the event and the owned,
56    /// scrubbed record is never built; one that returns `false` (the
57    /// default) receives it through [`Self::emit`] as before. The telemetry
58    /// pipe takes this path: it collects ten of the engine's events, reads
59    /// no identifier from any of them, and would otherwise pay a clone of
60    /// every event it drops. A sink that takes this path must do its own
61    /// scrubbing, because the engine's has not run yet.
62    fn try_emit_protocol_event(&self, _event: &Event) -> bool {
63        false
64    }
65}
66
67/// Default sink that discards every record.
68///
69/// Used by the SDK when no telemetry sink has been installed. Proves the
70/// zero-cost baseline: the compiler inlines the empty body at every call site
71/// that can statically resolve the concrete type, and the indirect dispatch
72/// cost through `Arc<dyn TelemetrySink>` is a single virtual call that returns
73/// immediately.
74#[derive(Debug, Default, Clone, Copy)]
75pub struct NoopTelemetrySink;
76
77impl TelemetrySink for NoopTelemetrySink {
78    fn emit(&self, _record: &TelemetryRecord) {}
79}
80
81#[cfg(test)]
82mod tests {
83    use super::*;
84    use crate::mls_observability::{MlsLifecycleEvent, MlsOperationContext};
85    use std::sync::Arc;
86
87    #[test]
88    fn noop_sink_is_send_sync() {
89        fn assert_send_sync<T: Send + Sync>() {}
90        assert_send_sync::<NoopTelemetrySink>();
91    }
92
93    #[test]
94    fn trait_is_object_safe() {
95        let _: Arc<dyn TelemetrySink> = Arc::new(NoopTelemetrySink);
96    }
97
98    #[test]
99    fn noop_sink_accepts_records_without_panic() {
100        let sink = NoopTelemetrySink;
101        let record = TelemetryRecord::Mls(MlsLifecycleEvent::Initialized {
102            timestamp_ms: 0,
103            session_id: "s".into(),
104            group_id: None,
105            peer_id: None,
106            context: MlsOperationContext::Initialize,
107            error_category: None,
108        });
109        sink.emit(&record);
110    }
111}
offline_protocol/telemetry/sink.rs

offline_protocol/telemetry/
sink.rs
