test: close tracked gaps (detect() e2e tests, EWMA persistence, mocked store tests)

services/anomaly-rust/src/detector/mod.rs

@@ -49,14 +49,18 @@ pub struct ErrorRateBaseline {
 }
 
 impl ErrorRateBaseline {
+    /// Returns `(baseline_before_update, z_score, severity)` on anomaly,
+    /// or `None` if the current reading isn't anomalous. The baseline is
+    /// the EWMA value *before* this reading was folded in, so callers can
+    /// persist "what we expected" alongside "what we saw".
     pub fn update_and_check(
         &mut self,
         service: &str,
         current_error_rate: f64,
         alpha: f64,
         warning: f64,
         critical: f64,
-    ) -> Option<(f64, &'static str)> {
+    ) -> Option<(f64, f64, &'static str)> {
         let baseline = *self.ewma.get(service).unwrap_or(&current_error_rate);
         let deviation = current_error_rate - baseline;
 
@@ -71,7 +75,7 @@ impl ErrorRateBaseline {
         let scale = baseline.max(0.01); // avoid division blowing up near zero
         let z = deviation / scale;
 
-        severity_for(z, warning, critical).map(|sev| (z, sev))
+        severity_for(z, warning, critical).map(|sev| (baseline, z, sev))
     }
 }
 
@@ -161,7 +165,7 @@ pub fn detect(
         // suspicious drops toward zero are worth flagging) ---
         if current.event_count > 0 {
             let current_error_rate = current.error_count as f64 / current.event_count as f64;
-            if let Some((z, sev)) = error_baseline.update_and_check(
+            if let Some((baseline, z, sev)) = error_baseline.update_and_check(
                 service,
                 current_error_rate,
                 cfg.ewma_alpha,
@@ -173,7 +177,9 @@ pub fn detect(
                     service: service.to_string(),
                     anomaly_type: "error_burst".to_string(),
                     metric_value: current_error_rate,
-                    baseline_mean: 0.0,
+                    baseline_mean: baseline,
+                    // EWMA doesn't track variance, so there's no meaningful
+                    // stddev to report here — 0.0 is accurate, not a placeholder.
                     baseline_stddev: 0.0,
                     z_score: z,
                     severity: sev.to_string(),
@@ -238,5 +244,152 @@ mod tests {
         // A sudden spike should be flagged against the now-stable low baseline.
         let result = baseline.update_and_check("checkout", 0.6, 0.3, 2.5, 4.0);
         assert!(result.is_some(), "expected spike to be flagged");
+        let (baseline_value, _z, _sev) = result.unwrap();
+        assert!(
+            (baseline_value - 0.02).abs() < 0.05,
+            "expected baseline near the primed 0.02 rate, got {baseline_value}"
+        );
+    }
+
+    fn make_bucket(
+        service: &str,
+        minute_offset: i64,
+        p95: f64,
+        max: f64,
+        errors: i64,
+        total: i64,
+    ) -> MinuteBucket {
+        MinuteBucket {
+            bucket: chrono::Utc::now() + chrono::Duration::minutes(minute_offset),
+            service: service.to_string(),
+            event_count: total,
+            error_count: errors,
+            avg_latency_ms: Some(p95 * 0.7),
+            p95_latency_ms: Some(p95),
+            max_latency_ms: Some(max),
+        }
+    }
+
+    fn default_cfg() -> DetectionConfig {
+        DetectionConfig {
+            z_score_warning: 2.5,
+            z_score_critical: 4.0,
+            ewma_alpha: 0.3,
+        }
+    }
+
+    #[test]
+    fn detect_flags_nothing_for_stable_service() {
+        let buckets = vec![
+            make_bucket("checkout", -5, 100.0, 150.0, 1, 100),
+            make_bucket("checkout", -4, 105.0, 155.0, 1, 100),
+            make_bucket("checkout", -3, 98.0, 148.0, 1, 100),
+            make_bucket("checkout", -2, 102.0, 152.0, 1, 100),
+            make_bucket("checkout", -1, 101.0, 151.0, 1, 100),
+        ];
+        let mut baseline = ErrorRateBaseline::default();
+        let anomalies = detect(&buckets, &mut baseline, &default_cfg());
+        assert!(
+            anomalies.is_empty(),
+            "expected no anomalies for a stable service, got: {anomalies:?}"
+        );
+    }
+
+    #[test]
+    fn detect_flags_latency_spike_but_not_a_speedup() {
+        // Five stable baseline minutes, then one minute that spikes p95 way up.
+        let mut buckets = vec![
+            make_bucket("checkout", -6, 100.0, 150.0, 1, 100),
+            make_bucket("checkout", -5, 102.0, 152.0, 1, 100),
+            make_bucket("checkout", -4, 98.0, 148.0, 1, 100),
+            make_bucket("checkout", -3, 101.0, 151.0, 1, 100),
+            make_bucket("checkout", -2, 99.0, 149.0, 1, 100),
+        ];
+        buckets.push(make_bucket("checkout", -1, 5000.0, 6000.0, 1, 100));
+
+        let mut baseline = ErrorRateBaseline::default();
+        let anomalies = detect(&buckets, &mut baseline, &default_cfg());
+
+        let latency_anomalies: Vec<_> = anomalies
+            .iter()
+            .filter(|a| {
+                a.anomaly_type == "latency_spike" || a.anomaly_type == "max_latency_outlier"
+            })
+            .collect();
+        assert!(
+            !latency_anomalies.is_empty(),
+            "expected a latency spike to be detected"
+        );
+        for a in &latency_anomalies {
+            assert!(
+                a.z_score > 0.0,
+                "latency anomaly must have a positive z_score, got {} for {}",
+                a.z_score,
+                a.anomaly_type
+            );
+        }
+
+        // Regression guard for the original bug: a service getting FASTER
+        // than baseline must never be flagged.
+        let mut faster_buckets = vec![
+            make_bucket("payments", -6, 500.0, 600.0, 1, 100),
+            make_bucket("payments", -5, 510.0, 610.0, 1, 100),
+            make_bucket("payments", -4, 490.0, 590.0, 1, 100),
+            make_bucket("payments", -3, 505.0, 605.0, 1, 100),
+            make_bucket("payments", -2, 495.0, 595.0, 1, 100),
+        ];
+        faster_buckets.push(make_bucket("payments", -1, 10.0, 15.0, 1, 100));
+        let mut baseline2 = ErrorRateBaseline::default();
+        let anomalies2 = detect(&faster_buckets, &mut baseline2, &default_cfg());
+        let false_positives: Vec<_> = anomalies2
+            .iter()
+            .filter(|a| {
+                a.anomaly_type == "latency_spike" || a.anomaly_type == "max_latency_outlier"
+            })
+            .collect();
+        assert!(
+            false_positives.is_empty(),
+            "a service running FASTER than baseline must not be flagged, got: {false_positives:?}"
+        );
+    }
+
+    #[test]
+    fn detect_flags_error_burst() {
+        let mut buckets = vec![];
+        for i in (-6..-1).rev() {
+            buckets.push(make_bucket("inventory", i, 100.0, 150.0, 2, 100));
+        }
+        buckets.push(make_bucket("inventory", -1, 100.0, 150.0, 60, 100));
+
+        let mut baseline = ErrorRateBaseline::default();
+        // Prime the EWMA baseline the same way main.rs would across prior cycles.
+        for i in 0..5 {
+            baseline.update_and_check("inventory", 0.02, 0.3, 2.5, 4.0);
+            let _ = i;
+        }
+
+        let anomalies = detect(&buckets, &mut baseline, &default_cfg());
+        let error_anomalies: Vec<_> = anomalies
+            .iter()
+            .filter(|a| a.anomaly_type == "error_burst")
+            .collect();
+        assert!(
+            !error_anomalies.is_empty(),
+            "expected an error burst to be detected, got: {anomalies:?}"
+        );
+    }
+
+    #[test]
+    fn detect_skips_services_with_insufficient_history() {
+        let buckets = vec![
+            make_bucket("new-service", -1, 100.0, 150.0, 1, 100),
+            make_bucket("new-service", 0, 5000.0, 6000.0, 50, 100),
+        ];
+        let mut baseline = ErrorRateBaseline::default();
+        let anomalies = detect(&buckets, &mut baseline, &default_cfg());
+        assert!(
+            anomalies.is_empty(),
+            "expected no anomalies for a service with fewer than 3 buckets of history"
+        );
     }
 }

services/ingestion-go/go.mod

@@ -17,6 +17,7 @@ require (
 	github.com/jackc/puddle/v2 v2.2.2 // indirect
 	github.com/klauspost/compress v1.17.9 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
+	github.com/pashagolub/pgxmock/v4 v4.3.0 // indirect
 	github.com/pierrec/lz4/v4 v4.1.15 // indirect
 	github.com/prometheus/client_model v0.6.1 // indirect
 	github.com/prometheus/common v0.55.0 // indirect

services/ingestion-go/go.sum

@@ -24,6 +24,8 @@ github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0
 github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
+github.com/pashagolub/pgxmock/v4 v4.3.0 h1:DqT7fk0OCK6H0GvqtcMsLpv8cIwWqdxWgfZNLeHCb/s=
+github.com/pashagolub/pgxmock/v4 v4.3.0/go.mod h1:9VoVHXwS3XR/yPtKGzwQvwZX1kzGB9sM8SviDcHDa3A=
 github.com/pierrec/lz4/v4 v4.1.15 h1:MO0/ucJhngq7299dKLwIMtgTfbkoSPF6AoMYDd8Q4q0=
 github.com/pierrec/lz4/v4 v4.1.15/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=

services/ingestion-go/internal/migrate/migrate.go

@@ -15,6 +15,10 @@ var initSchema string
 // It is idempotent: every statement uses IF NOT EXISTS / ON CONFLICT-style
 // guards, so it's safe to run on every service startup, whether the schema
 // already exists or the volume is brand new.
+//
+// Takes a concrete *pgxpool.Pool (rather than store.PgxPool) specifically to
+// avoid an import cycle: store's integration tests import migrate to apply
+// the schema, so migrate cannot import store back.
 func Apply(ctx context.Context, pool *pgxpool.Pool, logger *slog.Logger) error {
 	logger.Info("applying database schema migration")
 	if _, err := pool.Exec(ctx, initSchema); err != nil {

services/ingestion-go/internal/store/store.go

@@ -7,13 +7,25 @@ import (
 	"time"
 
 	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
 	"github.com/jackc/pgx/v5/pgxpool"
 	"github.com/yahiaongh/sentinelops/services/ingestion-go/internal/consumer"
 	"github.com/yahiaongh/sentinelops/services/ingestion-go/internal/metrics"
 )
 
+// PgxPool is the subset of *pgxpool.Pool's behavior that Store depends on.
+// Defined as an interface (rather than using *pgxpool.Pool directly) so
+// tests can substitute a mock pool without needing a live database.
+type PgxPool interface {
+	Ping(ctx context.Context) error
+	Begin(ctx context.Context) (pgx.Tx, error)
+	Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)
+	QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
+	Close()
+}
+
 type Store struct {
-	pool   *pgxpool.Pool
+	pool   PgxPool
 	logger *slog.Logger
 }
 
@@ -33,9 +45,12 @@ func (s *Store) Close() {
 }
 
 // Pool exposes the underlying connection pool for use by other internal
-// packages (e.g. migrate) that need direct database access.
+// packages (e.g. migrate) that need direct database access. Returns the
+// concrete *pgxpool.Pool (not the PgxPool interface) since callers like
+// migrate.Apply need the real type, and s.pool is always a genuine
+// *pgxpool.Pool outside of unit tests.
 func (s *Store) Pool() *pgxpool.Pool {
-	return s.pool
+	return s.pool.(*pgxpool.Pool)
 }
 
 const insertLogEventQuery = `

services/ingestion-go/internal/store/store_test.go

@@ -0,0 +1,62 @@
+package store
+
+import (
+	"context"
+	"log/slog"
+	"os"
+	"testing"
+
+	"github.com/pashagolub/pgxmock/v4"
+	"github.com/yahiaongh/sentinelops/services/ingestion-go/internal/consumer"
+)
+
+func testLogger() *slog.Logger {
+	return slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError}))
+}
+
+func TestFlushBatch_EmptySlice_NoOp(t *testing.T) {
+	mock, err := pgxmock.NewPool()
+	if err != nil {
+		t.Fatalf("failed to create mock pool: %v", err)
+	}
+	defer mock.Close()
+
+	s := &Store{pool: mock, logger: testLogger()}
+
+	// No expectations set on the mock — if FlushBatch tries to touch the
+	// database for an empty slice, ExpectationsWereMet will fail below,
+	// proving the early-return short-circuit actually works.
+	if err := s.FlushBatch(context.Background(), nil); err != nil {
+		t.Errorf("expected nil error for empty slice, got: %v", err)
+	}
+
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Errorf("expected zero database interaction for an empty batch, got: %v", err)
+	}
+}
+
+func TestFlushBatch_AllInvalidTimestamps_NoOp(t *testing.T) {
+	mock, err := pgxmock.NewPool()
+	if err != nil {
+		t.Fatalf("failed to create mock pool: %v", err)
+	}
+	defer mock.Close()
+
+	s := &Store{pool: mock, logger: testLogger()}
+
+	events := []consumer.LogEvent{
+		{EventID: "a", Timestamp: "not-a-timestamp"},
+		{EventID: "b", Timestamp: ""},
+	}
+
+	// Every event has an unparsable timestamp, so FlushBatch should skip
+	// all of them and never open a transaction — again proven by the mock
+	// having zero expectations set.
+	if err := s.FlushBatch(context.Background(), events); err != nil {
+		t.Errorf("expected nil error when all events have invalid timestamps, got: %v", err)
+	}
+
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Errorf("expected zero database interaction when no events are valid, got: %v", err)
+	}
+}