etdl_core/

monitor.rs

use crate::chaos::ChaosController;
use crate::observation::{
    generate_observation_id, now_rfc3339, NoopSink, ReliabilityObservation, SharedSink,
};
use crate::sla::SlaTracker;
use std::sync::Arc;
use std::sync::Mutex;

pub struct BranchMonitor {
    node_id: String,
    sla_tracker: Arc<Mutex<SlaTracker>>,
    chaos: Arc<Mutex<ChaosController>>,
    observation_sink: SharedSink,
    service: Option<String>,
    service_version: Option<String>,
    deployment: Option<String>,
    build_ref: Option<String>,
}

impl BranchMonitor {
    pub fn new(node_id: &str) -> Self {
        BranchMonitor {
            node_id: node_id.to_string(),
            sla_tracker: Arc::new(Mutex::new(SlaTracker::new())),
            chaos: Arc::new(Mutex::new(ChaosController::new())),
            observation_sink: Arc::new(NoopSink),
            service: None,
            service_version: None,
            deployment: None,
            build_ref: None,
        }
    }

    /// Attach an observation sink (JSON Lines, OTel, database adapter, ...).
    pub fn with_sink(mut self, sink: SharedSink) -> Self {
        self.observation_sink = sink;
        self
    }

    /// Identify the service/component this monitor runs in, for the
    /// observations it emits.
    pub fn with_service(mut self, service: impl Into<String>) -> Self {
        self.service = Some(service.into());
        self
    }

    /// Identify the software version generating observations, distinct from
    /// the deployment slot. Lets an analyst compare "predictions from build
    /// v1" against "observations generated by v1".
    pub fn with_service_version(mut self, version: impl Into<String>) -> Self {
        self.service_version = Some(version.into());
        self
    }

    /// Identify the deployment/environment (e.g. `prod-us-east-1`).
    pub fn with_deployment(mut self, deployment: impl Into<String>) -> Self {
        self.deployment = Some(deployment.into());
        self
    }

    /// A stable reference to the compiled reliability artifact/build that
    /// produced this service, so observations can be traced back to the
    /// model that predicted them without embedding the whole artifact.
    pub fn with_build_ref(mut self, build_ref: impl Into<String>) -> Self {
        self.build_ref = Some(build_ref.into());
        self
    }

    /// Build a runtime observation stamped with this monitor's identity.
    fn observation(&self, outcome: &str) -> ReliabilityObservation {
        ReliabilityObservation {
            id: generate_observation_id(),
            event: self.node_id.clone(),
            timestamp: now_rfc3339(),
            service: self.service.clone(),
            operation: None,
            environment: None,
            deployment: self.deployment.clone(),
            service_version: self.service_version.clone(),
            build_ref: self.build_ref.clone(),
            outcome: outcome.to_string(),
            conditions: Vec::new(),
            duration_ms: None,
            trace_id: None,
        }
    }

    /// Record an immutable failure observation for offline analysis.
    /// Lightweight: does not run statistics, query databases, or call AI.
    pub fn record_failure_observation(&self, observation: ReliabilityObservation) {
        self.observation_sink.emit(&observation);
    }

    pub fn with_sla(node_id: &str, sla_tracker: Arc<Mutex<SlaTracker>>) -> Self {
        BranchMonitor {
            node_id: node_id.to_string(),
            sla_tracker,
            chaos: Arc::new(Mutex::new(ChaosController::new())),
            observation_sink: Arc::new(NoopSink),
            service: None,
            service_version: None,
            deployment: None,
            build_ref: None,
        }
    }

    pub fn with_chaos(node_id: &str, chaos: Arc<Mutex<ChaosController>>) -> Self {
        BranchMonitor {
            node_id: node_id.to_string(),
            sla_tracker: Arc::new(Mutex::new(SlaTracker::new())),
            chaos,
            observation_sink: Arc::new(NoopSink),
            service: None,
            service_version: None,
            deployment: None,
            build_ref: None,
        }
    }

    pub fn record_branch(&mut self, outcome: &str, declared_probability: f64) {
        let should_chaos = {
            let mut chaos = self.chaos.lock().unwrap();
            chaos.should_inject_chaos(&self.node_id)
        };
        if should_chaos {
            return;
        }

        let mut sla = self.sla_tracker.lock().unwrap();
        let anomaly = sla.record(&self.node_id, outcome, declared_probability, true);

        if anomaly {
            crate::telemetry::emit_anomaly_event(
                &self.node_id,
                outcome,
                declared_probability,
                sla.observed_frequency(&self.node_id, outcome),
            );
        }
        drop(sla);

        crate::telemetry::attach_node_span_attribute(&self.node_id);

        // Record what happened. The declared probability is the prediction;
        // it lives in the compiled artifact/build manifest, not on every
        // observation, so it is not duplicated here (it would go stale on
        // recalibration). The runtime only records data; interpretation
        // (predicted vs observed) happens offline, in etdl-reliability.
        self.observation_sink.emit(&self.observation(outcome));
    }

