Merge pull request #716 from portableDD/portabledd/alerting-tests-bench

test(devkit/alerting): cooldown, history & integration tests + evaluator bench

Author: Tinna23

packages/devkit/Cargo.toml

@@ -75,3 +75,7 @@ harness = false
 [[bench]]
 name = "streaming_bench"
 harness = false
+
+[[bench]]
+name = "alerting_bench"
+harness = false

packages/devkit/benches/alerting_bench.rs

@@ -0,0 +1,57 @@
+//! Throughput benchmark for the alert evaluator (#643).
+//!
+//! Measures rule evaluations per second with 10 active rules.
+//! Target: > 100,000 evaluations/sec.
+
+use criterion::{criterion_group, criterion_main, Criterion, Throughput};
+
+use stellar_devkit::alerting::evaluator::{FeeSnapshot, RuleEvaluator};
+use stellar_devkit::alerting::rule::{AlertCondition, AlertRule};
+
+fn ten_rules() -> Vec<AlertRule> {
+    let conditions = [
+        AlertCondition::FeeAbove,
+        AlertCondition::FeeBelow,
+        AlertCondition::P95Above,
+        AlertCondition::SpikeCountExceeds,
+        AlertCondition::CapacityUsageAbove,
+    ];
+    (0..10)
+        .map(|i| {
+            AlertRule::new(
+                format!("rule-{i}"),
+                format!("rule {i}"),
+                conditions[i % conditions.len()],
+                50,
+            )
+            // cooldown 0 so every evaluation exercises the full trigger path.
+            .with_cooldown(0)
+        })
+        .collect()
+}
+
+fn bench_evaluator(c: &mut Criterion) {
+    let rules = ten_rules();
+    let snapshot = FeeSnapshot {
+        base_fee: 500,
+        p95: 500,
+        spike_count: 500,
+        capacity_usage_pct: 99,
+    };
+
+    let mut group = c.benchmark_group("alerting_evaluator");
+    group.throughput(Throughput::Elements(rules.len() as u64));
+
+    group.bench_function("evaluate_10_rules", |b| {
+        let evaluator = RuleEvaluator::new();
+        b.iter(|| {
+            let events = evaluator.evaluate(&rules, &snapshot);
+            criterion::black_box(events)
+        });
+    });
+
+    group.finish();
+}
+
+criterion_group!(benches, bench_evaluator);
+criterion_main!(benches);

packages/devkit/tests/alerting_cooldown.rs

@@ -0,0 +1,60 @@
+//! Unit tests for the alert cooldown tracker (#640).
+
+use std::time::{Duration, Instant};
+
+use stellar_devkit::alerting::evaluator::CooldownTracker;
+
+#[test]
+fn second_alert_within_window_is_suppressed() {
+    let ct = CooldownTracker::new();
+    let t0 = Instant::now();
+
+    assert!(ct.should_fire_at("r", 300, t0), "first fire allowed");
+    ct.record_at("r", t0);
+    assert!(
+        !ct.should_fire_at("r", 300, t0 + Duration::from_secs(120)),
+        "within the cooldown window must be suppressed"
+    );
+}
+
+#[test]
+fn alert_fires_after_cooldown_expires() {
+    let ct = CooldownTracker::new();
+    let t0 = Instant::now();
+    ct.record_at("r", t0);
+
+    assert!(!ct.should_fire_at("r", 300, t0 + Duration::from_secs(299)));
+    // At exactly the boundary (elapsed == cooldown) the rule may fire again.
+    assert!(ct.should_fire_at("r", 300, t0 + Duration::from_secs(300)));
+    assert!(ct.should_fire_at("r", 300, t0 + Duration::from_secs(301)));
+}
+
+#[test]
+fn reset_cooldown_allows_immediate_fire() {
+    let ct = CooldownTracker::new();
+    let t0 = Instant::now();
+    ct.record_at("r", t0);
+    assert!(!ct.should_fire_at("r", 300, t0));
+
+    ct.reset_cooldown("r");
+    assert!(ct.should_fire_at("r", 300, t0), "reset clears the cooldown");
+}
+
+#[test]
+fn unknown_rule_is_always_allowed() {
+    let ct = CooldownTracker::new();
+    assert!(ct.should_fire("never-recorded", 300));
+}
+
+#[test]
+fn cooldowns_are_tracked_per_rule() {
+    let ct = CooldownTracker::new();
+    let t0 = Instant::now();
+    ct.record_at("a", t0);
+
+    assert!(!ct.should_fire_at("a", 300, t0), "a is on cooldown");
+    assert!(
+        ct.should_fire_at("b", 300, t0),
+        "b has an independent cooldown"
+    );
+}

packages/devkit/tests/alerting_history.rs

