fix(metrics): pop_last_seen_seconds uses monotonic clock, not epoch

CHANGELOG.md

@@ -9,6 +9,7 @@ All notable changes to this project are documented here, following
 - Speedtest now runs providers sequentially, measures each download over the full window, and reports the fastest clean provider — fixing wildly variable, under-reporting results on fast links.
 
 ### Fixed
+- `blackwall_flow_pop_last_seen_seconds` reported a nonsense ~epoch-sized value (so any "POP sensor down/stale" alert built on it was dead). Fallout from the monotonic-clock change: the collector stamps `AgentStat.last_seen_ms` from the monotonic clock, but the `/metrics` renderer computed the age against wall-clock epoch — mixing the two domains, the gauge always read ~current-epoch-seconds regardless of how recently the agent was seen. The renderer now uses the collector's monotonic clock (exposed as `blackwall_flow::monotonic_now_ms()`), so a freshly-observed agent reads ~0 s. Found validating the M0 sockpuppet fleet. Detection was never affected (it used the monotonic clock consistently); only the liveness gauge was wrong.
 - Pre-M0 detection-quality batch (makes the shadow observation's FP/FN signal trustworthy before the AS214806 detection-only deploy; from a full-codebase review + re-triage). **Detector correctness:** windowing/eviction/hold-down now use a **monotonic clock** (`Instant` baseline) so an NTP/wall-clock step can't freeze or force-evict the window; a new **minimum-sample gate** (`--min-samples`, default 8) stops sampling-variance false positives (2 packets at 1-in-65536 no longer extrapolate to a 131k-pps "attack"); the sampling-rate sanity clamp is now **direction-aware** — a low reported rate is clamped up (anti-suppression) while a legitimately-high adaptive-sampling rate is *trusted* up to `expected × --max-sampling-factor` (default 64, floored at 4) instead of being clamped down and hiding a real flood; sFlow decoding is **per-sample resilient** so one malformed sample no longer discards a whole datagram's valid observations (envelope-framing errors stay fatal); and mitigation selection treats destination **port 0** (fragmented / non-TCP-UDP flows) as "no port data" → RTBH, never emitting a useless `dst_port:0` FlowSpec rule (`blackwall-flow`). **Config/API:** the config lexer now treats `#` as a comment only at line-start or after whitespace, so a value like `md5=sec#ret` is no longer silently truncated (`blackwall-config`); `parse_and_resolve()` validates the policy at load for `blackwalld flow` (the M0 entry point) and `bird-config`, so a semantically-invalid config fails at container start instead of mis-behaving through the observation window; and the read API's `?limit=` is clamped to `[1, 1000]` so a negative or huge value can't 500 or dump a whole table (`blackwall-api`). **New observability** (`/metrics`, `blackwall_flow_*`): `min_sample_suppressed_total`, `sampling_near_ceiling_total{pop}`, `sample_decode_errors_total` (distinct from the per-datagram `decode_errors`), and `detections_opened_total` / `detections_cleared_total` — together they let an operator tell a quiet POP from one silently dropping samples. Non-breaking (new CLI flags default to prior behaviour; `--bps-threshold` is documented as L2/frame-length based).
 - Lab CI reliability (blackwall#88): a stuck in-scenario command could hang a gate silently until the job cap. Root cause was two-fold — the harness ran one-shot commands via an unbounded `Command::output()`, and gate drivers ran `cargo test` *inside* the netns, whose long-lived cargo process (holding the Cargo build lock + a subprocess tree) could wedge in CI where a `sudo`-spawned tree escapes GitHub's step timeout. Fixes: `netns::run` now bounds and kills every command (own process group, drained pipes, reader-join timeouts) so it can never block the caller; daemon teardown reaps with a bounded wait; `spawn_bird` redirects its stdio like the other daemons so a killed lab can't leak a pipe-holder; and gates now run the **pre-built** interop binary directly (`scripts/build-lab-tests.sh` → `target/debug/lab-tests/<pkg>-<test>`) instead of `cargo test`, removing the cargo lock/daemon from the scenario entirely.
 - RTBH hardening ahead of production wiring (sub-project C, C1c Part A) — fixes a code review found in the C1b controller and C1a speaker. **The RTBH controller leaked blackholes permanently:** the flow detector emits `Cleared` once, and if it arrived before the RTBH `hold_down` elapsed the withdraw was dropped with no retry, so the victim `/32` stayed blackholed (self-DoSed) forever. The controller now defers such a clear and completes it from a periodic `tick`, adds a `max_ttl` auto-blackhole backstop against a dropped/missed `Cleared`, tracks per-entry origin so an operator (`Manual`) blackhole is never auto-cleared, and cancels a deferred clear when a re-attack (`Opened`/`Updated`) re-asserts the target (`blackwall-rtbh`). **The BGP speaker could panic on a large config:** `push_attr` truncated attribute lengths to one byte, so ≥64 communities overflowed and killed the session task silently — it now emits RFC 4271 §4.3 extended-length attributes; `encode_nlri` also truncates host bits so a non-host prefix can't emit malformed NLRI (`blackwall-bgp`). The speaker further validates its config as iBGP (`local_asn == peer_asn`) and the peer's OPEN (ASN and hold-time, rejecting with the correct NOTIFICATION codes), waits for the peer's KEEPALIVE before declaring the session Established (bounded by the OpenConfirm hold timer), and surfaces send failures on `BgpHandle` instead of silently dropping routes. Controller logic is unit-tested; the speaker changes are covered by the `bgp-bird` lab gate.

bin/blackwalld/src/metrics.rs

@@ -10,17 +10,6 @@ use std::sync::Arc;
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::TcpListener;
 
-/// Current wall-clock time in ms since epoch; `0` if the clock is somehow
-/// before the epoch. Used only to compute the per-POP last-seen-seconds
-/// gauge at scrape time.
-fn now_ms() -> u64 {
-    std::time::SystemTime::now()
-        .duration_since(std::time::UNIX_EPOCH)
-        .ok()
-        .and_then(|d| u64::try_from(d.as_millis()).ok())
-        .unwrap_or(0)
-}
-
 /// Everything the scrape handler reads at request time. Cheap to clone (an
 /// `Arc<Store>`, a cloneable `BgpHandle`, an `Arc<CollectorMetrics>`).
 #[derive(Clone)]
@@ -245,6 +234,62 @@ fn xdp_block(sources: &MetricsSources) -> Option<String> {
     }))
 }
 