    pub fn record_failure(
        &mut self,
        operation_id: &str,
        error: &dyn std::error::Error,
        declared_probability: Option<f64>,
    ) {
        let key = format!("{}.failure", operation_id);
        let outcome = "FAILURE";

        if let Some(prob) = declared_probability {
            let mut sla = self.sla_tracker.lock().unwrap();
            let anomaly = sla.record(&key, outcome, prob, true);

            if anomaly {
                crate::telemetry::emit_anomaly_event(
                    &key,
                    outcome,
                    prob,
                    sla.observed_frequency(&key, outcome),
                );
            }
        }

        crate::telemetry::attach_node_span_attribute(&key);
        eprintln!("[etdl] operation '{}' failed: {}", operation_id, error);

        let mut o = self.observation("failed");
        o.event = key;
        o.operation = Some(operation_id.to_string());
        self.observation_sink.emit(&o);
    }

    /// Record that `operation_id` completed *without* failing. Must be
    /// called on the same `"{operation_id}.failure"` SLA key
    /// [`record_failure`] uses, with `occurred = false`, or that key's
    /// rolling window only ever sees failures (every entry `record_failure`
    /// ever pushes) and its observed frequency is permanently `1.0` —
    /// which made [`record_failure`]'s anomaly check fire unconditionally
    /// for any operation that failed at least a handful of times over its
    /// lifetime (`sla::MIN_OBSERVATIONS`), regardless of its actual overall
    /// failure rate. Generated code calls this from
    /// the operation's `Ok` arm whenever it calls `record_failure` from the
    /// matching `Err` arm (i.e. whenever `onFailureProbabilitySource` is
    /// declared and resolves) — see `codegen/rust.rs`'s `Ok(_result) =>`
    /// arm.
    pub fn record_success(&mut self, operation_id: &str, declared_probability: Option<f64>) {
        let key = format!("{}.failure", operation_id);
        let outcome = "FAILURE";

        if let Some(prob) = declared_probability {
            let mut sla = self.sla_tracker.lock().unwrap();
            let anomaly = sla.record(&key, outcome, prob, false);

            if anomaly {
                crate::telemetry::emit_anomaly_event(
                    &key,
                    outcome,
                    prob,
                    sla.observed_frequency(&key, outcome),
                );
            }
        }
    }

    pub fn flush(&self) {
        let sla = self.sla_tracker.lock().unwrap();
        eprintln!(
            "[etdl] node '{}': {} evaluations recorded",
            self.node_id,
            sla.total_evaluations()
        );
    }
}

impl Drop for BranchMonitor {
    fn drop(&mut self) {
        self.flush();
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_record_branch() {
        let mut monitor = BranchMonitor::new("test_barrier");
        monitor.record_branch("SUCCESS", 0.95);
        monitor.record_branch("SUCCESS", 0.95);
        monitor.record_branch("FAILURE", 0.05);
    }

    #[derive(Debug)]
    struct FakeError;
    impl std::fmt::Display for FakeError {
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
            write!(f, "fake error")
        }
    }
    impl std::error::Error for FakeError {}

    /// Regression test for the false-alarm bug: `record_failure` alone
    /// pushes every call into a `"{op}.failure"`-only SLA window, so its
    /// observed frequency was permanently `1.0` regardless of the
    /// operation's actual overall failure rate — any operation with a
    /// declared failure probability below `1.0 - threshold` would
    /// eventually, and permanently, be flagged anomalous. `record_success`
    /// must be called on the same key with `occurred = false` (as
    /// generated code now does in the `Ok` arm) so the window reflects the
    /// operation's real success/failure mix.
    #[test]
    fn record_success_keeps_observed_frequency_meaningful_not_permanently_one() {
        let mut monitor = BranchMonitor::new("op");
        // 5% declared failure probability, 20 attempts: 1 failure, 19
        // successes — exactly matching the declared rate.
        for _ in 0..19 {
            monitor.record_success("checkout", Some(0.05));
        }
        monitor.record_failure("checkout", &FakeError, Some(0.05));

        let observed = monitor
            .sla_tracker
            .lock()
            .unwrap()
            .observed_frequency("checkout.failure", "FAILURE");

        // Without record_success, this would be 1.0 (every recorded entry
        // is a failure) regardless of how rarely the operation actually
        // fails. With it, the window reflects the true 1-in-20 rate.
        assert!(
            (observed - 0.05).abs() < 1e-9,
            "expected observed frequency ~0.05 (1 failure in 20 attempts), got {observed} \
             (1.0 would mean the false-alarm bug regressed)"
        );
    }
}