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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XFUNZwCvz3kcod3352ZR2o

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<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()));
                 }
@@ -962,6 +964,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,
@@ -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();