Merge pull request #18 from AstroHQ/jfro/fix-avg-metrics-test

fix: average_metrics_from_signpost using wrong start signpost, causing long durations

Author: jfro

Cargo.toml

@@ -26,6 +26,9 @@ tokio = { version = "1.20", features = ["sync"] }
 [dev-dependencies]
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 anyhow = "1.0"
+clap = { version = "4", features = ["derive"] }
+chrono = "0.4"
+tempfile = "3"
 
 [features]
 default = []

examples/summary.rs

@@ -1,30 +1,72 @@
-use metrics_sqlite::{MetricsDb, Session};
+use std::time::Duration;
+
+use chrono::{DateTime, Local, TimeZone};
+use clap::Parser;
+use metrics_sqlite::MetricsDb;
 use tracing_subscriber::{fmt, layer::SubscriberExt, util::SubscriberInitExt, EnvFilter};
 
+#[derive(Parser)]
+struct Args {
+    /// Path to the metrics database file
+    db_path: String,
+}
+
 fn main() -> anyhow::Result<()> {
+    let args = Args::parse();
     let fmt_layer = fmt::layer();
     let filter_layer = EnvFilter::try_from_default_env().or_else(|_| EnvFilter::try_new("info"))?;
     tracing_subscriber::registry()
         .with(filter_layer)
         .with(fmt_layer)
         .init();
-    let mut db = MetricsDb::new("metrics.db")?;
+
+    let mut db = MetricsDb::new(&args.db_path)?;
     let session = db.session_from_signpost("liquid.connected")?;
-    let rtt = average(&mut db, "net_quality.rtt", &session)?;
-    let throughput = average(&mut db, "net_quality.throughput", &session)? / 1024.0; // KB/s
-    let metered_throughput =
-        average(&mut db, "rate_control.metered_throughput", &session)? / 1024.0; // KB/s
-    println!("Last connection: {session:#?}");
-    println!("RTT: Average {rtt:.2}ms");
-    println!("Throughput: Average {throughput:.2}KB/s");
-    println!("Metered: Average {metered_throughput:.2}KB/s");
+    let averages = db.average_metrics_from_signpost(
+        "liquid.connected",
+        &[
+            "net_quality.rtt",
+            "net_quality.throughput",
+            // "rate_control.metered_throughput",
+        ],
+    )?;
+
+    println!("Last connection:");
+    println!("  Start:    {}", format_timestamp(session.start_time));
+    println!("  End:      {}", format_timestamp(session.end_time));
+    println!("  Duration: {}", format_duration(session.duration));
+
+    if let Some(&rtt) = averages.get("net_quality.rtt") {
+        println!("RTT: Average {rtt:.2}ms");
+    }
+    if let Some(&throughput) = averages.get("net_quality.throughput") {
+        println!("Throughput: Average {:.2}KB/s", throughput / 1024.0);
+    }
+    if let Some(&metered) = averages.get("rate_control.metered_throughput") {
+        println!("Metered: Average {:.2}KB/s", metered / 1024.0);
+    }
 
     Ok(())
 }