+/// Render only the per-POP telemetry blocks (`blackwall_flow_pop_last_seen_seconds`,
+/// `blackwall_flow_agent_sampling_mismatch_total`,
+/// `blackwall_flow_sampling_near_ceiling_total`) from an already-sorted `stats`
+/// slice. Pure and DB-free (so it is unit-testable). `now_ms` must be the same
+/// monotonic clock as `AgentStat.last_seen_ms` — see [`agent_stats_block`].
+fn render_agent_pop_stats(stats: &[blackwall_flow::AgentStat], now_ms: u64) -> String {
+    let mut out = String::new();
+    if stats.is_empty() {
+        return out;
+    }
+    let _ = writeln!(
+        out,
+        "# HELP blackwall_flow_pop_last_seen_seconds Seconds since this POP's sFlow agent was last observed"
+    );
+    let _ = writeln!(out, "# TYPE blackwall_flow_pop_last_seen_seconds gauge");
+    for s in stats {
+        let age_secs = now_ms.saturating_sub(s.last_seen_ms) / 1000;
+        let _ = writeln!(
+            out,
+            "blackwall_flow_pop_last_seen_seconds{{pop=\"{}\"}} {age_secs}",
+            s.pop
+        );
+    }
+    let _ = writeln!(
+        out,
+        "\n# HELP blackwall_flow_agent_sampling_mismatch_total Sampling-rate mismatches clamped per POP"
+    );
+    let _ = writeln!(
+        out,
+        "# TYPE blackwall_flow_agent_sampling_mismatch_total counter"
+    );
+    for s in stats {
+        let _ = writeln!(
+            out,
+            "blackwall_flow_agent_sampling_mismatch_total{{pop=\"{}\"}} {}",
+            s.pop, s.mismatches
+        );
+    }
+    let _ = writeln!(
+        out,
+        "\n# HELP blackwall_flow_sampling_near_ceiling_total Samples per POP whose trusted sampling rate landed at or above half the max-sampling-factor ceiling"
+    );
+    let _ = writeln!(
+        out,
+        "# TYPE blackwall_flow_sampling_near_ceiling_total counter"
+    );
+    for s in stats {
+        let _ = writeln!(
+            out,
+            "blackwall_flow_sampling_near_ceiling_total{{pop=\"{}\"}} {}",
+            s.pop, s.near_ceiling
+        );
+    }
+    out
+}
+
 /// Render the per-POP telemetry blocks (`blackwall_flow_pop_last_seen_seconds`,
 /// `blackwall_flow_agent_sampling_mismatch_total`,
 /// `blackwall_flow_sampling_near_ceiling_total`) plus the
