From d18bfe1ee920730735dddcd55a573fee26bcb534 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 11 Jun 2026 12:34:02 -0600 Subject: [PATCH] =?UTF-8?q?feat(monitor):=20dynamic=20coordinator=E2=86=92?= =?UTF-8?q?edge=20sampling=20coupling=20(coordinator=20half)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wires the coordinator side: it tracks each edge's reported per-window rate, computes a per-edge update-sampling probability p_i that minimizes total edge update work at a target accuracy, and ships it in each edge's SlackGrant.sample_p — orthogonal to the slack countdown (sampling is an additive grant field). Proto: SlackGrant.sample_p (field 6) + MonitorReport.rate (field 8); prost regenerates the bindings at build. - sampling_alloc.rs (new): Rust port of the Go AllocateSampleRates — p_i = clamp(sqrt(λ)·sqrt(f_i/rate_i), 0, 1] with λ tuned by bisection so the merged sampling-variance Σ f_i(1−p_i)/p_i ≤ var_budget binds; uniform_sample_rate; and epsilon_sample_floor (p ≥ 1/(1+ε²·rate)) — the ε_cdm ≳ ε_s coupling rule made pointwise. Edge cases mirror the Go impl (rate≤0/freq≤0 ⇒ p=1). - coordinator.rs: EdgeView.rate tracked from MonitorReport.rate via on_report; Action::Grant carries sample_p; allocate_p() builds the per-edge rate vector, calls allocate_sample_rates, clamps each p_i up to the ε-floor; rebroadcast grants p_i only with ≥2 edges (single-edge/unknown ⇒ p=1). Slack/alert logic untouched. - server.rs: apply_report/handle_msg thread rep.rate → on_report; dispatch sets SlackGrant.sample_p from Action::Grant. - var_budget = (ε·τ)²: the CDM tolerance on the monitored value, in quantities the coordinator already holds (ε, τ from MonitorConfig) — the design doc's ε_cdm ≳ ε_s coupling. The per-edge floor enforces it pointwise too. Tests: allocation (skewed fleet beats uniform-p at equal variance; flat fleet no win; floor/edge cases); coordinator (single edge ⇒ p=1; skewed rates ⇒ differentiated sample_p, hot --- .../proto/monitor/monitor.proto | 2 + .../tests/coupling_wire_compat.rs | 56 ++++ data_plane/src/monitor/coordinator.rs | 144 ++++++++-- data_plane/src/monitor/mod.rs | 2 + data_plane/src/monitor/sampling_alloc.rs | 252 ++++++++++++++++++ data_plane/src/monitor/server.rs | 6 +- 6 files changed, 444 insertions(+), 18 deletions(-) create mode 100644 crates/asap_otel_proto/tests/coupling_wire_compat.rs create mode 100644 data_plane/src/monitor/sampling_alloc.rs diff --git a/crates/asap_otel_proto/proto/monitor/monitor.proto b/crates/asap_otel_proto/proto/monitor/monitor.proto index 6a1d9123..781cb2e5 100644 --- a/crates/asap_otel_proto/proto/monitor/monitor.proto +++ b/crates/asap_otel_proto/proto/monitor/monitor.proto @@ -53,6 +53,7 @@ message MonitorReport { double local_value = 5; // current additive local value at report time uint64 round = 6; // round this report answers uint64 seq = 7; // per-edge monotonic counter; idempotent-retransmit dedup + double rate = 8; // edge's observed items/window — feeds the coordinator's p_i allocation } // coordinator → edge: this round's per-edge slack budget. The edge reports once @@ -63,6 +64,7 @@ message SlackGrant { uint64 round = 3; double local_slack = 4; uint64 window_start_ms = 5; + double sample_p = 6; // distributed-NitroSketch update-sampling prob the coordinator allocates (0=unset/p=1); edge applies via WithSampleP at EpochReset } // coordinator → edge: demand the current local value now (round close). diff --git a/crates/asap_otel_proto/tests/coupling_wire_compat.rs b/crates/asap_otel_proto/tests/coupling_wire_compat.rs new file mode 100644 index 00000000..c4266964 --- /dev/null +++ b/crates/asap_otel_proto/tests/coupling_wire_compat.rs @@ -0,0 +1,56 @@ +//! Cross-language wire-compat gate for the dynamic coordinator<->sampling +//! coupling fields: bytes are emitted by the Go edge +//! (asap-precompute-go/monitor/grpcclient/monitorpb TestCouplingFixture) and +//! MUST decode field-for-field here via prost — proving SlackGrant.sample_p +//! (coord->edge) and MonitorReport.rate (edge->coord) cross the wire. +use asap_otel_proto::monitor::v1::{ + coord_to_edge, edge_to_coord, CoordToEdge, EdgeToCoord, +}; +use prost::Message; + +fn unhex(s: &str) -> Vec { + (0..s.len()).step_by(2).map(|i| u8::from_str_radix(&s[i..i + 2], 16).unwrap()).collect() +} + +#[test] +fn go_grant_sample_p_decodes_in_rust() { + let bytes = unhex("0a1d082a1807210000000000000c402880d095ffbc3131000000000000d03f"); + let env = CoordToEdge::decode(&bytes[..]).expect("decode CoordToEdge"); + match env.msg.expect("msg") { + coord_to_edge::Msg::Grant(g) => { + assert_eq!(g.agg_id, 42); + assert_eq!(g.round, 7); + assert!((g.sample_p - 0.25).abs() < 1e-12, "sample_p={}", g.sample_p); + } + other => panic!("expected Grant, got {:?}", other), + } +} + +#[test] +fn go_report_rate_decodes_in_rust() { + let bytes = unhex("12200a06656467652d37102a2900000000004a934030033809410000000000c09240"); + let env = EdgeToCoord::decode(&bytes[..]).expect("decode EdgeToCoord"); + match env.msg.expect("msg") { + edge_to_coord::Msg::Report(r) => { + assert_eq!(r.edge_id, "edge-7"); + assert_eq!(r.agg_id, 42); + assert!((r.rate - 1200.0).abs() < 1e-9, "rate={}", r.rate); + } + other => panic!("expected Report, got {:?}", other), + } +} + +#[test] +fn rust_grant_sample_p_emit_for_go() { + // The REAL coord->edge direction: Rust(prost) encodes a grant; Go decodes it. + use asap_otel_proto::monitor::v1::SlackGrant; + let env = CoordToEdge { + msg: Some(coord_to_edge::Msg::Grant(SlackGrant { + agg_id: 99, round: 4, local_slack: 1.0, window_start_ms: 1_700_000_000_000, + sample_p: 0.3, ..Default::default() + })), + }; + let mut buf = Vec::new(); + env.encode(&mut buf).unwrap(); + println!("RUST_GRANT_HEX={}", buf.iter().map(|b| format!("{:02x}", b)).collect::()); +} diff --git a/data_plane/src/monitor/coordinator.rs b/data_plane/src/monitor/coordinator.rs index d7db4036..5ad71ef1 100644 --- a/data_plane/src/monitor/coordinator.rs +++ b/data_plane/src/monitor/coordinator.rs @@ -30,6 +30,8 @@ use std::collections::HashMap; +use super::sampling_alloc::{allocate_sample_rates, epsilon_sample_floor}; + /// Static configuration for one monitor, sourced from the streaming-config /// `monitors:` section (τ authoritative here, not at the edge). #[derive(Clone, Debug)] @@ -44,12 +46,16 @@ pub struct MonitorConfig { /// An action the coordinator wants the transport to perform. #[derive(Clone, Debug, PartialEq)] pub enum Action { - /// Send a per-round slack grant to one edge. + /// Send a per-round slack grant to one edge. `sample_p` is the distributed- + /// NitroSketch update-sampling probability the coordinator allocates for this + /// edge (1.0 = no sampling); it is an ADDITIONAL field orthogonal to the + /// slack countdown. Grant { edge_id: String, round: u64, local_slack: f64, window_start_ms: u64, + sample_p: f64, }, /// The global aggregate crossed τ (within ε): fire exactly once per epoch. Alert { @@ -64,6 +70,9 @@ struct EdgeView { known_value: f64, last_seq: u64, seen_seq: bool, + /// Last `rate` (items/window) the edge reported; 0 = unknown ⇒ no sampling + /// for this edge. Feeds the coordinated `p_i` allocation. + rate: f64, } /// One monitor's coordinator state for the current epoch. @@ -170,6 +179,7 @@ impl Monitor { window_start_ms: u64, local_value: f64, seq: u64, + rate: f64, ) -> Vec { if window_start_ms != self.window_start_ms { // Could be a future epoch we haven't advanced to yet — advance on @@ -194,6 +204,11 @@ impl Monitor { if local_value > edge.known_value { edge.known_value = local_value; } + // Track the edge's latest observed per-window rate (ignore non-positive + // = unknown), which drives the coordinated sampling allocation. + if rate > 0.0 { + edge.rate = rate; + } self.rebroadcast() } @@ -206,6 +221,48 @@ impl Monitor { self.rebroadcast() } + /// Coordinated update-sampling allocation for the current edge set, keyed by + /// edge id. Returns `p_i ∈ (0,1]` per edge, each clamped up to the CDM- + /// threshold coupling floor so sampling noise stays within ε. + /// + /// The merged sampling-variance budget `V` is derived from what the + /// coordinator already holds: the CDM tolerance on the monitored value is + /// `ε·τ`, and update-sampling injects a std-dev of `√(Σ f_i(1−p_i)/p_i)` into + /// `Σf̂`. Keeping that band within the threshold tolerance means + /// `√V ≤ ε·τ`, i.e. **`V = (ε·τ)²`** — the sampling noise is absorbed within + /// the CDM band (the "ε_cdm ≳ ε_s" coupling). Lacking a per-key frequency + /// split, we use each edge's `rate` as the `freqs` proxy (so `p_i ∝ + /// 1/√rate_i`). With <2 edges or all rates unknown the allocation degenerates + /// to `p=1` everywhere (no sampling). + fn allocate_p(&self) -> HashMap { + // Stable edge order so the rate vector aligns with the p vector. + let ids: Vec<&String> = self.edges.keys().collect(); + let rates: Vec = ids.iter().map(|id| self.edges[*id].rate).collect(); + // freqs proxy = rates (no per-key split available at the coordinator). + let var_budget = { + let band = self.cfg.epsilon * self.cfg.tau; + band * band + }; + let mut p_vec = allocate_sample_rates(&rates, &rates, var_budget); + // Enforce the CDM-threshold coupling floor per edge: never sample so hard + // that ε_s = √((1−p)/(p·rate)) exceeds the agg's ε. + for (i, &rate) in rates.iter().enumerate() { + let floor = epsilon_sample_floor(self.cfg.epsilon, rate); + if p_vec[i] < floor { + p_vec[i] = floor; + } + } + ids.into_iter() + .cloned() + .zip(p_vec) + .map(|(id, p)| { + // Single edge or unknown rate ⇒ no sampling (p=1). + let p = if p > 0.0 && p <= 1.0 { p } else { 1.0 }; + (id, p) + }) + .collect() + } + /// Recompute Δ; fire the alert once if within ε, otherwise advance the round /// and emit a fresh slack grant to every edge. fn rebroadcast(&mut self) -> Vec { @@ -225,6 +282,12 @@ impl Monitor { let slack = self.slack(); let round = self.round; let ws = self.window_start_ms; + // With a single edge there is no rate vector to coordinate over → p=1. + let p_by_edge = if self.edges.len() >= 2 { + self.allocate_p() + } else { + HashMap::new() + }; self.edges .keys() .map(|edge_id| Action::Grant { @@ -232,6 +295,7 @@ impl Monitor { round, local_slack: slack, window_start_ms: ws, + sample_p: p_by_edge.get(edge_id).copied().unwrap_or(1.0), }) .collect() } @@ -264,6 +328,13 @@ mod tests { } } + fn sample_p_in(actions: &[Action], edge: &str) -> f64 { + match grant_for(actions, edge) { + Some(Action::Grant { sample_p, .. }) => *sample_p, + _ => panic!("no grant for {edge}"), + } + } + #[test] fn registration_grants_initial_slack() { let mut m = Monitor::new(cfg(100.0)); @@ -286,10 +357,10 @@ mod tests { let mut m = Monitor::new(cfg(100.0)); m.on_register("e1", 60_000, 0); // slack 50 // e1 reports 50 → estimate=50, Δ=50, slack=25. - let a = m.on_report("e1", 0, 50.0, 1); + let a = m.on_report("e1", 0, 50.0, 1, 0.0); assert_eq!(slack_in(&a, "e1"), 25.0); // reports 75 → estimate=75, Δ=25, slack=12.5. - let a = m.on_report("e1", 0, 75.0, 2); + let a = m.on_report("e1", 0, 75.0, 2, 0.0); assert_eq!(slack_in(&a, "e1"), 12.5); } @@ -298,15 +369,15 @@ mod tests { let mut m = Monitor::new(cfg(100.0)); // epsilon 0.05 → alert when Δ ≤ 5 m.on_register("e1", 60_000, 0); assert!(matches!( - m.on_report("e1", 0, 50.0, 1).as_slice(), + m.on_report("e1", 0, 50.0, 1, 0.0).as_slice(), [Action::Grant { .. }] )); assert!(matches!( - m.on_report("e1", 0, 90.0, 2).as_slice(), + m.on_report("e1", 0, 90.0, 2, 0.0).as_slice(), [Action::Grant { .. }] )); // estimate 96 → Δ=4 ≤ 5 → alert. - let a = m.on_report("e1", 0, 96.0, 3); + let a = m.on_report("e1", 0, 96.0, 3, 0.0); match a.as_slice() { [Action::Alert { global_estimate, @@ -319,7 +390,7 @@ mod tests { other => panic!("expected alert, got {other:?}"), } // Further reports do not re-fire. - assert!(m.on_report("e1", 0, 200.0, 4).is_empty()); + assert!(m.on_report("e1", 0, 200.0, 4, 0.0).is_empty()); } #[test] @@ -329,9 +400,9 @@ mod tests { m.on_register("e2", 60_000, 0); // Both report modestly; estimate stays well below τ → only grants, never alert. for seq in 1..=5 { - let a = m.on_report("e1", 0, seq as f64 * 2.0, seq); + let a = m.on_report("e1", 0, seq as f64 * 2.0, seq, 0.0); assert!(a.iter().all(|x| matches!(x, Action::Grant { .. }))); - let b = m.on_report("e2", 0, seq as f64 * 2.0, seq); + let b = m.on_report("e2", 0, seq as f64 * 2.0, seq, 0.0); assert!(b.iter().all(|x| matches!(x, Action::Grant { .. }))); } assert!(m.global_estimate() < 100.0); @@ -344,7 +415,7 @@ mod tests { let a = m.on_register("e2", 60_000, 0); m.on_register("e3", 60_000, 0); // After 3 edges, a register re-grants ALL three; Δ=120, k=3 → slack=20. - let a3 = m.on_report("e1", 0, 0.0, 1); + let a3 = m.on_report("e1", 0, 0.0, 1, 0.0); assert_eq!(a3.len(), 3, "re-grant should reach all edges"); let _ = a; // Safety invariant: Σ slack = k·slack = Δ/2 at all times. @@ -358,10 +429,10 @@ mod tests { fn seq_dedup_ignores_retransmits() { let mut m = Monitor::new(cfg(100.0)); m.on_register("e1", 60_000, 0); - m.on_report("e1", 0, 40.0, 1); + m.on_report("e1", 0, 40.0, 1, 0.0); let before = m.global_estimate(); // Same seq re-delivered → ignored. - let a = m.on_report("e1", 0, 40.0, 1); + let a = m.on_report("e1", 0, 40.0, 1, 0.0); assert!(a.is_empty()); assert_eq!(m.global_estimate(), before); } @@ -372,7 +443,7 @@ mod tests { m.on_register("e1", 60_000, 120_000); // epoch starts at 120_000 assert_eq!(m.window_start_ms(), 120_000); // A report tagged with the previous epoch is dropped. - let a = m.on_report("e1", 60_000, 99.0, 1); + let a = m.on_report("e1", 60_000, 99.0, 1, 0.0); assert!(a.is_empty()); assert_eq!(m.global_estimate(), 0.0); } @@ -382,8 +453,8 @@ mod tests { let mut m = Monitor::new(cfg(100.0)); m.on_register("e1", 60_000, 0); m.on_register("e2", 60_000, 0); - m.on_report("e1", 0, 30.0, 1); - m.on_report("e2", 0, 20.0, 1); + m.on_report("e1", 0, 30.0, 1, 0.0); + m.on_report("e2", 0, 20.0, 1, 0.0); assert_eq!(m.global_estimate(), 50.0); // e2 leaves: its 20 stays in the estimate via departed_mass. m.on_leave("e2"); @@ -391,14 +462,53 @@ mod tests { assert_eq!(m.edge_count(), 1); } + #[test] + fn single_edge_grants_no_sampling() { + // With one edge there is no rate vector to coordinate over → p=1. + let mut m = Monitor::new(cfg(1000.0)); + m.on_register("e1", 60_000, 0); + let a = m.on_report("e1", 0, 10.0, 1, 50_000.0); + assert_eq!(sample_p_in(&a, "e1"), 1.0); + } + + #[test] + fn skewed_rates_yield_differentiated_sample_p_above_floor() { + // Two edges, skewed reported rates: the hot edge should be sampled + // harder (smaller p) than the quiet edge, and BOTH must sit at/above the + // ε-derived coupling floor. τ is large so the monitor stays in the grant + // (not alert) regime while we exercise the allocation. + let mut m = Monitor::new(cfg(1_000_000.0)); // epsilon 0.05 + m.on_register("e1", 60_000, 0); + m.on_register("e2", 60_000, 0); + // e1 hot (100k items/win), e2 quiet (1k/win). Small local values keep us + // far from τ so no alert fires. + m.on_report("e1", 0, 1.0, 1, 100_000.0); + let a = m.on_report("e2", 0, 1.0, 1, 1_000.0); + let p_hot = sample_p_in(&a, "e1"); + let p_quiet = sample_p_in(&a, "e2"); + assert!( + p_hot < p_quiet, + "hot edge p {p_hot} should be < quiet edge p {p_quiet}" + ); + // Both respect the ε-derived floor for their rate. + let eps = m.cfg.epsilon; + assert!(p_hot >= epsilon_sample_floor(eps, 100_000.0) - 1e-9); + assert!(p_quiet >= epsilon_sample_floor(eps, 1_000.0) - 1e-9); + assert!(p_hot > 0.0 && p_hot <= 1.0); + assert!(p_quiet > 0.0 && p_quiet <= 1.0); + // Slack countdown unaffected: both grants carry the same (positive) slack. + assert!(slack_in(&a, "e1") > 0.0); + assert_eq!(slack_in(&a, "e1"), slack_in(&a, "e2")); + } + #[test] fn epoch_advance_resets_estimate() { let mut m = Monitor::new(cfg(100.0)); m.on_register("e1", 60_000, 0); - m.on_report("e1", 0, 80.0, 1); + m.on_report("e1", 0, 80.0, 1, 0.0); assert_eq!(m.global_estimate(), 80.0); // A report for the next epoch advances and resets. - m.on_report("e1", 60_000, 5.0, 2); + m.on_report("e1", 60_000, 5.0, 2, 0.0); assert_eq!(m.window_start_ms(), 60_000); assert_eq!(m.global_estimate(), 5.0); } diff --git a/data_plane/src/monitor/mod.rs b/data_plane/src/monitor/mod.rs index 56ac3147..73cfed31 100644 --- a/data_plane/src/monitor/mod.rs +++ b/data_plane/src/monitor/mod.rs @@ -17,8 +17,10 @@ pub mod alert; pub mod coordinator; pub mod epoch; +pub mod sampling_alloc; pub mod server; pub use alert::{global_threshold_violation, AlertSink}; pub use coordinator::{Action, Monitor, MonitorConfig}; +pub use sampling_alloc::{allocate_sample_rates, epsilon_sample_floor, uniform_sample_rate}; pub use server::{MonitorCoordinator, MonitorServiceImpl}; diff --git a/data_plane/src/monitor/sampling_alloc.rs b/data_plane/src/monitor/sampling_alloc.rs new file mode 100644 index 00000000..3e8a8793 --- /dev/null +++ b/data_plane/src/monitor/sampling_alloc.rs @@ -0,0 +1,252 @@ +//! Distributed-NitroSketch coordinated update-sampling allocation — the +//! coordinator's per-edge sampling-probability `p_i` computation. Faithful Rust +//! port of the Go reference `asap-precompute-go/monitor/sampling_alloc.go` +//! (`AllocateSampleRates`), plus the CDM-threshold coupling floor from +//! `docs/distributed-nitrosketch-coordinated-sampling.md` +//! ("Combining with the CDM threshold"). +//! +//! Each edge reports its observed per-window `rate` (items/window) over the +//! reverse `MonitorService` channel (`MonitorReport.rate`); the coordinator +//! holds the full per-edge rate vector for a monitor and allocates a `p_i` that +//! MINIMIZES total edge update work `Σ rate_i·p_i` subject to a merged +//! sampling-variance budget `V ≥ Σ f_i·(1−p_i)/p_i`. The KKT solution is +//! +//! ```text +//! p_i = clamp( √λ · √(f_i / rate_i), 0, 1 ) +//! ``` +//! +//! i.e. sample harder (smaller `p`) on high-rate edges where local accuracy is +//! cheap, keep `p≈1` on low-rate edges. `√λ` is the single scale that makes the +//! variance constraint bind; solved by bisection (variance is monotone +//! decreasing in the scale). The result rides each edge's `SlackGrant.sample_p`. + +/// Compute per-edge update-sampling probabilities `p_i`. +/// +/// `rates[i]` = edge i's items/window; `freqs[i]` = edge i's mass of the queried +/// quantity (pass `rates` as a proxy when the per-key split is unknown — then +/// `p_i ∝ 1/√rate_i`). `var_budget` = `V` in the bound above. A `rate_i ≤ 0` or +/// `freqs_i ≤ 0` yields `p_i = 1` (nothing to sample). +pub fn allocate_sample_rates(rates: &[f64], freqs: &[f64], var_budget: f64) -> Vec { + let n = rates.len(); + let mut p = vec![0.0_f64; n]; + if n == 0 { + return p; + } + // base[i] = √(f_i/rate_i); 0 forces p_i = 1 (no rate ⇒ no sampling benefit). + let mut base = vec![0.0_f64; n]; + let mut max_base = 0.0_f64; + for i in 0..n { + if rates[i] <= 0.0 || freqs[i] <= 0.0 { + base[i] = 0.0; + continue; + } + base[i] = (freqs[i] / rates[i]).sqrt(); + if base[i] > max_base { + max_base = base[i]; + } + } + if max_base == 0.0 || var_budget <= 0.0 { + return vec![1.0; n]; + } + + // variance(scale) = Σ f_i·(1−p_i)/p_i with p_i = min(1, scale·base_i). + // Monotone decreasing in scale: scale→0 ⇒ p→0 ⇒ variance→∞; at + // scale = 1/min(positive base_i) all p_i hit 1 ⇒ variance = 0. + let variance = |scale: f64| -> f64 { + let mut v = 0.0; + for i in 0..n { + if base[i] == 0.0 { + continue; // p_i = 1, contributes 0 + } + let pi = scale * base[i]; + if pi >= 1.0 { + continue; + } + v += freqs[i] * (1.0 - pi) / pi; + } + v + }; + + // scale_hi: every positive-base edge clamped to p_i = 1 (variance 0 ≤ V). + let mut min_base = f64::INFINITY; + for &b in &base { + if b > 0.0 && b < min_base { + min_base = b; + } + } + let scale_hi = 1.0 / min_base; + if variance(scale_hi) >= var_budget { + // Even no sampling (all p=1) can't meet V → don't sample at all. + return vec![1.0; n]; + } + // Bisect for the smallest scale (= most sampling, least CPU) with + // variance(scale) ≤ var_budget. + let mut lo = 0.0_f64; + let mut hi = scale_hi; + for _ in 0..100 { + let mid = 0.5 * (lo + hi); + if mid <= 0.0 { + lo = mid; + continue; + } + if variance(mid) > var_budget { + lo = mid; // too much sampling (variance too high) → raise scale + } else { + hi = mid; + } + } + let scale = hi; + for i in 0..n { + if base[i] == 0.0 { + p[i] = 1.0; + } else { + p[i] = (scale * base[i]).min(1.0); + } + } + p +} + +/// The single uniform sampling probability `p` meeting the same variance budget +/// `V` with one rate everywhere (the per-edge-independent NitroSketch baseline): +/// `(1−p)/p·Σf = V ⇒ p = Σf/(Σf+V)`. Used to quantify the coordination win. +pub fn uniform_sample_rate(freqs: &[f64], var_budget: f64) -> f64 { + let sum_f: f64 = freqs.iter().filter(|&&f| f > 0.0).sum(); + if sum_f <= 0.0 || var_budget <= 0.0 { + return 1.0; + } + (sum_f / (sum_f + var_budget)).min(1.0) +} + +/// The CDM-threshold coupling floor on `p_i`: never sample so hard that the +/// sampling noise `ε_s = √((1−p)/(p·rate))` exceeds the CDM threshold tolerance +/// `ε` (the agg's epsilon). Solving `ε ≥ √((1−p)/(p·rate))` for `p`: +/// +/// ```text +/// ε²·p·rate ≥ 1 − p ⇒ p·(ε²·rate + 1) ≥ 1 ⇒ p ≥ 1/(1 + ε²·rate) +/// ``` +/// +/// With `rate ≤ 0` or `epsilon ≤ 0` the floor is 1.0 (no sampling permitted). The +/// floor is always in `(0,1]` and grows toward 1 as the edge's rate falls (a +/// low-`N` edge can't absorb sampling noise within the band). +pub fn epsilon_sample_floor(epsilon: f64, rate: f64) -> f64 { + if rate <= 0.0 || epsilon <= 0.0 { + return 1.0; + } + 1.0 / (1.0 + epsilon * epsilon * rate) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn merged_variance(freqs: &[f64], p: &[f64]) -> f64 { + let mut v = 0.0; + for i in 0..freqs.len() { + if p[i] > 0.0 && p[i] < 1.0 { + v += freqs[i] * (1.0 - p[i]) / p[i]; + } + } + v + } + fn total_cpu(rates: &[f64], p: &[f64]) -> f64 { + rates.iter().zip(p).map(|(r, p)| r * p).sum() + } + + /// Mirrors Go `TestAllocateSampleRates_SkewedFleetBeatsUniform`: on a skewed + /// fleet (one hot edge, a quiet tail) the coordinated p_i ∝ √(f_i/rate_i) + /// allocation hits the same merged-variance budget at strictly LESS total + /// edge update work than per-edge-uniform NitroSketch. + #[test] + fn skewed_fleet_beats_uniform() { + let rates = [100000.0, 1000.0, 1000.0, 1000.0, 1000.0]; // 1 hot + 4 quiet + let freqs = [100.0, 100.0, 100.0, 100.0, 100.0]; // key ~uniform across edges + let v = 2000.0; + + let p_coord = allocate_sample_rates(&rates, &freqs, v); + let p_uni = uniform_sample_rate(&freqs, v); + let uni = vec![p_uni; rates.len()]; + + let vc = merged_variance(&freqs, &p_coord); + let vu = merged_variance(&freqs, &uni); + assert!(vc <= v * 1.02, "coordinated variance {vc} exceeds budget {v}"); + assert!( + (vu - v).abs() <= v * 0.02, + "uniform variance {vu} should ≈ budget {v}" + ); + let cc = total_cpu(&rates, &p_coord); + let cu = total_cpu(&rates, &uni); + assert!( + cc < cu, + "coordinated CPU {cc} should be < uniform CPU {cu} at equal variance" + ); + // Hot edge sampled hard; quiet edges near p=1. + assert!( + p_coord[0] < p_coord[1], + "hot edge should get smaller p than quiet edges: {p_coord:?}" + ); + } + + /// Mirrors Go `TestAllocateSampleRates_FlatFleetNoWin`: on a uniform fleet + /// the coordinated allocation collapses to ≈ the uniform rate (no win). + #[test] + fn flat_fleet_no_win() { + let rates = [2000.0, 2000.0, 2000.0, 2000.0]; + let v = 3000.0; + let p_coord = allocate_sample_rates(&rates, &rates, v); + let p_uni = uniform_sample_rate(&rates, v); + let cc = total_cpu(&rates, &p_coord); + let cu = p_uni * (2000.0 * 4.0); + assert!( + (cc - cu).abs() <= 0.05 * cu, + "flat fleet: coordinated CPU {cc} should ≈ uniform {cu} (no win)" + ); + } + + #[test] + fn zero_or_negative_rate_gets_no_sampling() { + let rates = [0.0, -5.0, 1000.0]; + let freqs = [100.0, 100.0, 100.0]; + let p = allocate_sample_rates(&rates, &freqs, 50.0); + assert_eq!(p[0], 1.0); + assert_eq!(p[1], 1.0); + assert!(p[2] <= 1.0 && p[2] > 0.0); + } + + #[test] + fn empty_input() { + assert!(allocate_sample_rates(&[], &[], 100.0).is_empty()); + } + + #[test] + fn zero_budget_means_no_sampling() { + let rates = [1000.0, 2000.0]; + let p = allocate_sample_rates(&rates, &rates, 0.0); + assert_eq!(p, vec![1.0, 1.0]); + } + + #[test] + fn epsilon_floor_monotone_and_bounded() { + // Floor in (0,1], rises toward 1 as rate falls. + let eps = 0.05; + let hot = epsilon_sample_floor(eps, 100000.0); + let quiet = epsilon_sample_floor(eps, 1000.0); + assert!(hot > 0.0 && hot <= 1.0); + assert!(quiet > 0.0 && quiet <= 1.0); + assert!(quiet > hot, "lower-rate edge needs a higher floor"); + // Closed form: p = 1/(1 + ε²·rate). + assert!((hot - 1.0 / (1.0 + eps * eps * 100000.0)).abs() < 1e-12); + // Degenerate inputs ⇒ no sampling. + assert_eq!(epsilon_sample_floor(eps, 0.0), 1.0); + assert_eq!(epsilon_sample_floor(0.0, 1000.0), 1.0); + } + + #[test] + fn floor_enforced_keeps_sampling_noise_within_epsilon() { + // A floored p_i keeps ε_s = √((1−p)/(p·rate)) ≤ ε. + let eps = 0.05; + let rate = 100000.0; + let floor = epsilon_sample_floor(eps, rate); + let eps_s = ((1.0 - floor) / (floor * rate)).sqrt(); + assert!(eps_s <= eps + 1e-9, "ε_s {eps_s} should be ≤ ε {eps}"); + } +} diff --git a/data_plane/src/monitor/server.rs b/data_plane/src/monitor/server.rs index 2345f204..49e5872d 100644 --- a/data_plane/src/monitor/server.rs +++ b/data_plane/src/monitor/server.rs @@ -92,11 +92,12 @@ impl MonitorCoordinator { window_start_ms: u64, local_value: f64, seq: u64, + rate: f64, ) -> Option> { let mk = (agg_id, key); let mut monitors = self.monitors.lock().await; let mon = monitors.get_mut(&mk)?; - Some(mon.on_report(edge_id, window_start_ms, local_value, seq)) + Some(mon.on_report(edge_id, window_start_ms, local_value, seq, rate)) } /// Dispatch coordinator actions: grants to the addressed edge's stream, @@ -109,6 +110,7 @@ impl MonitorCoordinator { round, local_slack, window_start_ms, + sample_p, } => { let msg = CoordToEdge { msg: Some(coord_to_edge::Msg::Grant(SlackGrant { @@ -117,6 +119,7 @@ impl MonitorCoordinator { round, local_slack, window_start_ms, + sample_p, })), }; let tx = { @@ -186,6 +189,7 @@ impl MonitorCoordinator { rep.window_start_ms, rep.local_value, rep.seq, + rep.rate, ) .await {