@@ -0,0 +1,96 @@
+//! Unit tests for the alert history store (#641).
+
+use chrono::{TimeZone, Utc};
+
+use stellar_devkit::alerting::history::{AlertHistory, HistoryQuery};
+use stellar_devkit::alerting::rule::{AlertEvent, AlertSeverity};
+
+fn event(rule_id: &str, severity: AlertSeverity, ts_secs: i64, value: u64) -> AlertEvent {
+    AlertEvent {
+        rule_id: rule_id.to_string(),
+        rule_name: rule_id.to_string(),
+        severity,
+        triggered_at: Utc.timestamp_opt(ts_secs, 0).unwrap(),
+        current_value: value,
+        threshold: 0,
+        message: String::new(),
+    }
+}
+
+#[test]
+fn ring_buffer_caps_at_1000_and_drops_oldest() {
+    let mut history = AlertHistory::new();
+    for i in 0..1_500i64 {
+        history.record(event(&format!("r{i}"), AlertSeverity::Info, i, i as u64));
+    }
+    assert_eq!(history.len(), 1_000, "capped at 1,000");
+
+    let all = history.all();
+    // The first 500 (r0..r499) must have been evicted, oldest-first.
+    assert_eq!(all.first().unwrap().rule_id, "r500");
+    assert_eq!(all.last().unwrap().rule_id, "r1499");
+}
+
+#[test]
+fn query_filters_by_rule_id() {
+    let mut history = AlertHistory::new();
+    history.record(event("a", AlertSeverity::Info, 1, 1));
+    history.record(event("b", AlertSeverity::Info, 2, 2));
+    history.record(event("a", AlertSeverity::Warning, 3, 3));
+
+    let a = history.by_rule("a");
+    assert_eq!(a.len(), 2);
+    assert!(a.iter().all(|e| e.rule_id == "a"));
+}
+
+#[test]
+fn query_filters_by_severity() {
+    let mut history = AlertHistory::new();
+    history.record(event("a", AlertSeverity::Info, 1, 1));
+    history.record(event("b", AlertSeverity::Critical, 2, 2));
+    history.record(event("c", AlertSeverity::Critical, 3, 3));
+
+    let crit = history.by_severity(AlertSeverity::Critical);
+    assert_eq!(crit.len(), 2);
+}
+
+#[test]
+fn query_filters_by_time_range() {
+    let mut history = AlertHistory::new();
+    for i in 0..10i64 {
+        history.record(event(&format!("r{i}"), AlertSeverity::Info, 100 + i, 0));
+    }
+    let result = history.query(&HistoryQuery {
+        from: Some(Utc.timestamp_opt(103, 0).unwrap()),
+        to: Some(Utc.timestamp_opt(106, 0).unwrap()),
+        ..Default::default()
+    });
+    // inclusive 103..=106 → 4 events
+    assert_eq!(result.len(), 4);
+}
+
+#[test]
+fn combined_filters_are_anded() {
+    let mut history = AlertHistory::new();
+    history.record(event("a", AlertSeverity::Critical, 10, 0));
+    history.record(event("a", AlertSeverity::Info, 11, 0));
+    history.record(event("b", AlertSeverity::Critical, 12, 0));
+
+    let result = history.query(&HistoryQuery {
+        rule_id: Some("a".to_string()),
+        severity: Some(AlertSeverity::Critical),
+        ..Default::default()
+    });
+    assert_eq!(result.len(), 1);
+    assert_eq!(result[0].triggered_at.timestamp(), 10);
+}
+
+#[test]
+fn export_json_round_trips() {
+    let mut history = AlertHistory::new();
+    history.record(event("a", AlertSeverity::Warning, 1, 42));
+    let json = history.export_json().unwrap();
+    let parsed: Vec<AlertEvent> = serde_json::from_str(&json).unwrap();
+    assert_eq!(parsed.len(), 1);
+    assert_eq!(parsed[0].current_value, 42);
+}

packages/devkit/tests/alerting_integration.rs

@@ -0,0 +1,71 @@
+//! Integration test for the full alert pipeline (#642).
+//!
+//! simulation source → spike transformer → alerting engine.
+//! Asserts an alert is fired and recorded in history.
+
+use stellar_devkit::alerting::{
+    AlertCondition, AlertEngine, AlertRegistry, AlertRule, FeeSnapshot, StdoutDispatcher,
+};
+use stellar_devkit::streaming::transformer::FeeEvent as StreamFeeEvent;
+use stellar_devkit::streaming::{FeeRecord, SpikeDetectionTransformer, SpikeTransformerEvent};
+
+/// Deterministic simulation source: every 5th record is a large fee (spike).
+fn simulation_source(n: u64) -> Vec<FeeRecord> {
+    (0..n)
+        .map(|i| FeeRecord {
+            fee_amount: if i % 5 == 0 { 900 } else { 200 },
+            ledger_sequence: i,
+            timestamp_ms: 1_700_000_000_000 + i as i64,
+            transaction_hash: None,
+            is_spike: false,
+            created_at: "2026-01-01T00:00:00Z".to_string(),
+        })
+        .collect()
+}
+
+#[tokio::test]
+async fn full_pipeline_fires_and_records_alert() {
+    // 1. Source
+    let source = simulation_source(20);
+
+    // 2. Spike transformer (baseline 200, 2× threshold → >400 is a spike)
+    let transformer = SpikeDetectionTransformer::new(200, 2.0);
+    let spike_count = source
+        .iter()
+        .filter(|record| {
+            matches!(
+                transformer.transform(StreamFeeEvent::NewFeeRecord((*record).clone())),
+                Some(SpikeTransformerEvent::SpikeDetected(_))
+            )
+        })
+        .count() as u64;
+    assert!(spike_count > 0, "the source must produce spikes");
+
+    // 3. Alerting engine with a SpikeCountExceeds rule
+    let mut registry = AlertRegistry::new();
+    registry.add(
+        AlertRule::new("spike", "spike alert", AlertCondition::SpikeCountExceeds, 2)
+            .with_cooldown(0),
+    );
+    let mut engine = AlertEngine::with_registry(registry);
+    engine.add_dispatcher(Box::new(StdoutDispatcher::new()));
+
+    let snapshot = FeeSnapshot {
+        base_fee: 900,
+        spike_count,
+        ..Default::default()
+    };
+
+    let fired = engine.process(&snapshot).await;
+
+    // 4. Assertions
+    assert!(!fired.is_empty(), "an alert should have fired");
+    assert_eq!(fired[0].rule_id, "spike");
+    assert_eq!(fired[0].current_value, spike_count);
+    assert_eq!(
+        engine.history().len(),
+        fired.len(),
+        "fired events are recorded in history"
+    );
+    assert_eq!(engine.history().all()[0].rule_id, "spike");
+}