digdigdig3_station/cure/
integrity.rs

1//! Integrity check — reads a stored stream, reports stats + anomalies.
2//!
3//! Pure analysis, no mutation.
4
5use std::collections::BTreeMap;
6
7use sha2::{Digest, Sha256};
8
9use crate::storage::{StorageManager, StreamKey};
10
11// ── Public types ──────────────────────────────────────────────────────────────
12
13/// A period longer than the configured threshold with no events.
14#[derive(Debug, Clone)]
15pub struct TimeGap {
16    pub start_ms: i64,
17    pub end_ms: i64,
18    pub duration_ms: i64,
19}
20
21/// A jump in sequence numbers detected in orderbook delta payloads.
22#[derive(Debug, Clone)]
23pub struct SequenceGap {
24    pub from_seq: u64,
25    pub to_seq: u64,
26    pub ts_ms: i64,
27}
28
29/// Full analysis result for one stream range.
30#[derive(Debug, Clone)]
31pub struct IntegrityReport {
32    pub stream: StreamKey,
33    pub from_ms: i64,
34    pub to_ms: i64,
35    pub record_count: u64,
36    pub duplicate_count: u64,
37    pub out_of_order_count: u64,
38    pub parse_errors: u64,
39    /// Periods longer than `time_gap_threshold_ms` with no events.
40    pub time_gaps: Vec<TimeGap>,
41    /// Sequence number jumps detected in orderbook payloads.
42    pub sequence_gaps: Vec<SequenceGap>,
43    pub first_ts: Option<i64>,
44    pub last_ts: Option<i64>,
45    /// Average interval between successive records (ms).
46    pub avg_interval_ms: Option<f64>,
47}
48
49// ── IntegrityChecker ─────────────────────────────────────────────────────────
50
51/// Reads a stream range and produces an [`IntegrityReport`].
52pub struct IntegrityChecker<'a> {
53    storage: &'a StorageManager,
54    /// Minimum quiet period (ms) that is reported as a [`TimeGap`]. Default: 60 000 (1 min).
55    time_gap_threshold_ms: i64,
56}
57
58impl<'a> IntegrityChecker<'a> {
59    pub fn new(storage: &'a StorageManager) -> Self {
60        Self {
61            storage,
62            time_gap_threshold_ms: 60_000,
63        }
64    }
65
66    /// Override the time-gap threshold (milliseconds).
67    pub fn with_time_gap_threshold(mut self, ms: i64) -> Self {
68        self.time_gap_threshold_ms = ms;
69        self
70    }
71
72    /// Run the full integrity analysis over `[from_ms, to_ms]`.
73    pub async fn check(
74        &self,
75        key: &StreamKey,
76        from_ms: i64,
77        to_ms: i64,
78    ) -> std::io::Result<IntegrityReport> {
79        let records = self.storage.read_range(key, from_ms, to_ms).await?;
80
81        let mut report = IntegrityReport {
82            stream: key.clone(),
83            from_ms,
84            to_ms,
85            record_count: 0,
86            duplicate_count: 0,
87            out_of_order_count: 0,
88            parse_errors: 0,
89            time_gaps: Vec::new(),
90            sequence_gaps: Vec::new(),
91            first_ts: None,
92            last_ts: None,
93            avg_interval_ms: None,
94        };
95        report.record_count = records.len() as u64;
96
97        if records.is_empty() {
98            return Ok(report);
99        }
100
101        report.first_ts = records.first().map(|(t, _)| *t);
102        report.last_ts = records.last().map(|(t, _)| *t);
103
104        let mut last_ts: Option<i64> = None;
105        let mut last_seq: Option<u64> = None;
106        // Key: (ts_ms, sha256_first16) → occurrence count
107        let mut seen: BTreeMap<(i64, [u8; 16]), u32> = BTreeMap::new();
108
109        for (ts, payload) in &records {
110            // ── time gap / out-of-order ───────────────────────────────────────
111            if let Some(prev) = last_ts {
112                let delta = *ts - prev;
113                if delta < 0 {
114                    report.out_of_order_count += 1;
115                } else if delta > self.time_gap_threshold_ms {
116                    report.time_gaps.push(TimeGap {
117                        start_ms: prev,
118                        end_ms: *ts,
119                        duration_ms: delta,
120                    });
121                }
122            }
123            last_ts = Some(*ts);
124
125            // ── dedup fingerprint ─────────────────────────────────────────────
126            let hash = sha256_first16(payload);
127            *seen.entry((*ts, hash)).or_insert(0) += 1;
128
129            // ── sequence gap from JSON payload ────────────────────────────────
130            match serde_json::from_slice::<serde_json::Value>(payload) {
131                Ok(v) => {
132                    if let Some(seq) = extract_sequence(&v) {
133                        if let Some(prev) = last_seq {
134                            if seq > prev + 1 {
135                                report.sequence_gaps.push(SequenceGap {
136                                    from_seq: prev,
137                                    to_seq: seq,
138                                    ts_ms: *ts,
139                                });
140                            }
141                        }
142                        last_seq = Some(seq);
143                    }
144                }
145                Err(_) => {
146                    report.parse_errors += 1;
147                }
148            }
149        }
150
151        report.duplicate_count = seen
152            .values()
153            .filter(|&&c| c > 1)
154            .map(|&c| (c - 1) as u64)
155            .sum();
156
157        if let (Some(first), Some(last)) = (report.first_ts, report.last_ts) {
158            if report.record_count > 1 {
159                report.avg_interval_ms =
160                    Some((last - first) as f64 / (report.record_count - 1) as f64);
161            }
162        }
163
164        Ok(report)
165    }
166}
167
168// ── helpers ───────────────────────────────────────────────────────────────────
169
170pub(crate) fn sha256_first16(bytes: &[u8]) -> [u8; 16] {
171    let mut h = Sha256::new();
172    h.update(bytes);
173    let full = h.finalize();
174    let mut out = [0u8; 16];
175    out.copy_from_slice(&full[..16]);
176    out
177}
178
179/// Try to extract a sequence number from common orderbook event JSON shapes.
180///
181/// Tries fields: `last_update_id`, `sequence`, `u`, `seq`.
182fn extract_sequence(v: &serde_json::Value) -> Option<u64> {
183    v.get("last_update_id")
184        .and_then(|x| x.as_u64())
185        .or_else(|| v.get("sequence").and_then(|x| x.as_u64()))
186        .or_else(|| v.get("u").and_then(|x| x.as_u64()))
187        .or_else(|| v.get("seq").and_then(|x| x.as_u64()))
188}
digdigdig3_station/cure/integrity.rs

digdigdig3_station/cure/
integrity.rs
