feat(gos): split Discipline B — retire alerting, decouple sample_p from countdown

control_plane/src/monitor/mod.rs

@@ -27,11 +27,6 @@ pub enum ViolationKind {
     Bandwidth,
     Accuracy,
     Cpu,
-    /// A continuous-monitoring (CDM) global aggregate crossed its threshold τ.
-    /// Emitted by the data-plane monitor coordinator, not the metrics scraper:
-    /// `agent_id` carries the monitor id, `observed` the global estimate at
-    /// fire time, `threshold` the τ.
-    GlobalThresholdCrossed,
 }
 
 impl std::fmt::Display for ViolationKind {
@@ -40,7 +35,6 @@ impl std::fmt::Display for ViolationKind {
             ViolationKind::Bandwidth => write!(f, "bandwidth"),
             ViolationKind::Accuracy => write!(f, "accuracy"),
             ViolationKind::Cpu => write!(f, "cpu"),
-            ViolationKind::GlobalThresholdCrossed => write!(f, "global_threshold_crossed"),
         }
     }
 }

data_plane/src/bin/monitor_coordinator_harness.rs

@@ -1,20 +1,28 @@
 //! Minimal CDM monitor-coordinator harness for the cross-language e2e
-//! (deploy/mvp-multinode/scripts/monitor_e2e.sh in ASAPCollector). Runs ONLY
-//! the tonic MonitorService with a single Sum monitor and prints a line to
-//! stdout when the global threshold fires, then exits 0 — so a driver script
-//! can assert the Go edge ⇄ Rust coordinator wire path end-to-end without
-//! standing up the full data-plane.
+//! (deploy/mvp-multinode/scripts/monitor_e2e.sh in ASAPCollector). Runs the
+//! tonic MonitorService with a single monitor and prints a line to stdout once
+//! the coordinator has computed a REAL coordinated-sampling grant for the
+//! connecting edge (i.e. the edge's periodic rate report reached the
+//! coordinator and a non-trivial sample_p came back), then exits 0 — so a
+//! driver script can assert the Go edge ⇄ Rust coordinator wire path
+//! end-to-end without standing up the full data-plane.
+//!
+//! Global-threshold alerting is retired (see `data_plane::monitor` module
+//! docs) — this harness no longer waits for or reports an alert.
 //!
 //! Usage: monitor_coordinator_harness <port> <agg_id> <tau> <window_ms> [timeout_secs] [key]
 //!
-//! `key` is the monitor's group key (the edge's canonical `k=v;k2=v2` label
-//! encoding). Scalar (Sum) monitors register under their series-group key, so
-//! the harness config must carry the same key or registrations are rejected
-//! as unconfigured. Empty (default) = ungrouped.
+//! `tau` is accepted for CLI/back-compat but unused. `key` is the monitor's
+//! group key (the edge's canonical `k=v;k2=v2` label encoding). Scalar (Sum)
+//! monitors register under their series-group key, so the harness config must
+//! carry the same key or registrations are rejected as unconfigured. Empty
+//! (default) = ungrouped.
 
-use std::sync::Arc;
+use data_plane::monitor::{MonitorConfig, MonitorCoordinator, MonitorServiceImpl};
 
-use data_plane::monitor::{AlertSink, MonitorConfig, MonitorCoordinator, MonitorServiceImpl};
+/// The edge_id the Go e2edriver connects as (asap-precompute-go/monitor/
+/// grpcclient/cmd/e2edriver/main.go: `monitor.NewEngine("e2e-edge", ...)`).
+const E2E_EDGE_ID: &str = "e2e-edge";
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
@@ -26,40 +34,46 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let timeout_secs: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(30);
     let key: Vec<u8> = args.next().map(|s| s.into_bytes()).unwrap_or_default();
 
-    // Exit-on-alert: the sink prints a machine-greppable line and signals the
-    // run to finish.
-    let (done_tx, mut done_rx) = tokio::sync::mpsc::channel::<()>(1);
-    let sink: AlertSink = Arc::new(move |v| {
-        println!(
-            "MONITOR_ALERT monitor={} observed={} threshold={}",
-            v.agent_id, v.observed, v.threshold
-        );
-        let _ = done_tx.try_send(());
-    });
-
     let cfg = MonitorConfig {
         agg_id,
-        key,
+        key: key.clone(),
         tau,
         epsilon: 0.05,
         window_ms,
         ..Default::default()
     };
