From 3751508de83c364c29bbb55f686c419838b6ee05 Mon Sep 17 00:00:00 2001 From: Hugh Kaznowski Date: Mon, 7 Sep 2026 13:55:51 +0100 Subject: [PATCH] Move alert evaluation to the profile actor (ticket 105) Alert rules are profile-scoped and read metric buckets, which now live on the profile actor (ticket 104) rather than the primary. alert_states and the per-line evaluate_alerts/deliver_for_rule calls move from the primary's Ingest arm to the profile actor's per-line loop, evaluated against its own snapshot right after that line's apply_all_metrics publishes -- matching the primary's original per-message publish-then- evaluate ordering. The primary's run() no longer needs a SettingsHandle at all now that alerts are gone from its Ingest arm; settings (for alert rules and ntfy config, read fresh per line) moves to engine::profile::spawn instead. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01TeyhRfMYjMditek6geDRoQ --- src/engine/handle.rs | 50 +++--------- src/engine/profile.rs | 80 +++++++++++++++---- ...-move-alert-evaluation-to-profile-actor.md | 52 ------------ 3 files changed, 73 insertions(+), 109 deletions(-) delete mode 100644 tickets/105-move-alert-evaluation-to-profile-actor.md diff --git a/src/engine/handle.rs b/src/engine/handle.rs index a1f9b75..6361517 100644 --- a/src/engine/handle.rs +++ b/src/engine/handle.rs @@ -3,7 +3,6 @@ use crate::config::SettingsHandle; use crate::consts::{INGEST_LAG_WATERMARK_MAX_ENTRIES, INGEST_LAG_WATERMARK_WINDOW_MS}; -use crate::engine::alerts::{self, AlertRuntimeState}; use crate::engine::index_key::IndexKey; use crate::engine::message::EngineMessage; use crate::engine::metric_primitive::{MetricPrimitive, Number}; @@ -12,7 +11,7 @@ use crate::engine::snapshot::params::log_stream_header_transition; use crate::engine::timestamp::Timestamp; use crate::engine::{ClearParamsResult, IngestedLine}; use left_right::Absorb; -use std::collections::{BTreeMap, HashMap}; +use std::collections::BTreeMap; use std::sync::Arc; use tokio::sync::{mpsc, oneshot, watch}; @@ -146,13 +145,13 @@ fn new_channel_and_handle( pub fn spawn(settings: SettingsHandle, channel_capacity: usize) -> EngineHandle { let (tx, rx, reader, write, version_rx) = new_channel_and_handle(channel_capacity); - let profile = crate::engine::profile::spawn(reader.clone(), version_rx); + let profile = crate::engine::profile::spawn(reader.clone(), version_rx, settings); let handle = EngineHandle { tx, reader, profile_reader: profile.reader.clone(), }; - tokio::spawn(run(rx, settings, write, profile)); + tokio::spawn(run(rx, write, profile)); handle } @@ -179,25 +178,24 @@ pub(crate) fn spawn_with_profile_observer( channel_capacity: usize, ) -> (EngineHandle, tokio::sync::mpsc::UnboundedReceiver) { let (tx, rx, reader, write, version_rx) = new_channel_and_handle(channel_capacity); - let (profile, observed) = crate::engine::profile::spawn_observed(reader.clone(), version_rx); + let (profile, observed) = + crate::engine::profile::spawn_observed(reader.clone(), version_rx, settings); let handle = EngineHandle { tx, reader, profile_reader: profile.reader.clone(), }; - tokio::spawn(run(rx, settings, write, profile)); + tokio::spawn(run(rx, write, profile)); (handle, observed) } async fn run( mut rx: mpsc::Receiver, - settings: SettingsHandle, mut write: IndexWriter, profile: crate::engine::profile::ProfileHandle, ) { let mut max_logs: Option = None; let mut metric_sample_sequence: u64 = 0; - let mut alert_states: HashMap = HashMap::new(); let mut log_stream_header: Option = None; let mut ingest_lag_samples: Vec<(Timestamp, i64)> = Vec::new(); @@ -226,48 +224,18 @@ async fn run( log_stream_header.clone() }; - let (rules, ntfy_topic, ntfy_server) = { - let settings = settings.lock().unwrap(); - ( - settings.alert_rules.clone(), - settings.ntfy_topic.clone(), - settings.ntfy_server.clone(), - ) - }; - let text_for_op = Arc::clone(&text); write.mutate(move |snap: &mut EngineSnapshot| { // Primary/param extraction only (ticket 101) -- metric computation - // (ticket 104) now happens on the profile actor, which wakes on this - // publish and reads the newly-visible line itself. + // (ticket 104) and alert evaluation (ticket 105) now happen on the + // profile actor, which wakes on this publish and reads the + // newly-visible line itself. snap.ingest_params(sequence, timestamp, &text_for_op, row_header.as_ref()); snap.update_sort_helper_indexes_on_ingest(sequence, timestamp, &text_for_op); snap.purge_stale_sort_helper_indexes(); snap.evict_excess_logs(max_logs); snap.ingest_lag_high_watermark_ms = ingest_lag_high_watermark_ms; }); - - let fired = write - .peek() - .map(|guard| { - alerts::evaluate_alerts( - &rules, - &mut alert_states, - &guard, - sequence, - timestamp, - &text, - ) - }) - .unwrap_or_default(); - for (rule, message) in fired { - tokio::spawn(alerts::deliver_for_rule( - rule.name, - message, - ntfy_topic.clone(), - ntfy_server.clone(), - )); - } } EngineMessage::IngestMetricSample(sample) => { let labels: BTreeMap = sample diff --git a/src/engine/profile.rs b/src/engine/profile.rs index 2b78857..d9a4d88 100644 --- a/src/engine/profile.rs +++ b/src/engine/profile.rs @@ -1,18 +1,20 @@ -//! The profile actor (ticket 104, building on ticket 103's skeleton). Owns metric declarations -//! and bucket computation, moved off the primary per `ideas/0001-actor-split.md`'s resolved -//! decision that the primary "extracts declared + auto params only... no metrics, no profile -//! logic." Reuses `EngineSnapshot` itself as this actor's own snapshot type — `declare_metric`/ -//! `update_metric`/`remove_metric_declaration`/`apply_all_metrics` all work unchanged — rather -//! than inventing a parallel type, since metric evaluation needs param declarations present in -//! the same snapshot it reads (`resolved_param_declarations`); this actor mirrors the primary's -//! param declarations into its own snapshot on every wake so that stays true, but stores no raw -//! log text of its own, reading it through `primary_reader` (a clone of the primary's -//! `IndexReader`) instead. `EngineHandle`'s public API is unchanged by this ticket — -//! `DeclareMetric`/`UpdateMetric`/`RemoveMetricDeclaration` are still sent to the primary's -//! `mpsc` channel and proxied here by `handle.rs`'s `run()` loop; ticket 106 gives API handlers -//! a direct route. +//! The profile actor (tickets 103-105). Owns metric declarations, bucket computation, and alert +//! evaluation, moved off the primary per `ideas/0001-actor-split.md`'s resolved decision that +//! the primary "extracts declared + auto params only... no metrics, no profile logic" and that +//! alerts move with metric computation since they read the buckets it produces. Reuses +//! `EngineSnapshot` itself as this actor's own snapshot type — `declare_metric`/`update_metric`/ +//! `remove_metric_declaration`/`apply_all_metrics` all work unchanged — rather than inventing a +//! parallel type, since metric evaluation needs param declarations present in the same snapshot +//! it reads (`resolved_param_declarations`); this actor mirrors the primary's param declarations +//! into its own snapshot on every wake so that stays true, but stores no raw log text of its +//! own, reading it through `primary_reader` (a clone of the primary's `IndexReader`) instead. +//! `EngineHandle`'s public API is unchanged by this ticket — `DeclareMetric`/`UpdateMetric`/ +//! `RemoveMetricDeclaration` are still sent to the primary's `mpsc` channel and proxied here by +//! `handle.rs`'s `run()` loop; ticket 106 gives API handlers a direct route. +use crate::config::SettingsHandle; use crate::engine::DeclareMetricRequest; +use crate::engine::alerts::{self, AlertRuntimeState}; use crate::engine::data_value::{DataValue, ParamDeclaration}; use crate::engine::handle::IndexReader; use crate::engine::index_key::IndexKey; @@ -20,6 +22,7 @@ use crate::engine::snapshot::EngineSnapshot; use crate::engine::timestamp::Timestamp; use left_right::Absorb; use regex::Regex; +use std::collections::HashMap; use tokio::sync::{mpsc, oneshot, watch}; struct ProfileOp(Box); @@ -76,16 +79,26 @@ pub(crate) fn inert_reader() -> IndexReader { /// this actor its own read-only view of the primary's index (raw log text, param declarations); /// `primary_version` is bumped by the primary's `IndexWriter::mutate` after every `left_right` /// publish, so this actor blocks on `changed()` between primary writes rather than polling. +/// `settings` is used to read alert rules and ntfy config fresh for each evaluated line (ticket +/// 105 -- alert evaluation moved here from the primary alongside the metric buckets it reads). pub(crate) fn spawn( primary_reader: IndexReader, primary_version: watch::Receiver, + settings: SettingsHandle, ) -> ProfileHandle { let (tx, rx) = mpsc::channel(crate::consts::ENGINE_CHANNEL_CAPACITY); let (write, read) = left_right::new::(); let reader = IndexReader { factory: read.factory(), }; - tokio::spawn(run(primary_reader, primary_version, rx, write, None)); + tokio::spawn(run( + primary_reader, + primary_version, + rx, + write, + settings, + None, + )); ProfileHandle { tx, reader } } @@ -96,6 +109,7 @@ pub(crate) fn spawn( pub(crate) fn spawn_observed( primary_reader: IndexReader, primary_version: watch::Receiver, + settings: SettingsHandle, ) -> (ProfileHandle, mpsc::UnboundedReceiver) { let (tx, rx) = mpsc::channel(crate::consts::ENGINE_CHANNEL_CAPACITY); let (write, read) = left_right::new::(); @@ -108,6 +122,7 @@ pub(crate) fn spawn_observed( primary_version, rx, write, + settings, Some(advance_tx), )); (ProfileHandle { tx, reader }, advance_rx) @@ -118,9 +133,11 @@ async fn run( mut primary_version: watch::Receiver, mut rx: mpsc::Receiver, mut write: left_right::WriteHandle, + settings: SettingsHandle, on_advance: Option>, ) { let mut last_seen_sequence: Option = None; + let mut alert_states: HashMap = HashMap::new(); loop { tokio::select! { @@ -143,12 +160,43 @@ async fn run( drop(primary); for (sequence, timestamp, text) in new_lines { + let text_for_op = text.clone(); write.append(ProfileOp(Box::new(move |snap: &mut EngineSnapshot| { - snap.apply_all_metrics(sequence, timestamp, &text); + snap.apply_all_metrics(sequence, timestamp, &text_for_op); }))); + write.publish(); last_seen_sequence = Some(sequence); + + let (rules, ntfy_topic, ntfy_server) = { + let settings = settings.lock().unwrap(); + ( + settings.alert_rules.clone(), + settings.ntfy_topic.clone(), + settings.ntfy_server.clone(), + ) + }; + let fired = write + .enter() + .map(|guard| { + alerts::evaluate_alerts( + &rules, + &mut alert_states, + &guard, + sequence, + timestamp, + &text, + ) + }) + .unwrap_or_default(); + for (rule, message) in fired { + tokio::spawn(alerts::deliver_for_rule( + rule.name, + message, + ntfy_topic.clone(), + ntfy_server.clone(), + )); + } } - write.publish(); if let (Some(sequence), Some(tx)) = (last_seen_sequence, on_advance.as_ref()) { let _ = tx.send(sequence); } diff --git a/tickets/105-move-alert-evaluation-to-profile-actor.md b/tickets/105-move-alert-evaluation-to-profile-actor.md deleted file mode 100644 index 99451ab..0000000 --- a/tickets/105-move-alert-evaluation-to-profile-actor.md +++ /dev/null @@ -1,52 +0,0 @@ -# 105 — Move alert evaluation onto the profile actor - -## Summary - -Alert rules are profile-scoped and read metric buckets -(`alert_states: HashMap` lives on the -single actor's stack frame today), so per [[0001-actor-split]]'s -resolved decision, alert evaluation moves with metric computation. -Depends on [[104-move-metric-computation-to-profile-actor]] landing -first — alert evaluation reads the metric buckets that ticket moves -onto `ProfileSnapshot`. - -## Where - -- `src/engine/handle.rs:141` — `let mut alert_states: - HashMap = HashMap::new();`, currently a - local in the single actor's `run()` loop. -- `src/engine/handle.rs`'s `Ingest` arm — today, right after - `write.mutate(...)` publishes, the actor does `write.peek()` and - calls `alerts::evaluate_alerts(&rules, &mut alert_states, &guard, - sequence, timestamp, &text)`, then spawns - `alerts::deliver_for_rule` for each fired alert. This whole block - moves to wherever the profile actor's per-line processing lives - (the watch-wake loop from ticket 103/104). -- `src/engine/alerts.rs:139` (`AlertRuntimeState`), `:241` - (`evaluate_alerts`), `:384` (`deliver_for_rule`) — signatures take - the rules list and a `&EngineSnapshot`-shaped guard; check whether - `evaluate_alerts` needs to change its snapshot parameter type to - `ProfileSnapshot` (likely yes, since it reads bucket data that now - lives there) or whether it can stay snapshot-shape-agnostic via - whatever trait/accessor bucket-reading already uses. -- Settings access for alert rules/ntfy config - (`src/engine/handle.rs:170-177`, `settings.lock().unwrap()`) — this - stays available to the profile actor the same way (it's a - `SettingsHandle` clone, not primary-actor state), so no primary - dependency here. - -## Scope - -- Move `alert_states` ownership to the profile actor's task-local - state. -- Move the post-publish `evaluate_alerts` call and `deliver_for_rule` - spawn to fire from the profile actor's per-line processing, reading - `ProfileSnapshot` (post-metric-computation) instead of the primary's - `EngineSnapshot`. -- Update `evaluate_alerts`'s signature in `src/engine/alerts.rs` if it - needs to accept a different snapshot type; keep its actual - evaluation logic unchanged — this ticket is a relocation, not a - rewrite of alert semantics. -- Out of scope: any change to alert rule semantics, ntfy delivery, or - the `AlertRuntimeState` model itself — purely which actor owns and - drives evaluation.