Merge branch 'worktree-agent-a9ca45be433ffca98'

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<i64>,
-    ratio_a_matches: VecDeque<i64>,
-    ratio_b_matches: VecDeque<i64>,
+    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<String>,
 }
 
-/// `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<i64>, 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<IndexKey, DataValue>,
+    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<String, AlertRuntimeState>,
+    index: &BTreeMap<IndexKey, DataValue>,
     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<IndexKey, DataValue>,
     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<IndexKey, DataValue> = 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<IndexKey, DataValue> = 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<IndexKey, DataValue> = 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<IndexKey, DataValue> = 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<IndexKey, DataValue> = 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<IndexKey, DataValue> = 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<IndexKey, DataValue> = 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());
     }
 

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;
@@ -159,7 +160,8 @@ pub async fn run(mut rx: mpsc::Receiver<EngineMessage>, 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()));
                 }
@@ -1050,6 +1052,24 @@ fn logs_range(index: &BTreeMap<IndexKey, DataValue>) -> impl Iterator<Item = (u6
     })
 }
 
+/// Every currently-ingested log with `sequence < before_sequence` and timestamp in
+/// `(since_ms, now_ms]`, in ascending order. `pub(crate)` so `crate::alerts` can use it to
+/// backfill a freshly-created `Threshold`/`Ratio` window from already-ingested history — bounded
+/// to the alert's own window, not the full log history, so the cost stays proportional to the
+/// window rather than the whole ingest history (ticket 040).
+pub(crate) fn logs_since(
+    index: &BTreeMap<IndexKey, DataValue>,
+    before_sequence: u64,
+    since_ms: i64,
+    now_ms: i64,
+) -> impl Iterator<Item = (u64, Timestamp, &str)> {
+    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<IndexKey, DataValue>,
     query: EngineQuery,
@@ -1250,20 +1270,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()
 }
 
@@ -2307,6 +2324,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();

src/main.rs

@@ -10,6 +10,7 @@ mod ingest;
 mod metric_primitive;
 mod profile;
 mod promql;
+mod rate;
 mod settings;
 mod timestamp;
 

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<Timestamp>,
+}
+
+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<Item = Timestamp>, 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);
+    }
+}