diff --git a/data_plane/tests/monitor_grpc.rs b/data_plane/tests/monitor_grpc.rs index aa9e212f..3bf21abf 100644 --- a/data_plane/tests/monitor_grpc.rs +++ b/data_plane/tests/monitor_grpc.rs @@ -156,3 +156,87 @@ async fn stays_quiet_below_tau() { "edge should have received at least the initial grant" ); } + +fn register_edge(edge_id: &str) -> EdgeToCoord { + EdgeToCoord { + msg: Some(edge_to_coord::Msg::Reg(MonitorRegister { + edge_id: edge_id.into(), + agg_id: 1, + key: vec![], + epoch_window_ms: 60_000, + window_start_ms: 0, + })), + } +} + +fn report_edge(edge_id: &str, value: f64, seq: u64, rate: f64) -> EdgeToCoord { + EdgeToCoord { + msg: Some(edge_to_coord::Msg::Report(MonitorReport { + edge_id: edge_id.into(), + agg_id: 1, + key: vec![], + window_start_ms: 0, + local_value: value, + round: 0, + seq, + rate, + })), + } +} + +/// Live multi-edge coupling: two edges open real gRPC streams, report SKEWED +/// rates, and the coordinator ships back differentiated `SlackGrant.sample_p` +/// over the wire — the hot edge sampled harder (smaller p) than the quiet one, +/// both at/above the ε-derived coupling floor. τ is large so only grants flow. +#[tokio::test] +async fn two_edges_get_differentiated_sample_p_over_grpc() { + use asap_otel_proto::monitor::v1::coord_to_edge; + + async fn connect_edge(url: &str) -> (mpsc::Sender, Arc>>) { + let mut client = MonitorServiceClient::connect(url.to_string()).await.unwrap(); + let (tx, rx) = mpsc::channel(32); + let mut inbound = client + .monitor(ReceiverStream::new(rx)) + .await + .unwrap() + .into_inner(); + let ps = Arc::new(Mutex::new(Vec::::new())); + let psc = ps.clone(); + tokio::spawn(async move { + while let Some(Ok(env)) = inbound.next().await { + if let Some(coord_to_edge::Msg::Grant(g)) = env.msg { + psc.lock().unwrap().push(g.sample_p); + } + } + }); + (tx, ps) + } + + let (url, _alerts) = start_server(vec![sum_cfg(1_000_000.0)]).await; + let (tx_hot, p_hot) = connect_edge(&url).await; + let (tx_quiet, p_quiet) = connect_edge(&url).await; + + tx_hot.send(register_edge("e1")).await.unwrap(); + tx_quiet.send(register_edge("e2")).await.unwrap(); + tokio::time::sleep(Duration::from_millis(60)).await; + + // e1 hot (100k items/win), e2 quiet (1k/win); small local values stay far + // below τ so only grants (carrying sample_p) flow, never an alert. + for seq in 1..=4 { + tx_hot.send(report_edge("e1", 1.0, seq, 100_000.0)).await.unwrap(); + tx_quiet.send(report_edge("e2", 1.0, seq, 1_000.0)).await.unwrap(); + tokio::time::sleep(Duration::from_millis(40)).await; + } + tokio::time::sleep(Duration::from_millis(250)).await; + + let hot = *p_hot.lock().unwrap().last().expect("hot edge received a grant"); + let quiet = *p_quiet.lock().unwrap().last().expect("quiet edge received a grant"); + println!("LIVE coupling over gRPC: hot(e1,rate=100k) sample_p={hot} | quiet(e2,rate=1k) sample_p={quiet}"); + + assert!( + hot < quiet, + "hot edge sample_p {hot} should be < quiet edge sample_p {quiet}" + ); + assert!(hot > 0.0 && hot <= 1.0, "hot p out of range: {hot}"); + assert!(quiet > 0.0 && quiet <= 1.0, "quiet p out of range: {quiet}"); +}