-    let coord = MonitorCoordinator::new(vec![cfg], sink);
-    let svc = MonitorServiceImpl::new(coord).into_server();
+    let coord = MonitorCoordinator::new(vec![cfg]);
+    let svc = MonitorServiceImpl::new(coord.clone()).into_server();
     let addr = format!("0.0.0.0:{port}").parse()?;
-    eprintln!("harness: serving MonitorService on {addr} (agg_id={agg_id}, tau={tau}, window_ms={window_ms})");
+    eprintln!("harness: serving MonitorService on {addr} (agg_id={agg_id}, window_ms={window_ms})");
     println!("HARNESS_READY {port}");
 
+    // Poll for a real coordinated-sampling grant reaching this edge — the
+    // proof that the edge's periodic rate report made the live gRPC round
+    // trip and the coordinator answered with a computed sample_p.
+    let poll = {
+        let coord = coord.clone();
+        let key = key.clone();
+        async move {
+            loop {
+                if let Some(p) = coord.granted_sample_p(agg_id, &key, E2E_EDGE_ID).await {
+                    println!("MONITOR_GRANT edge={E2E_EDGE_ID} agg_id={agg_id} sample_p={p}");
+                    return;
+                }
+                tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+            }
+        }
+    };
+
     tokio::select! {
         r = tonic::transport::Server::builder().add_service(svc).serve(addr) => {
             r?;
         }
-        _ = done_rx.recv() => {
-            eprintln!("harness: alert fired — exiting");
+        _ = poll => {
+            eprintln!("harness: coordinated-sampling grant observed — exiting");
         }
         _ = tokio::time::sleep(std::time::Duration::from_secs(timeout_secs)) => {
-            eprintln!("harness: timeout after {timeout_secs}s with no alert");
+            eprintln!("harness: timeout after {timeout_secs}s with no grant");
             std::process::exit(2);
         }
     }

data_plane/src/main.rs

@@ -594,14 +594,12 @@ async fn main() -> Result<()> {
         None
     };
 
-    // CDM monitor coordinator: bidi MonitorService gRPC that runs the
-    // slack-countdown protocol over the streaming-config `monitors:` specs and
-    // fires a global-threshold alert through the violation sink (logged here;
-    // the control-plane replanner can subscribe via the same Violation shape).
+    // Coordinated-sampling monitor coordinator: bidi MonitorService gRPC that
+    // answers each edge's periodic rate report with its whole-sketch ε-floor
+    // sample_p grant. Global-threshold alerting is retired (see
+    // data_plane::monitor module docs) — this coordinator never fires one.
     let monitor_handle = if args.enable_monitor_coordinator {
-        use data_plane::monitor::{
-            AlertSink, Functional, MonitorConfig, MonitorCoordinator, MonitorServiceImpl,
-        };
+        use data_plane::monitor::{Functional, MonitorConfig, MonitorCoordinator, MonitorServiceImpl};
         let specs: Vec<MonitorConfig> = streaming_config
             .monitors()
             .iter()
@@ -617,15 +615,7 @@ async fn main() -> Result<()> {
         if specs.is_empty() {
             warn!("--enable-monitor-coordinator set but streaming-config has no `monitors:` yet — the coordinator will pick them up live when the control plane pushes a config (hot-reload)");
         }
-        let sink: AlertSink = Arc::new(|v| {
-            warn!(
-                monitor = %v.agent_id,
-                observed = v.observed,
-                threshold = v.threshold,
-                "CDM global threshold crossed"
-            );
-        });
-        let coord = MonitorCoordinator::new(specs, sink);
+        let coord = MonitorCoordinator::new(specs);
 
         // Hot-reload watcher: the coordinator reads `monitors:` once at boot, but
         // the control plane pushes the real config slightly AFTER boot via the