From ef60dd267bc8d1ce867fca2c159b199e7d698468 Mon Sep 17 00:00:00 2001 From: Hugh Kaznowski Date: Sat, 5 Sep 2026 22:34:13 +0100 Subject: [PATCH] Unify PromQL rate() and alert Threshold/Ratio windowing (ticket 040) Extract the trailing-window rate/count implementation into a new src/rate.rs, shared by PromQL's rate() (engine::evaluate_promql, via a thin i64<->Timestamp adapter) and alert Threshold/Ratio conditions (alerts::evaluate_condition, via the new incremental RateWindow), replacing alerts' bespoke VecDeque+prune loop. Both now agree on what "windowed rate" means for the same data, verified by a new regression test (engine::tests::dashboard_rate_and_alert_threshold_agree_on_same_windowed_data). Behavior change: alert Threshold/Ratio windows are now backfilled once, on first evaluation, from already-ingested logs within their own trailing window (bounded cost, not a full history rescan) - a rule created mid-session, or a process that replays existing log content (e.g. --file tailing, which rereads the whole file from the start on every process start), no longer starts with a silently empty window. A bare restart with no replayed history (e.g. stdin ingestion) still starts empty, since there is genuinely no log history to backfill from in that case - alert state itself remains in-memory only. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01XFUNZwCvz3kcod3352ZR2o --- src/alerts.rs | 133 +++++++++++++++++++++++++++++++--------------- src/engine.rs | 108 +++++++++++++++++++++++++++++++++----- src/main.rs | 1 + src/rate.rs | 143 ++++++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 328 insertions(+), 57 deletions(-) create mode 100644 src/rate.rs diff --git a/src/alerts.rs b/src/alerts.rs index 5340c21..65f403d 100644 --- a/src/alerts.rs +++ b/src/alerts.rs @@ -1,9 +1,16 @@ //! Alert rules: definitions, trigger evaluation, and ntfy delivery. //! //! Alert queries are restricted to regex/sequence/timestamp/json[field] — a metric[param] comparison would require a full bucket rescan per line, breaking the proportional-cost guarantee. json[field] is fine here since, like regex/sequence/timestamp, it only inspects the line's own text/fields, no index lookup. +//! +//! `Threshold`/`Ratio` windows are maintained via `crate::rate::RateWindow` — the same windowed-rate +//! implementation PromQL's `rate()` uses (`crate::engine::evaluate_promql`) — rather than a +//! bespoke `VecDeque`+prune loop, so the two agree on what "windowed rate" means (ticket 040). A +//! window is also backfilled once, on first evaluation, from already-ingested logs within its own +//! trailing window (see `backfill_window`): a rule created mid-session (or a process that replays +//! existing log content, e.g. `--file` tailing) no longer starts with a silently empty window. use std::collections::hash_map::DefaultHasher; -use std::collections::{HashMap, HashSet, VecDeque}; +use std::collections::{BTreeMap, HashMap, HashSet}; use std::fmt; use std::hash::{Hash, Hasher}; use std::io; @@ -13,9 +20,12 @@ use std::time::{SystemTime, UNIX_EPOCH}; use regex::Regex; use serde::{Deserialize, Serialize}; -use crate::engine::{apply_compare_op, eval_json_comparison}; +use crate::data_value::DataValue; +use crate::engine::{self, apply_compare_op, eval_json_comparison}; use crate::filter_datetime; use crate::filter_parser::{self, CompareOp, FilterExpr, FilterValue}; +use crate::index_key::IndexKey; +use crate::rate::RateWindow; use crate::settings::Settings; use crate::timestamp::Timestamp; @@ -135,24 +145,43 @@ impl Settings { } } -/// Per-rule in-memory evaluation state; lives only for the process's lifetime — not persisted, not backfilled against already-ingested logs. +/// Per-rule in-memory evaluation state; lives only for the process's lifetime — not persisted +/// across a restart with no replayed log history. Each `RateWindow` is backfilled once (see +/// `backfill_window`) from whatever matching history is already in the index at first evaluation, +/// so — unlike before ticket 040 — a rule doesn't start blind to matches ingested before it was +/// declared, or before this evaluation loop last started, as long as that history is still in the +/// in-memory log index (e.g. `--file` tailing replays existing file content from the start on +/// every process start). #[derive(Debug, Default)] pub struct AlertRuntimeState { - threshold_matches: VecDeque, - ratio_a_matches: VecDeque, - ratio_b_matches: VecDeque, + threshold_window: RateWindow, + threshold_backfilled: bool, + ratio_a_window: RateWindow, + ratio_a_backfilled: bool, + ratio_b_window: RateWindow, + ratio_b_backfilled: bool, seen_values: HashSet, } -/// `timestamps` is push_back-ordered, so once the front entry is within the window, every entry behind it is too. -fn prune_window(timestamps: &mut VecDeque, now_ms: i64, window_ms: i64) { - while let Some(&front) = timestamps.front() { - if now_ms - front > window_ms { - timestamps.pop_front(); - } else { - break; - } - } +/// Builds a `RateWindow` seeded from already-ingested logs matching `query`, bounded to +/// `window_ms` before `now` (not the full log history) so the cost stays proportional to the +/// window rather than everything ever ingested. `before_sequence` excludes the line currently +/// being evaluated, which the caller pushes separately if it matches. +fn backfill_window( + index: &BTreeMap, + query: &str, + before_sequence: u64, + now: Timestamp, + window_ms: i64, +) -> RateWindow { + let since_ms = now.as_millis() - window_ms; + let matching_timestamps = + engine::logs_since(index, before_sequence, since_ms, now.as_millis()).filter_map( + |(sequence, timestamp, text)| { + (eval_alert_query(query, sequence, timestamp, text) == Ok(true)).then_some(timestamp) + }, + ); + RateWindow::backfill(matching_timestamps, now, window_ms) } /// Evaluates a FilterBar query against a single line's own fields only (metric[param] comparisons aren't supported — see module doc; json[field] is supported since it's also line-local). @@ -205,6 +234,7 @@ fn eval_alert_expr(expr: &FilterExpr, sequence: u64, timestamp: Timestamp, text: pub fn evaluate_alerts( rules: &[AlertRule], states: &mut HashMap, + index: &BTreeMap, sequence: u64, timestamp: Timestamp, text: &str, @@ -215,7 +245,9 @@ pub fn evaluate_alerts( continue; } let state = states.entry(rule.id.clone()).or_default(); - if let Some(message) = evaluate_condition(&rule.condition, state, sequence, timestamp, text, &rule.name) { + if let Some(message) = + evaluate_condition(&rule.condition, state, index, sequence, timestamp, text, &rule.name) + { fired.push((rule.clone(), message)); } } @@ -225,6 +257,7 @@ pub fn evaluate_alerts( fn evaluate_condition( condition: &AlertCondition, state: &mut AlertRuntimeState, + index: &BTreeMap, sequence: u64, timestamp: Timestamp, text: &str, @@ -240,12 +273,14 @@ fn evaluate_condition( } }, AlertCondition::Threshold { query, op, threshold, window_ms } => { + if !state.threshold_backfilled { + state.threshold_window = backfill_window(index, query, sequence, timestamp, *window_ms); + state.threshold_backfilled = true; + } match eval_alert_query(query, sequence, timestamp, text) { Ok(true) => { - let now_ms = timestamp.as_millis(); - state.threshold_matches.push_back(now_ms); - prune_window(&mut state.threshold_matches, now_ms, *window_ms); - let count = state.threshold_matches.len() as f64; + state.threshold_window.push(timestamp, *window_ms); + let count = state.threshold_window.count() as f64; apply_compare_op(count.total_cmp(threshold), *op).then(|| { format!("{rule_name}: match count {count} {op:?} threshold {threshold} (window {window_ms}ms)") }) @@ -285,17 +320,22 @@ fn evaluate_condition( if !matched_a && !matched_b { return None; } - let now_ms = timestamp.as_millis(); + if !state.ratio_a_backfilled { + state.ratio_a_window = backfill_window(index, query_a, sequence, timestamp, *window_ms); + state.ratio_a_backfilled = true; + } + if !state.ratio_b_backfilled { + state.ratio_b_window = backfill_window(index, query_b, sequence, timestamp, *window_ms); + state.ratio_b_backfilled = true; + } if matched_a { - state.ratio_a_matches.push_back(now_ms); - prune_window(&mut state.ratio_a_matches, now_ms, *window_ms); + state.ratio_a_window.push(timestamp, *window_ms); } if matched_b { - state.ratio_b_matches.push_back(now_ms); - prune_window(&mut state.ratio_b_matches, now_ms, *window_ms); + state.ratio_b_window.push(timestamp, *window_ms); } - let count_a = state.ratio_a_matches.len() as f64; - let count_b = state.ratio_b_matches.len() as f64; + let count_a = state.ratio_a_window.count() as f64; + let count_b = state.ratio_b_window.count() as f64; if count_b == 0.0 { // Zero denominator: skip rather than treat the ratio as infinite. return None; @@ -417,12 +457,13 @@ mod tests { fn immediate_fires_on_matching_line_only() { let rules = vec![immediate_rule(r#"regex("error")"#)]; let mut states = HashMap::new(); + let index: BTreeMap = BTreeMap::new(); let (seq, ts, text) = line("all good", 0, 0); - assert!(evaluate_alerts(&rules, &mut states, seq, ts, text).is_empty()); + assert!(evaluate_alerts(&rules, &mut states, &index, seq, ts, text).is_empty()); let (seq, ts, text) = line("an error happened", 1, 1); - let fired = evaluate_alerts(&rules, &mut states, seq, ts, text); + let fired = evaluate_alerts(&rules, &mut states, &index, seq, ts, text); assert_eq!(fired.len(), 1); assert!(fired[0].1.contains("an error happened")); } @@ -432,8 +473,9 @@ mod tests { let mut rule = immediate_rule(r#"regex("error")"#); rule.enabled = false; let mut states = HashMap::new(); + let index: BTreeMap = BTreeMap::new(); let (seq, ts, text) = line("error!", 0, 0); - assert!(evaluate_alerts(&[rule], &mut states, seq, ts, text).is_empty()); + assert!(evaluate_alerts(&[rule], &mut states, &index, seq, ts, text).is_empty()); } #[test] @@ -450,14 +492,15 @@ mod tests { }, }; let mut states = HashMap::new(); + let index: BTreeMap = BTreeMap::new(); for (seq, ms) in [(0u64, 0i64), (1, 1000)] { - let fired = evaluate_alerts(&[rule.clone()], &mut states, seq, Timestamp::from_millis(ms), "error"); + let fired = evaluate_alerts(&[rule.clone()], &mut states, &index, seq, Timestamp::from_millis(ms), "error"); assert!(fired.is_empty(), "should not fire before threshold is crossed"); } - let fired = evaluate_alerts(&[rule.clone()], &mut states, 2, Timestamp::from_millis(2000), "error"); + let fired = evaluate_alerts(&[rule.clone()], &mut states, &index, 2, Timestamp::from_millis(2000), "error"); assert_eq!(fired.len(), 1); // Stays quiet for a non-matching line even though the window is full. - let fired = evaluate_alerts(&[rule], &mut states, 3, Timestamp::from_millis(2100), "fine"); + let fired = evaluate_alerts(&[rule], &mut states, &index, 3, Timestamp::from_millis(2100), "fine"); assert!(fired.is_empty()); } @@ -475,9 +518,10 @@ mod tests { }, }; let mut states = HashMap::new(); - assert!(evaluate_alerts(&[rule.clone()], &mut states, 0, Timestamp::from_millis(0), "error").is_empty()); + let index: BTreeMap = BTreeMap::new(); + assert!(evaluate_alerts(&[rule.clone()], &mut states, &index, 0, Timestamp::from_millis(0), "error").is_empty()); // Second match arrives well after the first has aged out of the window. - assert!(evaluate_alerts(&[rule], &mut states, 1, Timestamp::from_millis(5_000), "error").is_empty()); + assert!(evaluate_alerts(&[rule], &mut states, &index, 1, Timestamp::from_millis(5_000), "error").is_empty()); } #[test] @@ -492,11 +536,12 @@ mod tests { }, }; let mut states = HashMap::new(); - let fired = evaluate_alerts(&[rule.clone()], &mut states, 0, Timestamp::from_millis(0), "agent=curl"); + let index: BTreeMap = BTreeMap::new(); + let fired = evaluate_alerts(&[rule.clone()], &mut states, &index, 0, Timestamp::from_millis(0), "agent=curl"); assert_eq!(fired.len(), 1); - let fired = evaluate_alerts(&[rule.clone()], &mut states, 1, Timestamp::from_millis(1), "agent=curl"); + let fired = evaluate_alerts(&[rule.clone()], &mut states, &index, 1, Timestamp::from_millis(1), "agent=curl"); assert!(fired.is_empty()); - let fired = evaluate_alerts(&[rule], &mut states, 2, Timestamp::from_millis(2), "agent=wget"); + let fired = evaluate_alerts(&[rule], &mut states, &index, 2, Timestamp::from_millis(2), "agent=wget"); assert_eq!(fired.len(), 1); } @@ -515,16 +560,17 @@ mod tests { }, }; let mut states = HashMap::new(); + let index: BTreeMap = BTreeMap::new(); // Only an "error" so far; no "request" yet, so denominator is 0 -> skip. - assert!(evaluate_alerts(&[rule.clone()], &mut states, 0, Timestamp::from_millis(0), "error").is_empty()); + assert!(evaluate_alerts(&[rule.clone()], &mut states, &index, 0, Timestamp::from_millis(0), "error").is_empty()); // One request: ratio 1/1 = 1.0 >= 0.5 -> fires. - let fired = evaluate_alerts(&[rule.clone()], &mut states, 1, Timestamp::from_millis(1), "request"); + let fired = evaluate_alerts(&[rule.clone()], &mut states, &index, 1, Timestamp::from_millis(1), "request"); assert_eq!(fired.len(), 1); // A second request: ratio 1/2 = 0.5 still >= 0.5 -> fires again. - let fired = evaluate_alerts(&[rule.clone()], &mut states, 2, Timestamp::from_millis(2), "request"); + let fired = evaluate_alerts(&[rule.clone()], &mut states, &index, 2, Timestamp::from_millis(2), "request"); assert_eq!(fired.len(), 1); // A third request with no more errors: ratio drops to 1/3 -> no longer fires. - let fired = evaluate_alerts(&[rule], &mut states, 3, Timestamp::from_millis(3), "request"); + let fired = evaluate_alerts(&[rule], &mut states, &index, 3, Timestamp::from_millis(3), "request"); assert!(fired.is_empty()); } @@ -532,8 +578,9 @@ mod tests { fn metric_comparison_in_alert_query_is_rejected_not_panicking() { let rule = immediate_rule("some_metric[param]==5"); let mut states = HashMap::new(); + let index: BTreeMap = BTreeMap::new(); // Should log a warning and simply not fire, never panic. - let fired = evaluate_alerts(&[rule], &mut states, 0, Timestamp::from_millis(0), "anything"); + let fired = evaluate_alerts(&[rule], &mut states, &index, 0, Timestamp::from_millis(0), "anything"); assert!(fired.is_empty()); } diff --git a/src/engine.rs b/src/engine.rs index 71f8144..b8ce8b3 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -9,6 +9,7 @@ use crate::filter_parser::{self, CompareOp, FilterExpr, FilterValue}; use crate::index_key::IndexKey; use crate::metric_primitive::{MetricPrimitive, Number}; use crate::promql::{self, AggOp, LabelMatcher, PromqlQuery}; +use crate::rate; use crate::settings::SettingsHandle; use crate::timestamp::Timestamp; use regex::Regex; @@ -161,7 +162,8 @@ pub async fn run(mut rx: mpsc::Receiver, settings: SettingsHandle let settings = settings.lock().unwrap(); (settings.alert_rules.clone(), settings.ntfy_topic.clone(), settings.ntfy_server.clone()) }; - let fired = alerts::evaluate_alerts(&rules, &mut alert_states, sequence, timestamp, &text); + let fired = + alerts::evaluate_alerts(&rules, &mut alert_states, &index, sequence, timestamp, &text); for (rule, message) in fired { tokio::spawn(alerts::deliver_for_rule(rule.name, message, ntfy_topic.clone(), ntfy_server.clone())); } @@ -962,6 +964,24 @@ fn logs_range(index: &BTreeMap) -> impl Iterator, + before_sequence: u64, + since_ms: i64, + now_ms: i64, +) -> impl Iterator { + logs_range(index).filter(move |(sequence, timestamp, _)| { + *sequence < before_sequence + && timestamp.as_millis() > since_ms + && timestamp.as_millis() <= now_ms + }) +} + fn handle_query( index: &BTreeMap, query: EngineQuery, @@ -1159,20 +1179,17 @@ fn project_labels( .collect() } -/// Turns one series' occurrence points into a windowed rate: at each of the series' own timestamps, -/// counts how many points fall in the trailing `(ts - range_ms, ts]` window and divides by the window -/// width in seconds. Points are occurrence markers rather than cumulative counters, so this reads as -/// "occurrences per second over the trailing window", evaluated at every point instead of just the last. +/// Thin adapter over the single shared windowed-rate implementation (`crate::rate`, ticket 040): +/// PromQL's series here are plain millisecond `i64` timestamps rather than `Timestamp`, so this +/// just converts at the boundary and delegates — the actual window-scan logic lives only in +/// `crate::rate::windowed_rate_series`, also used (via `crate::rate::RateWindow`) by alert +/// `Threshold`/`Ratio` conditions in `crate::alerts`, so both agree on what "windowed rate" means. fn windowed_rate_series(points: &[(i64, f64)], range_ms: i64) -> Vec<(i64, f64)> { - let range_seconds = range_ms as f64 / 1000.0; - points - .iter() - .map(|(ts, _)| { - let window_start = ts - range_ms; - let count_in_window = - points.iter().filter(|(t, _)| *t > window_start && *t <= *ts).count(); - (*ts, count_in_window as f64 / range_seconds) - }) + let points: Vec<(Timestamp, f64)> = + points.iter().map(|(ms, value)| (Timestamp::from_millis(*ms), *value)).collect(); + rate::windowed_rate_series(&points, range_ms) + .into_iter() + .map(|(ts, value)| (ts.as_millis(), value)) .collect() } @@ -2058,6 +2075,69 @@ mod tests { assert_eq!(series[1].points, vec![(0, 0.5)]); } + /// Ticket 040 acceptance criterion: a dashboard `rate()` chart and an alert `Threshold` + /// condition over the same metric/window agree on the computed rate for the same historical + /// data, since both now go through the single shared implementation in `crate::rate` rather + /// than each scanning/windowing independently. + #[test] + fn dashboard_rate_and_alert_threshold_agree_on_same_windowed_data() { + let mut index = BTreeMap::new(); + declare_requests_metric(&mut index); + + // Five matching lines 500ms apart; `ingest_log`'s `sequence` doubles as the millisecond + // timestamp (see its definition above), so these are also the log timestamps. + let lines: Vec<(u64, &str)> = vec![ + (0, "request path=/a"), + (500, "request path=/a"), + (1000, "request path=/a"), + (1500, "request path=/a"), + (2000, "request path=/a"), + ]; + for (ms, text) in &lines { + ingest_log(&mut index, *ms, text); + } + + // Dashboard side: rate() over a 2s window, read at the series' own last timestamp. + let parsed = promql::parse_query("rate(requests[2s])").unwrap(); + let series = evaluate_promql(&index, &parsed, None, None); + assert_eq!(series.len(), 1); + let (_, dashboard_rate) = *series[0].points.last().unwrap(); + + // Alert side: a Threshold condition over the same query/window, replayed over the same + // lines in the same order. threshold=0/op=Ge always fires, so every fired message's + // reported match count is available to compare against the dashboard's rate. + let rule = alerts::AlertRule { + id: "r1".to_string(), + name: "spike".to_string(), + enabled: true, + condition: alerts::AlertCondition::Threshold { + query: r#"regex("request path=/a")"#.to_string(), + op: CompareOp::Ge, + threshold: 0.0, + window_ms: 2_000, + }, + }; + let mut alert_states = HashMap::new(); + let mut last_message = String::new(); + for (ms, text) in &lines { + let timestamp = Timestamp::from_millis(*ms as i64); + let fired = alerts::evaluate_alerts(&[rule.clone()], &mut alert_states, &index, *ms, timestamp, text); + assert_eq!(fired.len(), 1); + last_message = fired[0].1.clone(); + } + let alert_count: f64 = last_message + .split("match count ") + .nth(1) + .and_then(|rest| rest.split_whitespace().next()) + .and_then(|token| token.parse().ok()) + .expect("fired message should report a match count"); + + let window_seconds = 2.0; + assert_eq!(dashboard_rate, 2.0); + assert_eq!(alert_count, 4.0); + assert_eq!(alert_count, dashboard_rate * window_seconds); + } + #[test] fn sum_of_rate_without_by_clause_still_collapses_to_one_series() { let mut index = BTreeMap::new(); diff --git a/src/main.rs b/src/main.rs index bb3fdf4..d01d55e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -10,6 +10,7 @@ mod ingest; mod metric_primitive; mod profile; mod promql; +mod rate; mod settings; mod timestamp; diff --git a/src/rate.rs b/src/rate.rs new file mode 100644 index 0000000..71ace08 --- /dev/null +++ b/src/rate.rs @@ -0,0 +1,143 @@ +//! Ticket 040: the single windowed-rate/count implementation in the codebase, shared by +//! PromQL's `rate()` (`crate::engine::evaluate_promql`, via the `windowed_rate_series` adapter +//! there) and alert `Threshold`/`Ratio` conditions (`crate::alerts`). Previously each computed +//! its own trailing window independently — PromQL scanning indexed bucket data, alerts scanning +//! raw log-line matches with a hand-rolled `VecDeque` — and could disagree for equivalent-looking +//! rules simply because the window boundary was implemented twice. +//! +//! `count_in_window` is the one place that decides what "falls in the trailing window" means; +//! everything else in this module is built on top of it, either by batch-rescanning a full point +//! series (`rate_at`/`windowed_rate_series`, fine for PromQL's historical/one-shot evaluation) or +//! incrementally (`RateWindow`, for alerts, which see one point at a time and need cost +//! proportional to the window rather than a full rescan per ingested line). + +use std::collections::VecDeque; + +use crate::timestamp::Timestamp; + +/// How many `points` fall in the trailing `(now - window_ms, now]` window. Points are occurrence +/// markers rather than a continuous reading, so only their timestamps matter — the boundary is +/// exclusive on the old end and inclusive of `now` itself, matching the original +/// `windowed_rate_series` behavior this replaces. +fn count_in_window(points: &[(Timestamp, f64)], now: Timestamp, window_ms: i64) -> usize { + let now_ms = now.as_millis(); + let window_start = now_ms - window_ms; + points + .iter() + .filter(|(ts, _)| { + let ms = ts.as_millis(); + ms > window_start && ms <= now_ms + }) + .count() +} + +/// Trailing-window occurrence rate (points per second) ending at `now`: how many `points` fall in +/// `(now - window_ms, now]`, divided by the window width in seconds. +pub fn rate_at(points: &[(Timestamp, f64)], now: Timestamp, window_ms: i64) -> f64 { + let window_seconds = window_ms as f64 / 1000.0; + count_in_window(points, now, window_ms) as f64 / window_seconds +} + +/// Evaluates `rate_at` at every one of `points`' own timestamps, turning an occurrence series +/// into a windowed rate series — used by PromQL's `rate()` so it plots as a real line rather than +/// one trailing point. +pub fn windowed_rate_series(points: &[(Timestamp, f64)], window_ms: i64) -> Vec<(Timestamp, f64)> { + points.iter().map(|(ts, _)| (*ts, rate_at(points, *ts, window_ms))).collect() +} + +/// Incrementally maintained trailing window: for a caller that sees points arrive one at a time +/// (alert `Threshold`/`Ratio` conditions, evaluated per ingested line) and needs cost proportional +/// to the window rather than `count_in_window`'s full rescan per call. Assumes points are pushed +/// in non-decreasing timestamp order, true for ingested log lines. +#[derive(Debug, Default, Clone)] +pub struct RateWindow { + timestamps: VecDeque, +} + +impl RateWindow { + /// Builds a window pre-populated from already-seen history (e.g. currently-ingested logs + /// matching an alert's query), so a freshly-created window starts with whatever of that + /// history already falls in the trailing window instead of empty. + pub fn backfill(timestamps: impl IntoIterator, now: Timestamp, window_ms: i64) -> Self { + let mut window = Self::default(); + for ts in timestamps { + window.push(ts, window_ms); + } + window.prune(now, window_ms); + window + } + + /// Records one occurrence at `timestamp` and drops anything that has aged out of the + /// resulting trailing window. + pub fn push(&mut self, timestamp: Timestamp, window_ms: i64) { + self.timestamps.push_back(timestamp); + self.prune(timestamp, window_ms); + } + + /// Same boundary as `count_in_window`: drops entries at or before `now - window_ms`. + fn prune(&mut self, now: Timestamp, window_ms: i64) { + let window_start = now.as_millis() - window_ms; + while let Some(&front) = self.timestamps.front() { + if front.as_millis() <= window_start { + self.timestamps.pop_front(); + } else { + break; + } + } + } + + pub fn count(&self) -> usize { + self.timestamps.len() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn rate_at_matches_hand_computed_window() { + let points = vec![ + (Timestamp::from_millis(0), 1.0), + (Timestamp::from_millis(1000), 1.0), + (Timestamp::from_millis(2000), 1.0), + ]; + // (1000, 2000]: only the point at 2000. + assert_eq!(rate_at(&points, Timestamp::from_millis(2000), 1000), 1.0); + // (0, 2000]: points at 1000 and 2000 -> 2 points / 2s. + assert_eq!(rate_at(&points, Timestamp::from_millis(2000), 2000), 1.0); + } + + #[test] + fn windowed_rate_series_matches_rate_at_at_each_point() { + let points = vec![ + (Timestamp::from_millis(0), 1.0), + (Timestamp::from_millis(500), 1.0), + (Timestamp::from_millis(1000), 1.0), + ]; + let series = windowed_rate_series(&points, 1000); + for (ts, rate) in &series { + assert_eq!(*rate, rate_at(&points, *ts, 1000)); + } + } + + #[test] + fn rate_window_incremental_count_matches_batch_count_in_window() { + let mut window = RateWindow::default(); + let millis = [0, 200, 400, 1200]; + for ms in millis { + window.push(Timestamp::from_millis(ms), 1000); + } + let points: Vec<(Timestamp, f64)> = + millis.iter().map(|ms| (Timestamp::from_millis(*ms), 1.0)).collect(); + assert_eq!(window.count(), count_in_window(&points, Timestamp::from_millis(1200), 1000)); + } + + #[test] + fn rate_window_backfill_seeds_from_history_within_the_window() { + let history = [0, 500, 2000, 2500].map(Timestamp::from_millis); + // now=2500ms, window=1000ms -> window (1500, 2500]: only 2000 and 2500 qualify. + let window = RateWindow::backfill(history, Timestamp::from_millis(2500), 1000); + assert_eq!(window.count(), 2); + } +}