-fn average(db: &mut MetricsDb, key: &str, session: &Session) -> anyhow::Result<f64> {
-    let metrics = db.metrics_for_key(key, Some(session))?;
-    let sum: f64 = metrics.iter().map(|m| m.value).sum();
-    let samples = metrics.len();
-    let average = sum / samples as f64;
-    Ok(average)
+
+fn format_timestamp(secs: f64) -> String {
+    let dt: DateTime<Local> = Local
+        .timestamp_opt(secs as i64, ((secs.fract()) * 1_000_000_000.0) as u32)
+        .unwrap();
+    dt.format("%Y-%m-%d %H:%M:%S").to_string()
+}
+
+fn format_duration(d: Duration) -> String {
+    let total_secs = d.as_secs();
+    let hours = total_secs / 3600;
+    let minutes = (total_secs % 3600) / 60;
+    let seconds = total_secs % 60;
+    let millis = d.subsec_millis();
+    if hours > 0 {
+        format!("{hours}h {minutes:02}m {seconds:02}.{millis:03}s")
+    } else if minutes > 0 {
+        format!("{minutes}m {seconds:02}.{millis:03}s")
+    } else {
+        format!("{seconds}.{millis:03}s")
+    }
 }

src/metrics_db.rs

@@ -61,7 +61,7 @@ impl MetricsDb {
         self.sessions.clone()
     }
 
-    /// Returns a session (timestamp range) from the first occurrence of the signpost to the latest metric
+    /// Returns a session (timestamp range) from the last occurrence of the signpost to the latest metric
     pub fn session_from_signpost(&mut self, metric: &str) -> Result<Session> {
         query::session_from_signpost(&mut self.db, metric)
     }
@@ -102,6 +102,21 @@ impl MetricsDb {
         Ok(r)
     }
 
+    /// Returns average values for the given metric keys within the session defined by a signpost.
+    ///
+    /// Also includes `session.duration` in the results.
+    pub fn average_metrics_from_signpost(
+        &mut self,
+        signpost: &str,
+        keys: &[&str],
+    ) -> Result<std::collections::HashMap<String, f64>> {
+        query::metrics_summary_for_signpost_and_keys(
+            &mut self.db,
+            signpost,
+            keys.iter().map(|s| s.to_string()).collect(),
+        )
+    }
+
     /// Returns all metrics for given key in ascending timestamp order
     pub fn metrics_for_key(
         &mut self,

src/metrics_db/query.rs

@@ -17,11 +17,13 @@ pub(crate) fn metric_key_for_name<'a>(
 pub(crate) fn session_from_signpost(db: &mut SqliteConnection, metric: &str) -> Result<Session> {
     use crate::schema::metrics::dsl::*;
     let metric_key = metric_key_for_name(db, metric)?;
-    let query = metrics
-        .order(timestamp.asc())
+    // Use the most recent signpost occurrence as the session start
+    let start_query = metrics
+        .order(timestamp.desc())
         .filter(metric_key_id.eq(metric_key.id))
         .limit(1);
-    let start = query.first::<Metric>(db)?;
+    let start = start_query.first::<Metric>(db)?;
+    // End is the latest metric in the DB (approximates NOW)
     let end_query = metrics.order(timestamp.desc()).limit(1);
     let end = end_query.first::<Metric>(db)?;
     if end.timestamp <= start.timestamp {
@@ -87,3 +89,106 @@ pub(crate) fn metrics_summary_for_signpost_and_keys(
     results.insert("session.duration".to_string(), duration_secs);
     Ok(results)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::models::{MetricKey, NewMetric};
+
+    /// Create a temp DB and return the connection + temp dir (keep alive for DB lifetime)
+    fn setup_test_db() -> (SqliteConnection, tempfile::TempDir) {
+        let dir = tempfile::tempdir().unwrap();
+        let db_path = dir.path().join("test.db");
+        let db = crate::setup_db(&db_path).unwrap();
+        (db, dir)
+    }
+
+    /// Insert a metric key, returning its id
+    fn insert_key(db: &mut SqliteConnection, name: &str) -> i64 {
+        let key = MetricKey::key_by_name(name, db).unwrap();
+        key.id
+    }
+
+    /// Insert a metric row
+    fn insert_metric(db: &mut SqliteConnection, key_id: i64, ts: f64, val: f64) {
+        use crate::schema::metrics::dsl::*;
+        diesel::insert_into(metrics)
+            .values(&NewMetric {
+                timestamp: ts,
+                metric_key_id: key_id,
+                value: val,
+            })
+            .execute(db)
+            .unwrap();
+    }
+
+    /// Sets up a DB simulating stale signpost data from a previous session:
+    /// - "app.started" signpost at t=100.0 (old/stale session)
+    /// - "app.started" signpost at t=500.0 (current session)
+    /// - "cpu" metrics at t=500, 510, 520 with values 50, 60, 70 (current session)
+    /// - "cpu" metrics at t=100, 110 with values 10, 20 (old session)
+    /// - latest metric in DB is at t=520
+    fn populate_test_db(db: &mut SqliteConnection) {
+        let signpost_id = insert_key(db, "app.started");
+        let cpu_id = insert_key(db, "cpu");
+
+        // Old session data (stale)
+        insert_metric(db, signpost_id, 100.0, 1.0);
+        insert_metric(db, cpu_id, 100.0, 10.0);
+        insert_metric(db, cpu_id, 110.0, 20.0);
+
+        // Current session data
+        insert_metric(db, signpost_id, 500.0, 1.0);
+        insert_metric(db, cpu_id, 500.0, 50.0);
+        insert_metric(db, cpu_id, 510.0, 60.0);
+        insert_metric(db, cpu_id, 520.0, 70.0);
+    }
+
+    #[test]
+    fn test_session_from_signpost_uses_latest_signpost() {
+        // With stale signpost data at t=100 and current at t=500,
+        // start should be 500 (latest signpost), not 100 (oldest).
+        let (mut db, _dir) = setup_test_db();
+        populate_test_db(&mut db);
+
+        let session = session_from_signpost(&mut db, "app.started").unwrap();
+        assert_eq!(
+            session.start_time, 500.0,
+            "start should be the most recent signpost (t=500), not the stale one (t=100)"
+        );
+        assert_eq!(session.end_time, 520.0);
+    }
+
+    #[test]
+    fn test_summary_duration_not_inflated_by_stale_signpost() {
+        // Duration should be based on the latest signpost, not a stale one from a previous session.
+        let (mut db, _dir) = setup_test_db();
+        populate_test_db(&mut db);
+
+        let results =
+            metrics_summary_for_signpost_and_keys(&mut db, "app.started", vec!["cpu".to_string()])
+                .unwrap();
+
+        // cpu average across t=500..520 (current session only): (50+60+70)/3 = 60
+        assert!(
+            (results["cpu"] - 60.0).abs() < f64::EPSILON,
+            "cpu average should only include current session metrics"
+        );
+        // session.duration should be 520 - 500 = 20, not 520 - 100 = 420
+        assert_eq!(
+            results["session.duration"], 20.0,
+            "duration should be 20s (t=500..520), not inflated to 420s by stale signpost"
+        );
+    }
+
+    #[test]
+    fn test_average_for_session_with_explicit_range() {
+        let (mut db, _dir) = setup_test_db();
+        populate_test_db(&mut db);
+
+        // With a manually bounded session t=500..510, avg = (50+60)/2 = 55
+        let session = Session::new(500.0, 510.0);
+        let avg = average_for_session(&mut db, "cpu", &session).unwrap();
+        assert!((avg - 55.0).abs() < f64::EPSILON);
+    }
+}