streaming_crypto/core_api/telemetry/
snapshot.rs

1// ## src/telemetry/snapshot.rs
2
3// //! src/telemetry/snapshot.rs
4// //!
5// //! Telemetry snapshot structures and conversions.
6// //!
7// //! Design notes:
8// //! - `TelemetrySnapshot` is the core Rust struct with rich types (Duration, HashMap).
9// //! - Stage times are flattened into fixed fields for ABI stability.
10// //! - Conversions ensure elapsed time is represented in milliseconds for cross-language parity.
11
12use std::time::Duration;
13use serde::{Serialize, Deserialize};
14
15use crate::telemetry::counters::TelemetryCounters;
16use crate::telemetry::timers::{TelemetryTimer, StageTimes, Stage};
17
18/// Core telemetry snapshot.
19/// Captures counters, ratios, throughput, stage timings, and elapsed duration.
20#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
21pub struct TelemetrySnapshot {
22    pub segments_processed: u64,
23    pub frames_data: u64,
24    pub frames_terminator: u64,
25    pub frames_digest: u64,
26    pub bytes_plaintext: u64,
27    pub bytes_compressed: u64,
28    pub bytes_ciphertext: u64,
29    pub bytes_overhead: u64,
30    pub compression_ratio: f64,
31    pub throughput_plaintext_bytes_per_sec: f64,
32    pub elapsed: Duration,
33    pub stage_times: StageTimes, // HashMap<Stage, Duration>
34    /// The final encrypted stream bytes, if the output sink was memory-backed.
35    /// 
36    /// - `None` if the output was written directly to a file or external sink.
37    /// - `Some(Vec<u8>)` if the pipeline wrote into an in-memory buffer.
38    /// 
39    /// This field is primarily useful in tests, benchmarks, or integrations
40    /// where we want to inspect the produced ciphertext alongside telemetry
41    /// counters and stage timings.
42    pub output: Option<Vec<u8>>,
43}
44
45impl TelemetrySnapshot {
46    pub fn from(counters: &TelemetryCounters, timer: &TelemetryTimer, segments: Option<u32>) -> Self {
47        let elapsed = timer.elapsed();
48
49        let mut compression_ratio = if counters.bytes_plaintext > 0 {
50            counters.bytes_compressed as f64 / counters.bytes_plaintext as f64
51        } else {
52            0.0
53        };
54        compression_ratio = compression_ratio.min(1.0);
55
56        let throughput = if elapsed.as_secs_f64() > 0.0 {
57            counters.bytes_plaintext as f64 / elapsed.as_secs_f64()
58        } else {
59            0.0
60        };
61
62        Self {
63            segments_processed: segments.unwrap_or_default() as u64,
64            frames_data: counters.frames_data,
65            frames_terminator: counters.frames_terminator,
66            frames_digest: counters.frames_digest,
67            bytes_plaintext: counters.bytes_plaintext,
68            bytes_compressed: counters.bytes_compressed,
69            bytes_ciphertext: counters.bytes_ciphertext,
70            bytes_overhead: counters.bytes_overhead,
71            compression_ratio: compression_ratio,
72            throughput_plaintext_bytes_per_sec: throughput,
73            elapsed: elapsed,
74            stage_times: timer.stage_times.clone(),
75            output: None, // 🔧 initialize empty
76        }
77    }
78
79    pub fn total_stage_time(&self) -> Duration {
80        self.stage_times.iter().map(|(_, d)| *d).sum()
81    }
82
83    // - **Stage coverage sanity**  
84    // Add a helper that asserts all expected `Stage` variants are present in `stage_times`. 
85    // This prevents silent omissions when new stages are introduced.  
86    pub fn has_all_stages(&self, expected: &[Stage]) -> bool {
87        expected.iter().all(|s| self.stage_times.get(*s) > Duration::ZERO)
88        // expected.iter().all(|s| {
89        //     self.stage_times.get(s).map_or(false, |d| *d > Duration::ZERO)
90        // })
91    }
92
93    // - **Consistency checks**  
94    // Provide a method that validates internal invariants:  
95    // - `bytes_ciphertext >= bytes_compressed`  
96    // - `compression_ratio <= 1.0`  
97    // - `total_stage_time() <= elapsed`  
98
99    pub fn sanity_check(&self) -> bool {
100        self.bytes_ciphertext >= self.bytes_compressed &&
101        self.compression_ratio <= 1.0 &&
102        self.total_stage_time() <= self.elapsed
103    }
104    
105    pub fn output_bytes(&self) -> u64 {
106        self.bytes_ciphertext
107    }
108
109    /// 🔧 Attach output buffer to snapshot
110    pub fn attach_output(&mut self, buf: Vec<u8>) {
111        self.output = Some(buf);
112    }
113}
114
streaming_crypto/core_api/telemetry/snapshot.rs

streaming_crypto/core_api/telemetry/
snapshot.rs
