mctx_core/
metrics.rs

1use crate::{Context, Publication};
2use std::sync::atomic::{AtomicU64, Ordering};
3use std::time::{Duration, Instant, SystemTime};
4
5#[derive(Debug, Default)]
6pub(crate) struct MetricsSequence {
7    value: AtomicU64,
8}
9
10pub(crate) struct MetricsWriteGuard<'a> {
11    sequence: &'a MetricsSequence,
12}
13
14impl MetricsSequence {
15    pub(crate) fn write(&self) -> MetricsWriteGuard<'_> {
16        let mut current = self.value.load(Ordering::Relaxed);
17        loop {
18            if current % 2 == 1 {
19                std::hint::spin_loop();
20                current = self.value.load(Ordering::Relaxed);
21                continue;
22            }
23
24            match self.value.compare_exchange_weak(
25                current,
26                current.wrapping_add(1),
27                Ordering::Acquire,
28                Ordering::Relaxed,
29            ) {
30                Ok(_) => return MetricsWriteGuard { sequence: self },
31                Err(observed) => current = observed,
32            }
33        }
34    }
35
36    pub(crate) fn read_consistent<T>(&self, read: impl Fn() -> T) -> T {
37        loop {
38            let before = self.value.load(Ordering::Acquire);
39            if before % 2 == 1 {
40                std::hint::spin_loop();
41                continue;
42            }
43
44            let snapshot = read();
45            // Keep all counter reads between the two sequence observations.
46            std::sync::atomic::fence(Ordering::AcqRel);
47            let after = self.value.load(Ordering::Relaxed);
48            if before == after {
49                return snapshot;
50            }
51        }
52    }
53}
54
55impl Drop for MetricsWriteGuard<'_> {
56    fn drop(&mut self) {
57        self.sequence.value.fetch_add(1, Ordering::Release);
58    }
59}
60
61fn rate_per_sec(count: u64, interval_secs: f64) -> f64 {
62    if interval_secs > 0.0 {
63        count as f64 / interval_secs
64    } else {
65        0.0
66    }
67}
68
69/// A point-in-time snapshot of cumulative publication metrics.
70///
71/// Counter fields in this snapshot are cumulative from the lifetime of the
72/// publication and can be compared against an earlier snapshot to compute
73/// deltas and rates.
74#[derive(Debug, Clone, PartialEq, Eq)]
75pub struct PublicationMetricsSnapshot {
76    pub send_calls: u64,
77    pub packets_sent: u64,
78    pub bytes_sent: u64,
79    pub send_errors: u64,
80    pub captured_at: SystemTime,
81}
82
83/// The difference between two cumulative publication metrics snapshots.
84///
85/// This contains only counter-based deltas over the sampled interval.
86#[derive(Debug, Clone, PartialEq)]
87pub struct PublicationMetricsDelta {
88    pub interval_secs: f64,
89    pub send_calls: u64,
90    pub packets_sent: u64,
91    pub bytes_sent: u64,
92    pub send_errors: u64,
93}
94
95impl PublicationMetricsSnapshot {
96    /// Computes the counter deltas between this snapshot and an earlier one.
97    ///
98    /// Returns `None` if:
99    /// - `earlier` was captured after `self`
100    /// - any cumulative counter appears to have moved backwards
101    pub fn delta_since(&self, earlier: &Self) -> Option<PublicationMetricsDelta> {
102        let duration = self.captured_at.duration_since(earlier.captured_at).ok()?;
103        self.delta_since_duration(earlier, duration)
104    }
105
106    /// Computes counter deltas using a caller-supplied monotonic interval.
107    pub fn delta_since_duration(
108        &self,
109        earlier: &Self,
110        duration: Duration,
111    ) -> Option<PublicationMetricsDelta> {
112        Some(PublicationMetricsDelta {
113            interval_secs: duration.as_secs_f64(),
114            send_calls: self.send_calls.checked_sub(earlier.send_calls)?,
115            packets_sent: self.packets_sent.checked_sub(earlier.packets_sent)?,
116            bytes_sent: self.bytes_sent.checked_sub(earlier.bytes_sent)?,
117            send_errors: self.send_errors.checked_sub(earlier.send_errors)?,
118        })
119    }
120}
121
122impl PublicationMetricsDelta {
123    /// Returns the average send call count per second over the sampled interval.
124    pub fn send_calls_per_sec(&self) -> f64 {
125        rate_per_sec(self.send_calls, self.interval_secs)
126    }
127
128    /// Returns the average packets sent per second over the sampled interval.
129    pub fn packets_per_sec(&self) -> f64 {
130        rate_per_sec(self.packets_sent, self.interval_secs)
131    }
132
133    /// Returns the average bytes sent per second over the sampled interval.
134    pub fn bytes_per_sec(&self) -> f64 {
135        rate_per_sec(self.bytes_sent, self.interval_secs)
136    }
137
138    /// Returns the average send error count per second over the sampled interval.
139    pub fn send_errors_per_sec(&self) -> f64 {
140        rate_per_sec(self.send_errors, self.interval_secs)
141    }
142}
143
144/// Tracks successive publication metrics snapshots and computes deltas between them.
145#[derive(Debug, Clone)]
146pub struct PublicationMetricsSampler<'a> {
147    publication: &'a Publication,
148    previous: Option<PublicationMetricsSnapshot>,
149    previous_sampled_at: Option<Instant>,
150}
151
152impl<'a> PublicationMetricsSampler<'a> {
153    pub fn new(publication: &'a Publication) -> Self {
154        Self {
155            publication,
156            previous: None,
157            previous_sampled_at: None,
158        }
159    }
160
161    pub fn snapshot(&self) -> PublicationMetricsSnapshot {
162        self.publication.metrics_snapshot()
163    }
164
165    pub fn sample(&mut self) -> Option<PublicationMetricsDelta> {
166        let current = self.snapshot();
167        self.sample_snapshot_at(current, Instant::now())
168    }
169
170    /// Computes a delta from a caller-supplied snapshot using its wall-clock
171    /// `captured_at` timestamp. This clears the monotonic baseline used by
172    /// `sample()` and `sample_snapshot_at()`.
173    pub fn sample_snapshot(
174        &mut self,
175        current: PublicationMetricsSnapshot,
176    ) -> Option<PublicationMetricsDelta> {
177        let delta = self
178            .previous
179            .as_ref()
180            .and_then(|previous| current.delta_since(previous));
181        self.previous = Some(current);
182        self.previous_sampled_at = None;
183        delta
184    }
185
186    /// Computes a delta from a caller-supplied snapshot and monotonic capture
187    /// instant. The first call after `sample_snapshot()` establishes a new
188    /// monotonic baseline and returns `None`.
189    pub fn sample_snapshot_at(
190        &mut self,
191        current: PublicationMetricsSnapshot,
192        sampled_at: Instant,
193    ) -> Option<PublicationMetricsDelta> {
194        let delta = match (&self.previous, self.previous_sampled_at) {
195            (Some(previous), Some(previous_sampled_at)) => sampled_at
196                .checked_duration_since(previous_sampled_at)
197                .and_then(|duration| current.delta_since_duration(previous, duration)),
198            _ => None,
199        };
200        self.previous = Some(current);
201        self.previous_sampled_at = Some(sampled_at);
202        delta
203    }
204
205    pub fn reset(&mut self) {
206        self.previous = None;
207        self.previous_sampled_at = None;
208    }
209
210    pub fn previous(&self) -> Option<&PublicationMetricsSnapshot> {
211        self.previous.as_ref()
212    }
213
214    /// Convenience alias for `sample()`.
215    pub fn delta(&mut self) -> Option<PublicationMetricsDelta> {
216        self.sample()
217    }
218}
219
220/// A point-in-time snapshot of cumulative context metrics.
221///
222/// Counter fields in this snapshot are cumulative from the lifetime of the
223/// context for send activity issued through `Context` methods and can be
224/// compared against an earlier snapshot to compute deltas and rates.
225///
226/// Gauge-like fields such as `active_publications` represent the current state
227/// at the moment the snapshot was taken and should not be interpreted as
228/// cumulative counters.
229#[derive(Debug, Clone, PartialEq, Eq)]
230pub struct ContextMetricsSnapshot {
231    pub publications_added: u64,
232    pub publications_removed: u64,
233    pub active_publications: usize,
234    pub total_send_calls: u64,
235    pub total_packets_sent: u64,
236    pub total_bytes_sent: u64,
237    pub total_send_errors: u64,
238    pub captured_at: SystemTime,
239}
240
241/// The difference between two cumulative context metrics snapshots.
242///
243/// This contains only counter-based deltas over the sampled interval.
244/// Gauge-like values such as active publication counts are intentionally not
245/// included here; callers should inspect those directly from the latest
246/// snapshot instead.
247#[derive(Debug, Clone, PartialEq)]
248pub struct ContextMetricsDelta {
249    pub interval_secs: f64,
250    pub publications_added: u64,
251    pub publications_removed: u64,
252    pub send_calls: u64,
253    pub packets_sent: u64,
254    pub bytes_sent: u64,
255    pub send_errors: u64,
256}
257
258impl ContextMetricsSnapshot {
259    /// Computes the counter deltas between this snapshot and an earlier one.
260    ///
261    /// Returns `None` if:
262    /// - `earlier` was captured after `self`
263    /// - any cumulative counter appears to have moved backwards
264    ///
265    /// The resulting delta contains only counter-based values and the elapsed
266    /// interval in seconds. Gauge-like values such as active publication counts
267    /// should be read directly from the latest snapshot instead.
268    pub fn delta_since(&self, earlier: &Self) -> Option<ContextMetricsDelta> {
269        let duration = self.captured_at.duration_since(earlier.captured_at).ok()?;
270        self.delta_since_duration(earlier, duration)
271    }
272
273    /// Computes counter deltas using a caller-supplied monotonic interval.
274    pub fn delta_since_duration(
275        &self,
276        earlier: &Self,
277        duration: Duration,
278    ) -> Option<ContextMetricsDelta> {
279        Some(ContextMetricsDelta {
280            interval_secs: duration.as_secs_f64(),
281            publications_added: self
282                .publications_added
283                .checked_sub(earlier.publications_added)?,
284            publications_removed: self
285                .publications_removed
286                .checked_sub(earlier.publications_removed)?,
287            send_calls: self
288                .total_send_calls
289                .checked_sub(earlier.total_send_calls)?,
290            packets_sent: self
291                .total_packets_sent
292                .checked_sub(earlier.total_packets_sent)?,
293            bytes_sent: self
294                .total_bytes_sent
295                .checked_sub(earlier.total_bytes_sent)?,
296            send_errors: self
297                .total_send_errors
298                .checked_sub(earlier.total_send_errors)?,
299        })
300    }
301}
302
303impl ContextMetricsDelta {
304    /// Returns the average send call count per second over the sampled interval.
305    pub fn send_calls_per_sec(&self) -> f64 {
306        rate_per_sec(self.send_calls, self.interval_secs)
307    }
308
309    /// Returns the average packets sent per second over the sampled interval.
310    pub fn packets_per_sec(&self) -> f64 {
311        rate_per_sec(self.packets_sent, self.interval_secs)
312    }
313
314    /// Returns the average bytes sent per second over the sampled interval.
315    pub fn bytes_per_sec(&self) -> f64 {
316        rate_per_sec(self.bytes_sent, self.interval_secs)
317    }
318
319    /// Returns the average send error count per second over the sampled interval.
320    pub fn send_errors_per_sec(&self) -> f64 {
321        rate_per_sec(self.send_errors, self.interval_secs)
322    }
323}
324
325/// Tracks successive context metrics snapshots and computes deltas between them.
326#[derive(Debug, Clone)]
327pub struct ContextMetricsSampler<'a> {
328    context: &'a Context,
329    previous: Option<ContextMetricsSnapshot>,
330    previous_sampled_at: Option<Instant>,
331}
332
333impl<'a> ContextMetricsSampler<'a> {
334    pub fn new(context: &'a Context) -> Self {
335        Self {
336            context,
337            previous: None,
338            previous_sampled_at: None,
339        }
340    }
341
342    pub fn snapshot(&self) -> ContextMetricsSnapshot {
343        self.context.metrics_snapshot()
344    }
345
346    pub fn sample(&mut self) -> Option<ContextMetricsDelta> {
347        let current = self.snapshot();
348        self.sample_snapshot_at(current, Instant::now())
349    }
350
351    /// Computes a delta from a caller-supplied snapshot using its wall-clock
352    /// `captured_at` timestamp. This clears the monotonic baseline used by
353    /// `sample()` and `sample_snapshot_at()`.
354    pub fn sample_snapshot(
355        &mut self,
356        current: ContextMetricsSnapshot,
357    ) -> Option<ContextMetricsDelta> {
358        let delta = self
359            .previous
360            .as_ref()
361            .and_then(|previous| current.delta_since(previous));
362        self.previous = Some(current);
363        self.previous_sampled_at = None;
364        delta
365    }
366
367    /// Computes a delta from a caller-supplied snapshot and monotonic capture
368    /// instant. The first call after `sample_snapshot()` establishes a new
369    /// monotonic baseline and returns `None`.
370    pub fn sample_snapshot_at(
371        &mut self,
372        current: ContextMetricsSnapshot,
373        sampled_at: Instant,
374    ) -> Option<ContextMetricsDelta> {
375        let delta = match (&self.previous, self.previous_sampled_at) {
376            (Some(previous), Some(previous_sampled_at)) => sampled_at
377                .checked_duration_since(previous_sampled_at)
378                .and_then(|duration| current.delta_since_duration(previous, duration)),
379            _ => None,
380        };
381        self.previous = Some(current);
382        self.previous_sampled_at = Some(sampled_at);
383        delta
384    }
385
386    pub fn reset(&mut self) {
387        self.previous = None;
388        self.previous_sampled_at = None;
389    }
390
391    pub fn previous(&self) -> Option<&ContextMetricsSnapshot> {
392        self.previous.as_ref()
393    }
394
395    /// Convenience alias for `sample()`.
396    pub fn delta(&mut self) -> Option<ContextMetricsDelta> {
397        self.sample()
398    }
399}
400
401#[cfg(test)]
402mod tests {
403    use super::*;
404    use std::time::Duration;
405
406    fn publication_snapshot(
407        send_calls: u64,
408        packets_sent: u64,
409        bytes_sent: u64,
410        send_errors: u64,
411        captured_at: SystemTime,
412    ) -> PublicationMetricsSnapshot {
413        PublicationMetricsSnapshot {
414            send_calls,
415            packets_sent,
416            bytes_sent,
417            send_errors,
418            captured_at,
419        }
420    }
421
422    #[test]
423    fn context_delta_since_uses_lifetime_total_fields() {
424        let earlier = ContextMetricsSnapshot {
425            publications_added: 1,
426            publications_removed: 0,
427            active_publications: 1,
428            total_send_calls: 10,
429            total_packets_sent: 8,
430            total_bytes_sent: 800,
431            total_send_errors: 2,
432            captured_at: SystemTime::UNIX_EPOCH,
433        };
434        let later = ContextMetricsSnapshot {
435            publications_added: 2,
436            publications_removed: 1,
437            active_publications: 1,
438            total_send_calls: 14,
439            total_packets_sent: 11,
440            total_bytes_sent: 1250,
441            total_send_errors: 3,
442            captured_at: SystemTime::UNIX_EPOCH + Duration::from_secs(2),
443        };
444
445        let delta = later.delta_since(&earlier).unwrap();
446
447        assert_eq!(delta.interval_secs, 2.0);
448        assert_eq!(delta.publications_added, 1);
449        assert_eq!(delta.publications_removed, 1);
450        assert_eq!(delta.send_calls, 4);
451        assert_eq!(delta.packets_sent, 3);
452        assert_eq!(delta.bytes_sent, 450);
453        assert_eq!(delta.send_errors, 1);
454        assert_eq!(delta.packets_per_sec(), 1.5);
455        assert_eq!(delta.bytes_per_sec(), 225.0);
456    }
457
458    #[test]
459    fn publication_delta_since_uses_interval_and_rates() {
460        let earlier = publication_snapshot(4, 3, 300, 1, SystemTime::UNIX_EPOCH);
461        let later = publication_snapshot(
462            7,
463            5,
464            620,
465            2,
466            SystemTime::UNIX_EPOCH + Duration::from_secs(4),
467        );
468
469        let delta = later.delta_since(&earlier).unwrap();
470
471        assert_eq!(delta.interval_secs, 4.0);
472        assert_eq!(delta.send_calls, 3);
473        assert_eq!(delta.packets_sent, 2);
474        assert_eq!(delta.bytes_sent, 320);
475        assert_eq!(delta.send_errors, 1);
476        assert_eq!(delta.send_calls_per_sec(), 0.75);
477        assert_eq!(delta.packets_per_sec(), 0.5);
478        assert_eq!(delta.bytes_per_sec(), 80.0);
479        assert_eq!(delta.send_errors_per_sec(), 0.25);
480    }
481
482    #[test]
483    fn context_sampler_uses_monotonic_interval_when_wall_clock_moves_backwards() {
484        let context = Context::new();
485        let mut sampler = ContextMetricsSampler::new(&context);
486        let sampled_at = Instant::now();
487        let earlier = ContextMetricsSnapshot {
488            publications_added: 1,
489            publications_removed: 0,
490            active_publications: 1,
491            total_send_calls: 1,
492            total_packets_sent: 1,
493            total_bytes_sent: 10,
494            total_send_errors: 0,
495            captured_at: SystemTime::UNIX_EPOCH + Duration::from_secs(10),
496        };
497        let later = ContextMetricsSnapshot {
498            total_send_calls: 2,
499            total_packets_sent: 2,
500            total_bytes_sent: 20,
501            captured_at: SystemTime::UNIX_EPOCH + Duration::from_secs(5),
502            ..earlier.clone()
503        };
504
505        assert!(sampler.sample_snapshot_at(earlier, sampled_at).is_none());
506        let delta = sampler
507            .sample_snapshot_at(later, sampled_at + Duration::from_secs(2))
508            .unwrap();
509
510        assert_eq!(delta.interval_secs, 2.0);
511        assert_eq!(delta.packets_sent, 1);
512        assert_eq!(delta.bytes_sent, 10);
513    }
514
515    #[test]
516    fn context_sampler_uses_supplied_snapshot_timestamps() {
517        let context = Context::new();
518        let mut sampler = ContextMetricsSampler::new(&context);
519        let earlier = ContextMetricsSnapshot {
520            publications_added: 1,
521            publications_removed: 0,
522            active_publications: 1,
523            total_send_calls: 1,
524            total_packets_sent: 1,
525            total_bytes_sent: 10,
526            total_send_errors: 0,
527            captured_at: SystemTime::UNIX_EPOCH + Duration::from_secs(10),
528        };
529        let later = ContextMetricsSnapshot {
530            total_send_calls: 2,
531            total_packets_sent: 2,
532            total_bytes_sent: 20,
533            captured_at: SystemTime::UNIX_EPOCH + Duration::from_secs(15),
534            ..earlier.clone()
535        };
536
537        assert!(sampler.sample_snapshot(earlier).is_none());
538        let delta = sampler.sample_snapshot(later).unwrap();
539
540        assert_eq!(delta.interval_secs, 5.0);
541        assert_eq!(delta.packets_sent, 1);
542    }
543}
mctx_core/metrics.rs

mctx_core/
metrics.rs
