diff --git a/src/control.rs b/src/control.rs index 3284645..03b22ed 100644 --- a/src/control.rs +++ b/src/control.rs @@ -90,10 +90,7 @@ impl StatePublisher { safe_message, recovery_attempt, }; - if state.transitions.len() == TRANSITION_CAPACITY { - state.transitions.pop_front(); - } - state.transitions.push_back(transition.clone()); + crate::domain::push_transition(&mut state.transitions, transition.clone()); let _ = self.events.send(transition); } state.derive_aggregate(); @@ -123,10 +120,7 @@ impl StatePublisher { safe_message: safe_message.to_owned(), recovery_attempt: None, }; - if state.transitions.len() == TRANSITION_CAPACITY { - state.transitions.pop_front(); - } - state.transitions.push_back(transition.clone()); + crate::domain::push_transition(&mut state.transitions, transition.clone()); let _ = self.events.send(transition); self.snapshots.send_replace(state.clone()); } diff --git a/src/domain.rs b/src/domain.rs index 256674f..355156f 100644 --- a/src/domain.rs +++ b/src/domain.rs @@ -1,4 +1,4 @@ -use std::collections::{BTreeMap, VecDeque}; +use std::collections::{BTreeMap, HashMap, VecDeque}; use serde::{Deserialize, Serialize}; @@ -6,6 +6,47 @@ use crate::state::{ClientState, TrafficState}; pub const TRANSITION_CAPACITY: usize = 200; +/// Append a transition to the bounded snapshot window. +/// +/// The window is shared by every component, so evicting the oldest entry lets +/// whichever checks change most often decide how far back the whole thing +/// reaches. On a busy gateway three high-frequency checks held 98% of it and it +/// went back barely two hours, while `topology.route_intent` and +/// `topology.client_lifecycle` — the entries describing what the gateway +/// actually did to client routing, and the ones an operator goes looking for +/// when a container loses egress — held two slots each and were minutes from +/// being pushed out. +/// +/// Evicting from whichever component holds the most of the window instead lets a +/// chatty component churn against itself. Nothing is lost either way: every +/// transition is also persisted and served from `/api/v2/history/events`. This +/// is about what the at-a-glance view is worth. +pub fn push_transition(transitions: &mut VecDeque, transition: Transition) { + while transitions.len() >= TRANSITION_CAPACITY { + let Some(index) = busiest_components_oldest(transitions) else { + break; + }; + transitions.remove(index); + } + transitions.push_back(transition); +} + +/// The index of the oldest entry belonging to the component holding the most +/// slots. Ties go to whichever of them has been in the window longest. +fn busiest_components_oldest(transitions: &VecDeque) -> Option { + let mut holdings: HashMap<&str, (usize, usize)> = HashMap::new(); + for (index, transition) in transitions.iter().enumerate() { + let entry = holdings + .entry(transition.component.as_str()) + .or_insert((0, index)); + entry.0 += 1; + } + holdings + .into_values() + .max_by_key(|(count, oldest)| (*count, std::cmp::Reverse(*oldest))) + .map(|(_, oldest)| oldest) +} + #[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq, Serialize)] #[serde(rename_all = "snake_case")] pub enum Protection { @@ -295,6 +336,85 @@ impl CanonicalSnapshot { mod tests { use super::*; + fn transition(component: &str, sequence: u64) -> Transition { + Transition { + sequence, + timestamp_unix_ms: sequence, + component: component.to_owned(), + from_status: CheckStatus::Healthy, + to_status: CheckStatus::Degraded, + reason_code: "test".to_owned(), + safe_message: "test".to_owned(), + recovery_attempt: None, + } + } + + fn held(transitions: &VecDeque, component: &str) -> usize { + transitions + .iter() + .filter(|entry| entry.component == component) + .count() + } + + #[test] + fn a_chatty_component_cannot_evict_a_quiet_one() { + // The observed window: three high-frequency checks held 98% of it, and + // the route-intent entries an operator actually goes looking for were + // minutes from being pushed out. + let mut transitions = VecDeque::new(); + push_transition(&mut transitions, transition("topology.route_intent", 0)); + push_transition(&mut transitions, transition("topology.route_intent", 1)); + for sequence in 2..(TRANSITION_CAPACITY as u64 * 4) { + push_transition(&mut transitions, transition("dns.upstream_udp", sequence)); + } + assert_eq!(transitions.len(), TRANSITION_CAPACITY); + assert_eq!( + held(&transitions, "topology.route_intent"), + 2, + "the quiet component was evicted by the chatty one" + ); + } + + #[test] + fn the_window_stays_within_capacity() { + let mut transitions = VecDeque::new(); + for sequence in 0..(TRANSITION_CAPACITY as u64 * 3) { + push_transition(&mut transitions, transition("dns.upstream_udp", sequence)); + } + assert_eq!(transitions.len(), TRANSITION_CAPACITY); + } + + #[test] + fn one_component_on_its_own_still_evicts_oldest_first() { + let mut transitions = VecDeque::new(); + for sequence in 0..(TRANSITION_CAPACITY as u64 + 5) { + push_transition(&mut transitions, transition("dns.upstream_udp", sequence)); + } + assert_eq!(transitions.front().unwrap().sequence, 5); + assert_eq!( + transitions.back().unwrap().sequence, + TRANSITION_CAPACITY as u64 + 4 + ); + } + + #[test] + fn components_converge_on_a_share_rather_than_a_race() { + // Two components pushing at very different rates should end up sharing + // the window, not with the faster one owning it. + let mut transitions = VecDeque::new(); + for sequence in 0..(TRANSITION_CAPACITY as u64 * 5) { + push_transition(&mut transitions, transition("fast", sequence)); + if sequence % 10 == 0 { + push_transition(&mut transitions, transition("slow", sequence)); + } + } + let slow = held(&transitions, "slow"); + assert!( + slow >= TRANSITION_CAPACITY / 4, + "the slow component held only {slow} slots" + ); + } + #[test] fn protection_and_availability_are_independent() { let mut snapshot = CanonicalSnapshot::default();