@@ -256,8 +301,10 @@ fn xdp_block(sources: &MetricsSources) -> Option<String> {
 /// POP names are dynamic labels unknown at compile time, so — like
 /// [`xdp_block`]'s `reason` label — this hand-writes the exposition text
 /// directly rather than going through [`Metric`], which only carries a
-/// `&'static str` name. `now_ms` is the current wall-clock time, used to
-/// turn each POP's last-seen timestamp into an age in seconds.
+/// `&'static str` name. `now_ms` MUST be [`blackwall_flow::monotonic_now_ms`]
+/// (the same monotonic clock the collector stamps `AgentStat.last_seen_ms`
+/// with), so the per-POP last-seen age is real seconds — NOT wall-clock epoch,
+/// which would make every age a nonsense ~epoch-sized value.
 fn agent_stats_block(sources: &MetricsSources, now_ms: u64) -> Option<String> {
     let snapshot = sources.agent_stats.as_ref()?;
     let mut stats: Vec<blackwall_flow::AgentStat> = match snapshot.lock() {
@@ -267,52 +314,7 @@ fn agent_stats_block(sources: &MetricsSources, now_ms: u64) -> Option<String> {
     // Deterministic scrape output regardless of HashMap iteration order.
     stats.sort_by(|a, b| a.pop.cmp(&b.pop));
 
-    let mut out = String::new();
-    if !stats.is_empty() {
-        let _ = writeln!(
-            out,
-            "# HELP blackwall_flow_pop_last_seen_seconds Seconds since this POP's sFlow agent was last observed"
-        );
-        let _ = writeln!(out, "# TYPE blackwall_flow_pop_last_seen_seconds gauge");
-        for s in &stats {
-            let age_secs = now_ms.saturating_sub(s.last_seen_ms) / 1000;
-            let _ = writeln!(
-                out,
-                "blackwall_flow_pop_last_seen_seconds{{pop=\"{}\"}} {age_secs}",
-                s.pop
-            );
-        }
-        let _ = writeln!(
-            out,
-            "\n# HELP blackwall_flow_agent_sampling_mismatch_total Sampling-rate mismatches clamped per POP"
-        );
-        let _ = writeln!(
-            out,
-            "# TYPE blackwall_flow_agent_sampling_mismatch_total counter"
-        );
-        for s in &stats {
-            let _ = writeln!(
-                out,
-                "blackwall_flow_agent_sampling_mismatch_total{{pop=\"{}\"}} {}",
-                s.pop, s.mismatches
-            );
-        }
-        let _ = writeln!(
-            out,
-            "\n# HELP blackwall_flow_sampling_near_ceiling_total Samples per POP whose trusted sampling rate landed at or above half the max-sampling-factor ceiling"
-        );
-        let _ = writeln!(
-            out,
-            "# TYPE blackwall_flow_sampling_near_ceiling_total counter"
-        );
-        for s in &stats {
-            let _ = writeln!(
-                out,
-                "blackwall_flow_sampling_near_ceiling_total{{pop=\"{}\"}} {}",
-                s.pop, s.near_ceiling
-            );
-        }
-    }
+    let mut out = render_agent_pop_stats(&stats, now_ms);
 
     // Always emitted (a single scalar, not per-label) so the series exists
     // even before any known agent has been observed.
@@ -467,7 +469,7 @@ async fn handle_conn(mut sock: tokio::net::TcpStream, sources: &MetricsSources)
             }
             body.push_str(&xdp);
         }
-        if let Some(agent) = agent_stats_block(sources, now_ms()) {
+        if let Some(agent) = agent_stats_block(sources, blackwall_flow::monotonic_now_ms()) {
             if !body.is_empty() {
                 body.push('\n');
             }
@@ -493,10 +495,39 @@ async fn handle_conn(mut sock: tokio::net::TcpStream, sources: &MetricsSources)
 
 #[cfg(test)]
 mod tests {
-    use super::stateless_metrics;
+    use super::{render_agent_pop_stats, stateless_metrics};
     use blackwall_deception::transport::StatelessMetrics;
     use blackwall_metrics::render_prometheus;
 
+    #[test]
+    fn pop_last_seen_age_uses_monotonic_clock_not_epoch() {
+        // Regression for the clock-domain bug: the collector stamps
+        // `AgentStat.last_seen_ms` from the monotonic clock, so the renderer must
+        // compute the age against the SAME clock. A just-seen agent must read
+        // ~0 s, not ~epoch (the old bug rendered ~1.78e9 by subtracting a
+        // monotonic timestamp from an epoch "now").
+        let stats = vec![blackwall_flow::AgentStat {
+            pop: "kc".to_string(),
+            last_seen_ms: blackwall_flow::monotonic_now_ms(), // just observed
+            mismatches: 0,
+            near_ceiling: 0,
+        }];
+        let body = render_agent_pop_stats(&stats, blackwall_flow::monotonic_now_ms());
+        let line = body
+            .lines()
+            .find(|l| l.starts_with("blackwall_flow_pop_last_seen_seconds{pop=\"kc\"}"))
+            .expect("pop_last_seen line present");
+        let age: u64 = line
+            .rsplit(' ')
+            .next()
+            .and_then(|v| v.parse().ok())
+            .expect("age value parses");
+        assert!(
+            age < 5,
+            "a just-seen agent's last-seen age must be ~0s, got {age} (clock-domain regression)"
+        );
+    }
+
     #[test]
     fn stateless_metrics_renders_all_four_counters_with_expected_values() {
         let stateless = StatelessMetrics::new();

crates/blackwall-flow/src/collector_net.rs

@@ -19,10 +19,20 @@ fn clock_base() -> Instant {
 
 /// Milliseconds since process start (monotonic). Used for all detector windowing,
 /// eviction, and hold-down math — never affected by NTP/wall-clock steps.
-fn now_ms() -> u64 {
+///
+/// Exposed publicly so the `/metrics` renderer can compute each POP's last-seen
+/// age against the **same** clock the collector stamps observations with:
+/// `AgentStat.last_seen_ms` is a monotonic timestamp, so subtracting it from an
+/// epoch "now" yields a nonsense (~epoch-sized) age.
+#[must_use]
+pub fn monotonic_now_ms() -> u64 {
     u64::try_from(clock_base().elapsed().as_millis()).unwrap_or(u64::MAX)
 }
 
+fn now_ms() -> u64 {
+    monotonic_now_ms()
+}
+
 /// Run the collector until the process ends. Binds `listen`, decodes each
 /// datagram into the `detector`, and every `tick_interval_ms` evaluates the
 /// window and forwards events to `sink`. Decode errors are logged and skipped.
@@ -94,3 +104,22 @@ pub async fn run_collector(
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::monotonic_now_ms;
+
+    #[test]
+    fn monotonic_now_ms_is_process_uptime_not_epoch() {
+        let a = monotonic_now_ms();
+        let b = monotonic_now_ms();
+        assert!(b >= a, "monotonic clock must be non-decreasing");
+        // Process uptime in ms stays far below epoch-ms scale (~1.78e12) for
+        // years; this guards against a regression back to a wall-clock source,
+        // which is what made pop_last_seen_seconds read ~epoch.
+        assert!(
+            a < 1_000_000_000,
+            "must be process-uptime ms, not epoch ms; got {a}"
+        );
+    }
+}

crates/blackwall-flow/src/detector.rs

@@ -136,7 +136,10 @@ pub trait Detector {
 pub struct AgentStat {
     /// POP name for this agent, as configured in the registry.
     pub pop: String,
-    /// Timestamp (ms since epoch) this agent was last observed.
+    /// Monotonic timestamp (ms since process start, from
+    /// [`crate::monotonic_now_ms`]) this agent was last observed. Compare it
+    /// against `monotonic_now_ms()` — NOT wall-clock epoch — to get a staleness
+    /// age; mixing the two clocks yields a ~epoch-sized nonsense value.
     pub last_seen_ms: u64,
     /// Count of samples from this agent whose reported sampling rate was
     /// clamped because it deviated far from the agent's expected rate.

crates/blackwall-flow/src/lib.rs

@@ -12,7 +12,7 @@ mod sflow;
 mod sink;
 
 pub use agents::AgentRegistry;
-pub use collector_net::run_collector;
+pub use collector_net::{monotonic_now_ms, run_collector};
 pub use detector::{
     AgentStat, AttackKind, Detection, DetectionEvent, Detector, DetectorConfig, Severity,
     ThresholdDetector,