faucet_cli/sla/
state.rs

1//! Persisted SLA history: last-success timestamp + rolling volume baseline.
2//!
3//! Stored in the pipeline's `StateStore` under `{base_state_key}::__sla__`
4//! (mirroring the `{name}::__replication__` reserved-suffix convention), so
5//! the history rides whatever durability the user configured for bookmarks.
6
7use serde::{Deserialize, Serialize};
8use serde_json::Value;
9
10/// Reserved suffix appended to the invocation's state key.
11pub const SLA_STATE_SUFFIX: &str = "__sla__";
12
13/// The SLA-history key for one invocation: `{base}::__sla__`.
14pub fn sla_state_key(base: &str) -> String {
15    format!("{base}::{SLA_STATE_SUFFIX}")
16}
17
18/// Rolling SLA history for one root invocation.
19#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
20#[serde(default)]
21pub struct SlaState {
22    /// Unix seconds of the most recent successful run.
23    pub last_success_unix: Option<i64>,
24    /// Records written by recent successful runs, oldest first, trimmed to the
25    /// configured window.
26    pub volumes: Vec<u64>,
27}
28
29impl SlaState {
30    /// Decode a stored value; a corrupt/foreign shape degrades to an empty
31    /// history (with a warning) rather than failing the run.
32    pub fn from_value(v: Value) -> Self {
33        match serde_json::from_value(v) {
34            Ok(s) => s,
35            Err(e) => {
36                tracing::warn!(error = %e, "unreadable SLA state — starting a fresh baseline");
37                Self::default()
38            }
39        }
40    }
41
42    /// Encode for the state store.
43    pub fn to_value(&self) -> Value {
44        serde_json::to_value(self).unwrap_or(Value::Null)
45    }
46
47    /// Fold a successful run into the history, trimming to `window` volumes.
48    pub fn record_success(&mut self, rows: u64, now_unix: i64, window: usize) {
49        self.last_success_unix = Some(now_unix);
50        self.volumes.push(rows);
51        if self.volumes.len() > window {
52            let excess = self.volumes.len() - window;
53            self.volumes.drain(..excess);
54        }
55    }
56}
57
58#[cfg(test)]
59mod tests {
60    use super::*;
61    use serde_json::json;
62
63    #[test]
64    fn key_is_suffixed() {
65        assert_eq!(sla_state_key("orders::default"), "orders::default::__sla__");
66    }
67
68    #[test]
69    fn round_trips_through_value() {
70        let mut s = SlaState::default();
71        s.record_success(100, 1_750_000_000, 20);
72        s.record_success(120, 1_750_003_600, 20);
73        let v = s.to_value();
74        assert_eq!(SlaState::from_value(v), s);
75        assert_eq!(s.volumes, vec![100, 120]);
76        assert_eq!(s.last_success_unix, Some(1_750_003_600));
77    }
78
79    #[test]
80    fn corrupt_value_degrades_to_default() {
81        assert_eq!(
82            SlaState::from_value(json!({"volumes": "not-an-array"})),
83            SlaState::default()
84        );
85        assert_eq!(SlaState::from_value(json!([1, 2, 3])), SlaState::default());
86    }
87
88    #[test]
89    fn missing_fields_default() {
90        // A future field addition must not invalidate old stored state.
91        let s = SlaState::from_value(json!({"last_success_unix": 5}));
92        assert_eq!(s.last_success_unix, Some(5));
93        assert!(s.volumes.is_empty());
94    }
95
96    #[test]
97    fn window_trims_oldest_first() {
98        let mut s = SlaState::default();
99        for i in 0..25u64 {
100            s.record_success(i, i as i64, 20);
101        }
102        assert_eq!(s.volumes.len(), 20);
103        assert_eq!(s.volumes[0], 5);
104        assert_eq!(*s.volumes.last().unwrap(), 24);
105    }
106}
faucet_cli/sla/state.rs

faucet_cli/sla/
state.rs
