From 17969946fa6c4f251aaf64ff3db522415b8e6a65 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 16 Jun 2026 12:25:08 -0600 Subject: [PATCH 1/3] =?UTF-8?q?feat(monitor):=20distributed=20L2=20/=20F2?= =?UTF-8?q?=20(=E2=80=96f=E2=80=96=E2=82=82=C2=B2)=20threshold=20monitorin?= =?UTF-8?q?g?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Real distributed second-frequency-moment monitor — the L2 counterpart to the L1 slack-countdown coordinator. The hard part is that F2 does NOT merge linearly: F2(Σ f_i) = Σ F2(f_i) + 2 Σ_{i --- data_plane/src/monitor/f2.rs | 359 ++++++++++++++++++++++++++++++++++ data_plane/src/monitor/mod.rs | 1 + 2 files changed, 360 insertions(+) create mode 100644 data_plane/src/monitor/f2.rs diff --git a/data_plane/src/monitor/f2.rs b/data_plane/src/monitor/f2.rs new file mode 100644 index 00000000..06e1799c --- /dev/null +++ b/data_plane/src/monitor/f2.rs @@ -0,0 +1,359 @@ +//! Distributed **L2 / second-frequency-moment (`F2 = ‖f‖₂²`) threshold monitoring**. +//! +//! The L1 coordinator (`coordinator.rs`) tracks an **additive** functional, so +//! the global value is `Σ known_value_i` — a linear merge. `F2` does **not** +//! merge linearly: for `f = Σ_i f_i`, +//! +//! ```text +//! F2(f) = Σ_x (Σ_i f_i(x))² +//! = Σ_i F2(f_i) + 2 Σ_{i u64 { + x ^= x >> 30; + x = x.wrapping_mul(0xBF58476D1CE4E5B9); + x ^= x >> 27; + x = x.wrapping_mul(0x94D049BB133111EB); + x ^= x >> 31; + x +} + +/// `g_i(key) ∈ {+1, −1}` for estimator `i` under `seed` — pairwise-independent +/// enough across (i, key) for `E[g_i(x)g_i(y)] = 0` (x≠y), which is what makes +/// `E[Z_i²] = F2` unbiased. +#[inline] +fn sign(seed: u64, idx: usize, key: u64) -> i64 { + let h = mix64( + seed ^ (idx as u64).wrapping_mul(0x9E3779B97F4A7C15) ^ mix64(key), + ); + if h & 1 == 0 { + 1 + } else { + -1 + } +} + +/// AMS tug-of-war F2 sketch: `s1·s2` linear estimators `Z[i] = Σ_x g_i(x)·f(x)`. +/// Linear in updates ⇒ mergeable; `mean_{group}(Z²)` estimates F2, `median` +/// across groups bounds failure probability. +#[derive(Clone, Debug)] +pub struct AmsF2Sketch { + s1: usize, // averaging width per group (~1/ε²) + s2: usize, // median depth (# groups) (~log 1/δ) + seed: u64, + z: Vec, // length s1*s2 +} + +impl AmsF2Sketch { + pub fn new(s1: usize, s2: usize, seed: u64) -> Self { + assert!(s1 > 0 && s2 > 0, "AMS dimensions must be positive"); + Self { + s1, + s2, + seed, + z: vec![0; s1 * s2], + } + } + + pub fn dims(&self) -> (usize, usize, u64) { + (self.s1, self.s2, self.seed) + } + + /// Add `delta` occurrences of `key` (delta may be negative). + pub fn update(&mut self, key: u64, delta: i64) { + for i in 0..self.z.len() { + self.z[i] += sign(self.seed, i, key) * delta; + } + } + + /// Linearly merge another sketch (must share dimensions + seed) into this one. + /// This is the step that makes distributed F2 correct: `Z_merged = Σ Z_i`, + /// and the later square recovers the cross terms. + pub fn merge(&mut self, other: &AmsF2Sketch) { + assert_eq!( + (self.s1, self.s2, self.seed), + (other.s1, other.s2, other.seed), + "cannot merge AMS sketches with different (s1, s2, seed)" + ); + for i in 0..self.z.len() { + self.z[i] += other.z[i]; + } + } + + /// Unbiased F2 estimate: median over the `s2` groups of the per-group mean of + /// `Z²` (each `Z²` is an unbiased F2 estimator; the mean shrinks variance). + pub fn estimate_f2(&self) -> f64 { + let mut group_means: Vec = Vec::with_capacity(self.s2); + for g in 0..self.s2 { + let mut sum = 0.0; + for j in 0..self.s1 { + let z = self.z[g * self.s1 + j] as f64; + sum += z * z; + } + group_means.push(sum / self.s1 as f64); + } + median(&mut group_means) + } +} + +fn median(v: &mut [f64]) -> f64 { + if v.is_empty() { + return 0.0; + } + v.sort_by(|a, b| a.partial_cmp(b).unwrap()); + let n = v.len(); + if n % 2 == 1 { + v[n / 2] + } else { + 0.5 * (v[n / 2 - 1] + v[n / 2]) + } +} + +/// Result of an L2 monitor report ingestion. +#[derive(Clone, Debug, PartialEq)] +pub enum F2Action { + /// `F2̂` is comfortably below the threshold — nothing to do. + Ok { f2_estimate: f64 }, + /// `F2̂ ≥ (1−ε)·τ`: the monitored second moment crossed the threshold band. + Alert { f2_estimate: f64, tau: f64 }, +} + +/// Distributed F2 threshold monitor: holds each edge's latest AMS sketch, merges +/// them on every report, estimates the global `F2`, and fires once per epoch +/// when `F2̂ ≥ (1−ε)·τ`. `τ` is a threshold on `F2 = ‖f‖₂²`. +pub struct DistributedF2Monitor { + tau: f64, + epsilon: f64, + s1: usize, + s2: usize, + seed: u64, + window_start_ms: u64, + edges: HashMap, + alerted: bool, +} + +impl DistributedF2Monitor { + pub fn new(tau: f64, epsilon: f64, s1: usize, s2: usize, seed: u64) -> Self { + Self { + tau, + epsilon, + s1, + s2, + seed, + window_start_ms: 0, + edges: HashMap::new(), + alerted: false, + } + } + + pub fn edge_count(&self) -> usize { + self.edges.len() + } + + /// A fresh, correctly-dimensioned sketch an edge (or a test) can fill and + /// report — guarantees the (s1, s2, seed) match so merges never panic. + pub fn new_edge_sketch(&self) -> AmsF2Sketch { + AmsF2Sketch::new(self.s1, self.s2, self.seed) + } + + fn ensure_epoch(&mut self, window_start_ms: u64) -> bool { + if window_start_ms == self.window_start_ms { + return true; + } + if window_start_ms > self.window_start_ms { + self.window_start_ms = window_start_ms; + self.edges.clear(); + self.alerted = false; + return true; + } + false // stale epoch + } + + /// Ingest one edge's AMS sketch for `window_start_ms`. Replaces that edge's + /// prior sketch (the edge reports its running window sketch), re-estimates + /// the merged global F2, and returns an Alert iff it crossed `(1−ε)τ`. + pub fn on_report( + &mut self, + edge_id: &str, + window_start_ms: u64, + sketch: AmsF2Sketch, + ) -> Option { + if sketch.dims() != (self.s1, self.s2, self.seed) { + return None; // mis-dimensioned report — drop (caller should log) + } + if !self.ensure_epoch(window_start_ms) { + return None; // stale epoch + } + self.edges.insert(edge_id.to_string(), sketch); + let f2 = self.global_f2(); + if !self.alerted && f2 >= (1.0 - self.epsilon) * self.tau { + self.alerted = true; + return Some(F2Action::Alert { + f2_estimate: f2, + tau: self.tau, + }); + } + Some(F2Action::Ok { f2_estimate: f2 }) + } + + /// Current global `F2̂` = estimate from the linear merge of all edge sketches. + /// The merge is what captures the cross terms `2⟨f_i,f_j⟩` a per-edge + /// `Σ F2(f_i)` would miss. + pub fn global_f2(&self) -> f64 { + let mut merged = AmsF2Sketch::new(self.s1, self.s2, self.seed); + for s in self.edges.values() { + merged.merge(s); + } + merged.estimate_f2() + } + + /// L2 variance budget for the coordinated-sampling allocation: + /// `V = (ε·‖f‖₂)² = ε²·F2`. Feed this (with the SAME per-key `f_i`/`rate_i` + /// as L1) into `sampling_alloc::allocate_sample_rates` to size `p_i` for an + /// L2-monitored quantity. Unlike L1's `(ε·τ)²` this tracks the LIVE L2 mass. + pub fn sampling_var_budget(&self) -> f64 { + let eps = self.epsilon; + eps * eps * self.global_f2() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + // Exact F2 of a sparse vector given as (key, freq) pairs. + fn true_f2(v: &[(u64, i64)]) -> f64 { + v.iter().map(|&(_, f)| (f as f64) * (f as f64)).sum() + } + + fn fill(sk: &mut AmsF2Sketch, v: &[(u64, i64)]) { + for &(k, f) in v { + sk.update(k, f); + } + } + + #[test] + fn ams_estimates_f2_within_tolerance() { + // 200 keys with freq = key ⇒ F2 = Σ k² for k=1..200. + let v: Vec<(u64, i64)> = (1..=200u64).map(|k| (k, k as i64)).collect(); + let exact = true_f2(&v); + let mut sk = AmsF2Sketch::new(1024, 9, 0xABCD_1234); + fill(&mut sk, &v); + let est = sk.estimate_f2(); + let rel = (est - exact).abs() / exact; + assert!(rel < 0.12, "F2 est {est} vs exact {exact} (rel {rel})"); + } + + #[test] + fn merge_captures_cross_terms() { + // a = {1:10, 2:10}; b = {1:10, 3:10}. They SHARE key 1. + // F2(a) = 200, F2(b) = 200, Σ local = 400. + // merged f = {1:20, 2:10, 3:10} ⇒ F2 = 400 + 100 + 100 = 600. + // A linear-merge sketch must estimate ~600 (NOT 400) — the cross term + // 2·⟨a,b⟩ = 2·(10·10) = 200 is exactly the gap. This is the whole point. + let a = [(1u64, 10i64), (2, 10)]; + let b = [(1u64, 10i64), (3, 10)]; + let sum_local = true_f2(&a) + true_f2(&b); // 400 + let merged_exact = 600.0; + let seed = 0x5151_2727; + let mut sa = AmsF2Sketch::new(1024, 9, seed); + let mut sb = AmsF2Sketch::new(1024, 9, seed); + fill(&mut sa, &a); + fill(&mut sb, &b); + sa.merge(&sb); + let est = sa.estimate_f2(); + assert!( + (est - merged_exact).abs() / merged_exact < 0.15, + "merged F2 est {est} should track {merged_exact} (cross terms), not {sum_local}" + ); + assert!( + est > 0.5 * (sum_local + merged_exact), + "est {est} must clearly exceed the cross-term-free sum {sum_local}" + ); + } + + #[test] + fn monitor_alert_threshold_behaviour() { + let seed = 0x2468_1357; + let eps = 0.1; + // Build the merged-F2 ≈ 600 scenario across two edges. + let edge_a = [(1u64, 10i64), (2, 10)]; + let edge_b = [(1u64, 10i64), (3, 10)]; + + // tau well ABOVE 600 ⇒ no alert. + let mut hi = DistributedF2Monitor::new(2_000.0, eps, 1024, 9, seed); + let mut a1 = hi.new_edge_sketch(); + let mut b1 = hi.new_edge_sketch(); + fill(&mut a1, &edge_a); + fill(&mut b1, &edge_b); + hi.on_report("a", 0, a1); + let r_hi = hi.on_report("b", 0, b1).unwrap(); + assert!(matches!(r_hi, F2Action::Ok { .. }), "below tau ⇒ Ok, got {r_hi:?}"); + + // tau BELOW 600 ⇒ alert once. + let mut lo = DistributedF2Monitor::new(400.0, eps, 1024, 9, seed); + let mut a2 = lo.new_edge_sketch(); + let mut b2 = lo.new_edge_sketch(); + fill(&mut a2, &edge_a); + fill(&mut b2, &edge_b); + lo.on_report("a", 0, a2); + let r_lo = lo.on_report("b", 0, b2).unwrap(); + assert!(matches!(r_lo, F2Action::Alert { .. }), "above tau ⇒ Alert, got {r_lo:?}"); + } + + #[test] + fn sampling_var_budget_is_eps2_times_f2() { + let seed = 0x1357_2468; + let eps = 0.2; + let mut mon = DistributedF2Monitor::new(10_000.0, eps, 1024, 9, seed); + let mut e = mon.new_edge_sketch(); + fill(&mut e, &(1..=100u64).map(|k| (k, k as i64)).collect::>()); + mon.on_report("e", 0, e); + let f2 = mon.global_f2(); + let v = mon.sampling_var_budget(); + assert!((v - eps * eps * f2).abs() < 1e-6, "var_budget {v} != eps^2*F2 {}", eps * eps * f2); + assert!(v > 0.0); + } + + #[test] + fn epoch_advance_clears_edges() { + let mut mon = DistributedF2Monitor::new(1_000.0, 0.1, 256, 5, 7); + let mut e = mon.new_edge_sketch(); + e.update(1, 5); + mon.on_report("e", 0, e); + assert_eq!(mon.edge_count(), 1); + // New epoch resets membership. + let e2 = mon.new_edge_sketch(); + mon.on_report("e", 60_000, e2); + assert_eq!(mon.edge_count(), 1); + // A stale-epoch report is dropped. + let e3 = mon.new_edge_sketch(); + assert_eq!(mon.on_report("e", 0, e3), None); + } +} diff --git a/data_plane/src/monitor/mod.rs b/data_plane/src/monitor/mod.rs index 73cfed31..4196b44e 100644 --- a/data_plane/src/monitor/mod.rs +++ b/data_plane/src/monitor/mod.rs @@ -17,6 +17,7 @@ pub mod alert; pub mod coordinator; pub mod epoch; +pub mod f2; pub mod sampling_alloc; pub mod server; From 49e3d489902c1bb0301911e075405cb6c633711a Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 16 Jun 2026 13:06:34 -0600 Subject: [PATCH 2/3] refactor(monitor/f2): Count-Sketch instead of AMS + L2 allocate_p MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Switch the distributed-F2 estimator from AMS tug-of-war to a Count-Sketch (d rows × w buckets, ±1 sign), matching ASAP's warm-tier sketch family and the UnivMon/Cormode-Garofalakis line. Same linear-merge-captures-cross-terms property (C_merged = Σ C_i, F2̂ = median_r Σ_b C[r][b]²). Hashing = murmur3 fmix64 with a different seed per row and a different salt for bucket vs sign (engineering approximation, like xxhash/murmur — not provably 4-wise independent). DistributedF2Monitor now also does the L2 coordinated-sampling allocation: allocate_p() with freq_i = each edge's LOCAL F2̂_i (from its own sketch), rate_i = reported rate, var_budget = ε²·F2̂_global — so p_i ∝ √(F2̂_i/rate_i) (an edge carrying more L2 mass keeps a higher p). Mirrors the L1 coordinator::allocate_p with the L2 freq/budget. Division of labour: EDGE maintains its local Count-Sketch (shared seeds) + rate and ships (sketch, rate) per window; COORDINATOR merges linearly, estimates F2̂, fires at (1−ε)τ, and allocates p_i. 6 unit tests incl. cross-term merge and equal-rate-different-F2 allocation. Co-Authored-By: Claude Opus 4.8 --- data_plane/src/monitor/f2.rs | 347 ++++++++++++++++++++--------------- 1 file changed, 194 insertions(+), 153 deletions(-) diff --git a/data_plane/src/monitor/f2.rs b/data_plane/src/monitor/f2.rs index 06e1799c..c5d114fc 100644 --- a/data_plane/src/monitor/f2.rs +++ b/data_plane/src/monitor/f2.rs @@ -1,124 +1,130 @@ -//! Distributed **L2 / second-frequency-moment (`F2 = ‖f‖₂²`) threshold monitoring**. -//! -//! The L1 coordinator (`coordinator.rs`) tracks an **additive** functional, so -//! the global value is `Σ known_value_i` — a linear merge. `F2` does **not** -//! merge linearly: for `f = Σ_i f_i`, +//! Distributed **L2 / second-frequency-moment (`F2 = ‖f‖₂²`) threshold monitoring** +//! over linearly-mergeable **Count-Sketch**es. //! +//! `F2` does **not** merge linearly across edges — for `f = Σ_i f_i`, //! ```text -//! F2(f) = Σ_x (Σ_i f_i(x))² -//! = Σ_i F2(f_i) + 2 Σ_{i u64 { + k ^= k >> 33; + k = k.wrapping_mul(0xff51afd7ed558ccd); + k ^= k >> 33; + k = k.wrapping_mul(0xc4ceb9fe1a85ec53); + k ^= k >> 33; + k +} + +const SALT_BUCKET: u64 = 0x9E37_79B9_7F4A_7C15; +const SALT_SIGN: u64 = 0xC2B2_AE3D_27D4_EB4F; + +/// Per-row bucket index `h_r(key) ∈ [0, w)`. #[inline] -fn mix64(mut x: u64) -> u64 { - x ^= x >> 30; - x = x.wrapping_mul(0xBF58476D1CE4E5B9); - x ^= x >> 27; - x = x.wrapping_mul(0x94D049BB133111EB); - x ^= x >> 31; - x +fn bucket(seed: u64, row: usize, key: u64, w: usize) -> usize { + let row_seed = fmix64(seed ^ SALT_BUCKET ^ (row as u64).wrapping_mul(SALT_BUCKET)); + (fmix64(row_seed ^ key) % w as u64) as usize } -/// `g_i(key) ∈ {+1, −1}` for estimator `i` under `seed` — pairwise-independent -/// enough across (i, key) for `E[g_i(x)g_i(y)] = 0` (x≠y), which is what makes -/// `E[Z_i²] = F2` unbiased. +/// Per-row sign `s_r(key) ∈ {+1, −1}` (independent salt from the bucket hash). #[inline] -fn sign(seed: u64, idx: usize, key: u64) -> i64 { - let h = mix64( - seed ^ (idx as u64).wrapping_mul(0x9E3779B97F4A7C15) ^ mix64(key), - ); - if h & 1 == 0 { +fn sign(seed: u64, row: usize, key: u64) -> i64 { + let row_seed = fmix64(seed ^ SALT_SIGN ^ (row as u64).wrapping_mul(SALT_SIGN)); + if fmix64(row_seed ^ key) & 1 == 0 { 1 } else { -1 } } -/// AMS tug-of-war F2 sketch: `s1·s2` linear estimators `Z[i] = Σ_x g_i(x)·f(x)`. -/// Linear in updates ⇒ mergeable; `mean_{group}(Z²)` estimates F2, `median` -/// across groups bounds failure probability. +/// Count-Sketch: `d` rows × `w` buckets, signed. Linear in updates ⇒ mergeable; +/// `median_r Σ_b C[r][b]²` estimates `F2 = ‖f‖₂²`. #[derive(Clone, Debug)] -pub struct AmsF2Sketch { - s1: usize, // averaging width per group (~1/ε²) - s2: usize, // median depth (# groups) (~log 1/δ) +pub struct CountSketchF2 { + d: usize, // depth (rows / median groups) ~ log(1/δ) + w: usize, // width (buckets per row) ~ 1/ε² seed: u64, - z: Vec, // length s1*s2 + c: Vec, // length d*w, row-major } -impl AmsF2Sketch { - pub fn new(s1: usize, s2: usize, seed: u64) -> Self { - assert!(s1 > 0 && s2 > 0, "AMS dimensions must be positive"); +impl CountSketchF2 { + pub fn new(d: usize, w: usize, seed: u64) -> Self { + assert!(d > 0 && w > 0, "Count-Sketch dimensions must be positive"); Self { - s1, - s2, + d, + w, seed, - z: vec![0; s1 * s2], + c: vec![0; d * w], } } pub fn dims(&self) -> (usize, usize, u64) { - (self.s1, self.s2, self.seed) + (self.d, self.w, self.seed) } - /// Add `delta` occurrences of `key` (delta may be negative). + /// Add `delta` occurrences of `key` (delta may be negative / fractional via + /// pre-scaled integer counts). pub fn update(&mut self, key: u64, delta: i64) { - for i in 0..self.z.len() { - self.z[i] += sign(self.seed, i, key) * delta; + for r in 0..self.d { + let b = bucket(self.seed, r, key, self.w); + self.c[r * self.w + b] += sign(self.seed, r, key) * delta; } } - /// Linearly merge another sketch (must share dimensions + seed) into this one. - /// This is the step that makes distributed F2 correct: `Z_merged = Σ Z_i`, - /// and the later square recovers the cross terms. - pub fn merge(&mut self, other: &AmsF2Sketch) { + /// Linearly merge another sketch (must share `(d, w, seed)`). THIS is what + /// makes distributed F2 correct: `C_merged = Σ_i C_i`, and the later square + /// recovers the cross terms a per-edge `Σ F2(f_i)` would miss. + pub fn merge(&mut self, other: &CountSketchF2) { assert_eq!( - (self.s1, self.s2, self.seed), - (other.s1, other.s2, other.seed), - "cannot merge AMS sketches with different (s1, s2, seed)" + (self.d, self.w, self.seed), + (other.d, other.w, other.seed), + "cannot merge Count-Sketches with different (d, w, seed)" ); - for i in 0..self.z.len() { - self.z[i] += other.z[i]; + for i in 0..self.c.len() { + self.c[i] += other.c[i]; } } - /// Unbiased F2 estimate: median over the `s2` groups of the per-group mean of - /// `Z²` (each `Z²` is an unbiased F2 estimator; the mean shrinks variance). + /// Unbiased F2 estimate: median over the `d` rows of `Σ_b C[r][b]²`. pub fn estimate_f2(&self) -> f64 { - let mut group_means: Vec = Vec::with_capacity(self.s2); - for g in 0..self.s2 { - let mut sum = 0.0; - for j in 0..self.s1 { - let z = self.z[g * self.s1 + j] as f64; - sum += z * z; + let mut row_f2: Vec = Vec::with_capacity(self.d); + for r in 0..self.d { + let mut s = 0.0f64; + for b in 0..self.w { + let v = self.c[r * self.w + b] as f64; + s += v * v; } - group_means.push(sum / self.s1 as f64); + row_f2.push(s); } - median(&mut group_means) + median(&mut row_f2) } } @@ -135,36 +141,40 @@ fn median(v: &mut [f64]) -> f64 { } } -/// Result of an L2 monitor report ingestion. +/// Result of ingesting one edge's Count-Sketch report. #[derive(Clone, Debug, PartialEq)] pub enum F2Action { - /// `F2̂` is comfortably below the threshold — nothing to do. Ok { f2_estimate: f64 }, - /// `F2̂ ≥ (1−ε)·τ`: the monitored second moment crossed the threshold band. Alert { f2_estimate: f64, tau: f64 }, } -/// Distributed F2 threshold monitor: holds each edge's latest AMS sketch, merges -/// them on every report, estimates the global `F2`, and fires once per epoch -/// when `F2̂ ≥ (1−ε)·τ`. `τ` is a threshold on `F2 = ‖f‖₂²`. +#[derive(Clone, Debug)] +struct EdgeF2 { + sketch: CountSketchF2, + rate: f64, +} + +/// Distributed F2 threshold monitor + coordinated-sampling allocator. Holds each +/// edge's latest Count-Sketch + rate, merges on report, estimates the global +/// `F2̂`, fires at `F2̂ ≥ (1−ε)τ`, and allocates `p_i ∝ √(F2̂_i / rate_i)`. pub struct DistributedF2Monitor { tau: f64, epsilon: f64, - s1: usize, - s2: usize, + d: usize, + w: usize, seed: u64, window_start_ms: u64, - edges: HashMap, + edges: HashMap, alerted: bool, } impl DistributedF2Monitor { - pub fn new(tau: f64, epsilon: f64, s1: usize, s2: usize, seed: u64) -> Self { + pub fn new(tau: f64, epsilon: f64, d: usize, w: usize, seed: u64) -> Self { Self { tau, epsilon, - s1, - s2, + d, + w, seed, window_start_ms: 0, edges: HashMap::new(), @@ -176,10 +186,10 @@ impl DistributedF2Monitor { self.edges.len() } - /// A fresh, correctly-dimensioned sketch an edge (or a test) can fill and - /// report — guarantees the (s1, s2, seed) match so merges never panic. - pub fn new_edge_sketch(&self) -> AmsF2Sketch { - AmsF2Sketch::new(self.s1, self.s2, self.seed) + /// A correctly-dimensioned empty sketch for an edge to fill (guarantees the + /// (d, w, seed) match so merges never panic). + pub fn new_edge_sketch(&self) -> CountSketchF2 { + CountSketchF2::new(self.d, self.w, self.seed) } fn ensure_epoch(&mut self, window_start_ms: u64) -> bool { @@ -192,25 +202,26 @@ impl DistributedF2Monitor { self.alerted = false; return true; } - false // stale epoch + false } - /// Ingest one edge's AMS sketch for `window_start_ms`. Replaces that edge's - /// prior sketch (the edge reports its running window sketch), re-estimates - /// the merged global F2, and returns an Alert iff it crossed `(1−ε)τ`. + /// Ingest one edge's `(Count-Sketch, rate)` for `window_start_ms`. Replaces + /// that edge's prior report, re-estimates the merged global F2, and returns + /// an Alert iff it crossed `(1−ε)τ`. pub fn on_report( &mut self, edge_id: &str, window_start_ms: u64, - sketch: AmsF2Sketch, + sketch: CountSketchF2, + rate: f64, ) -> Option { - if sketch.dims() != (self.s1, self.s2, self.seed) { - return None; // mis-dimensioned report — drop (caller should log) + if sketch.dims() != (self.d, self.w, self.seed) { + return None; // mis-dimensioned — drop } if !self.ensure_epoch(window_start_ms) { return None; // stale epoch } - self.edges.insert(edge_id.to_string(), sketch); + self.edges.insert(edge_id.to_string(), EdgeF2 { sketch, rate }); let f2 = self.global_f2(); if !self.alerted && f2 >= (1.0 - self.epsilon) * self.tau { self.alerted = true; @@ -222,24 +233,46 @@ impl DistributedF2Monitor { Some(F2Action::Ok { f2_estimate: f2 }) } - /// Current global `F2̂` = estimate from the linear merge of all edge sketches. - /// The merge is what captures the cross terms `2⟨f_i,f_j⟩` a per-edge - /// `Σ F2(f_i)` would miss. + /// Global `F2̂` from the linear merge of all edge Count-Sketches (captures the + /// cross terms a per-edge `Σ F2(f_i)` would miss). pub fn global_f2(&self) -> f64 { - let mut merged = AmsF2Sketch::new(self.s1, self.s2, self.seed); - for s in self.edges.values() { - merged.merge(s); + let mut merged = CountSketchF2::new(self.d, self.w, self.seed); + for e in self.edges.values() { + merged.merge(&e.sketch); } merged.estimate_f2() } - /// L2 variance budget for the coordinated-sampling allocation: - /// `V = (ε·‖f‖₂)² = ε²·F2`. Feed this (with the SAME per-key `f_i`/`rate_i` - /// as L1) into `sampling_alloc::allocate_sample_rates` to size `p_i` for an - /// L2-monitored quantity. Unlike L1's `(ε·τ)²` this tracks the LIVE L2 mass. + /// L2 sampling variance budget `V = (ε·‖f‖₂)² = ε²·F2̂` (vs L1's `(ε·τ)²`). pub fn sampling_var_budget(&self) -> f64 { - let eps = self.epsilon; - eps * eps * self.global_f2() + self.epsilon * self.epsilon * self.global_f2() + } + + /// Coordinated per-edge sampling probability `p_i ∝ √(F2̂_i / rate_i)`: the + /// freq weight is each edge's LOCAL F2 (from its own sketch), the rate is its + /// reported update rate, the budget is `ε²·F2̂_global`. An edge contributing + /// more L2 mass is sampled less (higher p) to preserve the F2 estimate. + /// Mirrors the L1 `coordinator::allocate_p` but with L2 freq/budget. + pub fn allocate_p(&self) -> HashMap { + let ids: Vec<&String> = self.edges.keys().collect(); + let rates: Vec = ids.iter().map(|id| self.edges[*id].rate).collect(); + let freqs: Vec = ids + .iter() + .map(|id| self.edges[*id].sketch.estimate_f2()) // local F2̂_i + .collect(); + let var_budget = self.sampling_var_budget(); + let mut p = allocate_sample_rates(&rates, &freqs, var_budget); + for (i, &rate) in rates.iter().enumerate() { + let floor = epsilon_sample_floor(self.epsilon, rate); + if p[i] < floor { + p[i] = floor; + } + } + ids.into_iter() + .cloned() + .zip(p) + .map(|(id, pi)| (id, if pi > 0.0 && pi <= 1.0 { pi } else { 1.0 })) + .collect() } } @@ -247,23 +280,20 @@ impl DistributedF2Monitor { mod tests { use super::*; - // Exact F2 of a sparse vector given as (key, freq) pairs. fn true_f2(v: &[(u64, i64)]) -> f64 { v.iter().map(|&(_, f)| (f as f64) * (f as f64)).sum() } - - fn fill(sk: &mut AmsF2Sketch, v: &[(u64, i64)]) { + fn fill(sk: &mut CountSketchF2, v: &[(u64, i64)]) { for &(k, f) in v { sk.update(k, f); } } #[test] - fn ams_estimates_f2_within_tolerance() { - // 200 keys with freq = key ⇒ F2 = Σ k² for k=1..200. + fn count_sketch_estimates_f2_within_tolerance() { let v: Vec<(u64, i64)> = (1..=200u64).map(|k| (k, k as i64)).collect(); let exact = true_f2(&v); - let mut sk = AmsF2Sketch::new(1024, 9, 0xABCD_1234); + let mut sk = CountSketchF2::new(9, 4096, 0xABCD_1234); fill(&mut sk, &v); let est = sk.estimate_f2(); let rel = (est - exact).abs() / exact; @@ -272,88 +302,99 @@ mod tests { #[test] fn merge_captures_cross_terms() { - // a = {1:10, 2:10}; b = {1:10, 3:10}. They SHARE key 1. - // F2(a) = 200, F2(b) = 200, Σ local = 400. - // merged f = {1:20, 2:10, 3:10} ⇒ F2 = 400 + 100 + 100 = 600. - // A linear-merge sketch must estimate ~600 (NOT 400) — the cross term - // 2·⟨a,b⟩ = 2·(10·10) = 200 is exactly the gap. This is the whole point. + // a={1:10,2:10}, b={1:10,3:10} share key 1. Σ local F2 = 400, but merged + // f={1:20,2:10,3:10} ⇒ F2 = 600. The linear merge must estimate ~600. let a = [(1u64, 10i64), (2, 10)]; let b = [(1u64, 10i64), (3, 10)]; let sum_local = true_f2(&a) + true_f2(&b); // 400 let merged_exact = 600.0; let seed = 0x5151_2727; - let mut sa = AmsF2Sketch::new(1024, 9, seed); - let mut sb = AmsF2Sketch::new(1024, 9, seed); + let mut sa = CountSketchF2::new(9, 4096, seed); + let mut sb = CountSketchF2::new(9, 4096, seed); fill(&mut sa, &a); fill(&mut sb, &b); sa.merge(&sb); let est = sa.estimate_f2(); assert!( - (est - merged_exact).abs() / merged_exact < 0.15, - "merged F2 est {est} should track {merged_exact} (cross terms), not {sum_local}" - ); - assert!( - est > 0.5 * (sum_local + merged_exact), - "est {est} must clearly exceed the cross-term-free sum {sum_local}" + (est - merged_exact).abs() / merged_exact < 0.1, + "merged F2 {est} should track {merged_exact} (cross terms), not {sum_local}" ); + assert!(est > 0.5 * (sum_local + merged_exact)); } #[test] fn monitor_alert_threshold_behaviour() { let seed = 0x2468_1357; let eps = 0.1; - // Build the merged-F2 ≈ 600 scenario across two edges. let edge_a = [(1u64, 10i64), (2, 10)]; - let edge_b = [(1u64, 10i64), (3, 10)]; + let edge_b = [(1u64, 10i64), (3, 10)]; // merged F2 ≈ 600 - // tau well ABOVE 600 ⇒ no alert. - let mut hi = DistributedF2Monitor::new(2_000.0, eps, 1024, 9, seed); + let mut hi = DistributedF2Monitor::new(2_000.0, eps, 9, 4096, seed); let mut a1 = hi.new_edge_sketch(); let mut b1 = hi.new_edge_sketch(); fill(&mut a1, &edge_a); fill(&mut b1, &edge_b); - hi.on_report("a", 0, a1); - let r_hi = hi.on_report("b", 0, b1).unwrap(); + hi.on_report("a", 0, a1, 1000.0); + let r_hi = hi.on_report("b", 0, b1, 1000.0).unwrap(); assert!(matches!(r_hi, F2Action::Ok { .. }), "below tau ⇒ Ok, got {r_hi:?}"); - // tau BELOW 600 ⇒ alert once. - let mut lo = DistributedF2Monitor::new(400.0, eps, 1024, 9, seed); + let mut lo = DistributedF2Monitor::new(400.0, eps, 9, 4096, seed); let mut a2 = lo.new_edge_sketch(); let mut b2 = lo.new_edge_sketch(); fill(&mut a2, &edge_a); fill(&mut b2, &edge_b); - lo.on_report("a", 0, a2); - let r_lo = lo.on_report("b", 0, b2).unwrap(); + lo.on_report("a", 0, a2, 1000.0); + let r_lo = lo.on_report("b", 0, b2, 1000.0).unwrap(); assert!(matches!(r_lo, F2Action::Alert { .. }), "above tau ⇒ Alert, got {r_lo:?}"); } #[test] - fn sampling_var_budget_is_eps2_times_f2() { + fn allocate_p_higher_local_f2_keeps_higher_p_at_equal_rate() { + // Equal rate; edge A carries far more L2 mass than edge B (disjoint keys). + // p_i ∝ √(F2_i/rate) ⇒ A (high F2) keeps a HIGHER p (sampled less) to + // preserve the F2 estimate. High rate ⇒ low floor so the allocation, not + // the floor, drives the spread. let seed = 0x1357_2468; + let mut mon = DistributedF2Monitor::new(1.0e12, 0.1, 9, 4096, seed); + let mut a = mon.new_edge_sketch(); + let mut b = mon.new_edge_sketch(); + fill(&mut a, &[(1u64, 100i64)]); // local F2 ≈ 10000 + fill(&mut b, &[(2u64, 1i64)]); // local F2 ≈ 1 + mon.on_report("a", 0, a, 1_000_000.0); + mon.on_report("b", 0, b, 1_000_000.0); + let p = mon.allocate_p(); + let pa = p["a"]; + let pb = p["b"]; + let floor = epsilon_sample_floor(0.1, 1_000_000.0); + assert!(pa > pb, "high-F2 edge p {pa} should exceed low-F2 p {pb}"); + assert!(pa > floor + 1e-9 && pb > floor + 1e-9, "p ({pa},{pb}) above floor {floor}"); + } + + #[test] + fn sampling_var_budget_is_eps2_times_f2() { + let seed = 0x9999_1111; let eps = 0.2; - let mut mon = DistributedF2Monitor::new(10_000.0, eps, 1024, 9, seed); + let mut mon = DistributedF2Monitor::new(1.0e12, eps, 9, 4096, seed); let mut e = mon.new_edge_sketch(); fill(&mut e, &(1..=100u64).map(|k| (k, k as i64)).collect::>()); - mon.on_report("e", 0, e); + mon.on_report("e", 0, e, 1000.0); let f2 = mon.global_f2(); let v = mon.sampling_var_budget(); - assert!((v - eps * eps * f2).abs() < 1e-6, "var_budget {v} != eps^2*F2 {}", eps * eps * f2); + assert!((v - eps * eps * f2).abs() < 1e-6); assert!(v > 0.0); } #[test] fn epoch_advance_clears_edges() { - let mut mon = DistributedF2Monitor::new(1_000.0, 0.1, 256, 5, 7); + let mut mon = DistributedF2Monitor::new(1_000.0, 0.1, 5, 1024, 7); let mut e = mon.new_edge_sketch(); e.update(1, 5); - mon.on_report("e", 0, e); + mon.on_report("e", 0, e, 10.0); assert_eq!(mon.edge_count(), 1); - // New epoch resets membership. let e2 = mon.new_edge_sketch(); - mon.on_report("e", 60_000, e2); + mon.on_report("e", 60_000, e2, 10.0); assert_eq!(mon.edge_count(), 1); - // A stale-epoch report is dropped. let e3 = mon.new_edge_sketch(); - assert_eq!(mon.on_report("e", 0, e3), None); + assert_eq!(mon.on_report("e", 0, e3, 10.0), None); // stale } } From b09e1eb4324982c779903b72ceb07c3c765a24af Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 16 Jun 2026 13:53:34 -0600 Subject: [PATCH 3/3] feat(monitor/f2): geometric safe-zone layer (communication-efficient CDM-F2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds GeometricF2Monitor — the Sharfman–Schuster–Keren safe-zone layer that makes distributed F2 monitoring communication-efficient and BREAKS the circularity "need L2 to decide whether to send, but need the sketch to get L2": each site runs a PURELY LOCAL test — does its drift ball stay inside the safe ball B(0, √(d·τ))? — using only the last-broadcast reference C_ref and its OWN drift ΔC_i, never the live global. A site ships its sketch ONLY when locally unsafe, triggering a resync. Monitored quantity ‖C‖₂² (E=d·F2) vs d·τ; convexity theorem: global C=(1/k)Σu_i is in the convex hull of u_i=C_ref+k·ΔC_i, so all per-site bounding balls ⊆ B(0,R) ⇒ ‖C‖ --- data_plane/src/monitor/f2.rs | 214 +++++++++++++++++++++++++++++++++++ 1 file changed, 214 insertions(+) diff --git a/data_plane/src/monitor/f2.rs b/data_plane/src/monitor/f2.rs index c5d114fc..0527d5a3 100644 --- a/data_plane/src/monitor/f2.rs +++ b/data_plane/src/monitor/f2.rs @@ -113,6 +113,38 @@ impl CountSketchF2 { } } + /// Squared L2 norm of the whole counter vector, `‖C‖₂² = Σ_j C[j]²`. + /// `E[‖C‖₂²] = d·F2` (each of the `d` rows is an unbiased F2 estimator), so + /// the geometric layer monitors `‖C‖₂²` against `d·τ` — a smooth quadratic + /// whose sublevel set is a ball (unlike the robust median estimate_f2()). + pub fn l2_norm_sq(&self) -> f64 { + self.c.iter().map(|&x| (x as f64) * (x as f64)).sum() + } + + /// New sketch = `self − other` (same dims). The per-site DRIFT `ΔC_i`. + pub fn minus(&self, other: &CountSketchF2) -> CountSketchF2 { + assert_eq!((self.d, self.w, self.seed), (other.d, other.w, other.seed)); + let mut out = self.clone(); + for i in 0..out.c.len() { + out.c[i] -= other.c[i]; + } + out + } + + /// `‖ self + scale·other ‖₂²` without allocating — used by the geometric + /// drift-ball test (`‖C_ref + (k/2)·ΔC_i‖`). + pub fn norm_sq_combo(&self, other: &CountSketchF2, scale: f64) -> f64 { + assert_eq!((self.d, self.w, self.seed), (other.d, other.w, other.seed)); + self.c + .iter() + .zip(&other.c) + .map(|(&a, &b)| { + let v = a as f64 + scale * (b as f64); + v * v + }) + .sum() + } + /// Unbiased F2 estimate: median over the `d` rows of `Σ_b C[r][b]²`. pub fn estimate_f2(&self) -> f64 { let mut row_f2: Vec = Vec::with_capacity(self.d); @@ -276,6 +308,118 @@ impl DistributedF2Monitor { } } +/// Geometric (Sharfman–Schuster–Keren) safe-zone layer for **communication-efficient** +/// distributed F2 monitoring — breaks the "need-L2-to-decide / need-sketch-to-get-L2" +/// circularity. Instead of every site shipping its sketch every window, each site +/// runs a PURELY LOCAL test (does its drift ball stay inside the safe ball +/// `B(0, √(d·τ))`?) using only the last-broadcast reference `C_ref` and its OWN +/// drift `ΔC_i` — never the live global. A site ships its sketch ONLY when locally +/// unsafe, triggering a resync. +/// +/// Monitored quantity: `‖C‖₂²` (with `E[‖C‖₂²]=d·F2`) against `d·τ`; safe-ball +/// radius `R=√(d·τ)`. Convexity theorem: the global `C = (1/k)·Σ u_i` lies in the +/// convex hull of the drift vectors `u_i = C_ref + k·ΔC_i`, so if every site's +/// bounding ball `B((C_ref+u_i)/2, ‖u_i−C_ref‖/2) ⊆ B(0,R)` then `‖C‖, // each site's sketch at the last sync + c_ref: CountSketchF2, // Σ refs — the broadcast reference + resyncs: u64, + alerted: bool, +} + +impl GeometricF2Monitor { + pub fn new(tau: f64, epsilon: f64, d: usize, w: usize, seed: u64) -> Self { + Self { + tau, + epsilon, + d, + w, + seed, + refs: HashMap::new(), + c_ref: CountSketchF2::new(d, w, seed), + resyncs: 0, + alerted: false, + } + } + + pub fn new_edge_sketch(&self) -> CountSketchF2 { + CountSketchF2::new(self.d, self.w, self.seed) + } + pub fn site_count(&self) -> usize { + self.refs.len() + } + pub fn resync_count(&self) -> u64 { + self.resyncs + } + /// Safe-ball radius `R = √(d·τ)` on the merged-sketch scale. + pub fn safe_radius(&self) -> f64 { + (self.d as f64 * self.tau).sqrt() + } + + /// PURELY-LOCAL safety test (runs at the EDGE in deployment): given the site's + /// CURRENT sketch, may it stay SILENT this round? Uses only the broadcast + /// `C_ref` and the site's own reference/drift — NOT the live global ‖f‖₂. + /// Returns true ⇒ the site need not communicate. + pub fn is_locally_safe(&self, site: &str, current: &CountSketchF2) -> bool { + if current.dims() != (self.d, self.w, self.seed) { + return false; + } + let zero = CountSketchF2::new(self.d, self.w, self.seed); + let reference = self.refs.get(site).unwrap_or(&zero); + let delta = current.minus(reference); // ΔC_i + let k = self.refs.len().max(1) as f64; + // Bounding ball of {C_ref, u_i = C_ref + k·ΔC_i}: centre C_ref + (k/2)·ΔC_i, + // radius (k/2)·‖ΔC_i‖. It ⊆ B(0, R) iff ‖centre‖ + radius ≤ R. + let centre_norm = self.c_ref.norm_sq_combo(&delta, k / 2.0).sqrt(); + let radius = (k / 2.0) * delta.l2_norm_sq().sqrt(); + centre_norm + radius <= self.safe_radius() + } + + /// Full resync: every site contributes its CURRENT sketch (a site lands here + /// after a local violation; the coordinator pulls all current sketches). + /// Recomputes `C_ref`, re-checks the exact global `F2̂`, and returns an Alert + /// if `F2̂ ≥ (1−ε)τ`. + pub fn resync(&mut self, currents: &HashMap) -> Option { + self.refs.clear(); + for (id, sk) in currents { + if sk.dims() == (self.d, self.w, self.seed) { + self.refs.insert(id.clone(), sk.clone()); + } + } + self.recompute_ref(); + self.resyncs += 1; + let f2 = self.global_f2(); + if !self.alerted && f2 >= (1.0 - self.epsilon) * self.tau { + self.alerted = true; + return Some(F2Action::Alert { + f2_estimate: f2, + tau: self.tau, + }); + } + Some(F2Action::Ok { f2_estimate: f2 }) + } + + fn recompute_ref(&mut self) { + let mut m = CountSketchF2::new(self.d, self.w, self.seed); + for s in self.refs.values() { + m.merge(s); + } + self.c_ref = m; + } + + /// Global `F2̂` from the synced reference sketches: `‖C_ref‖₂² / d`. + pub fn global_f2(&self) -> f64 { + self.c_ref.l2_norm_sq() / self.d as f64 + } +} + #[cfg(test)] mod tests { use super::*; @@ -397,4 +541,74 @@ mod tests { let e3 = mon.new_edge_sketch(); assert_eq!(mon.on_report("e", 0, e3, 10.0), None); // stale } + + // ── geometric safe-zone layer ── + fn gcs(mon: &GeometricF2Monitor, pairs: &[(u64, i64)]) -> CountSketchF2 { + let mut s = mon.new_edge_sketch(); + for &(k, f) in pairs { + s.update(k, f); + } + s + } + + #[test] + fn geometric_silent_under_small_drift() { + // sync 2 sites each {1:40} ⇒ merged {1:80}, F2=6400 < τ=10000 (R=300). + let mut mon = GeometricF2Monitor::new(10_000.0, 0.1, 9, 4096, 0x1111); + let mut cur = HashMap::new(); + cur.insert("a".to_string(), gcs(&mon, &[(1, 40)])); + cur.insert("b".to_string(), gcs(&mon, &[(1, 40)])); + mon.resync(&cur); + assert!(mon.global_f2() < mon.tau); + // tiny drift (+1 each) ⇒ both stay locally SAFE ⇒ zero communication. + assert!(mon.is_locally_safe("a", &gcs(&mon, &[(1, 41)])), "small drift must be safe"); + assert!(mon.is_locally_safe("b", &gcs(&mon, &[(1, 41)]))); + } + + #[test] + fn geometric_triggers_when_drift_threatens_tau() { + let mut mon = GeometricF2Monitor::new(10_000.0, 0.1, 9, 4096, 0x2222); + let mut cur = HashMap::new(); + cur.insert("a".to_string(), gcs(&mon, &[(1, 40)])); + cur.insert("b".to_string(), gcs(&mon, &[(1, 40)])); + mon.resync(&cur); + // one site drifts hard ⇒ local safe-zone trips ⇒ it MUST report. + assert!(!mon.is_locally_safe("a", &gcs(&mon, &[(1, 200)])), "large drift must trip safe-zone"); + } + + #[test] + fn geometric_stays_silent_then_resyncs_on_violation() { + let mut mon = GeometricF2Monitor::new(10_000.0, 0.1, 9, 4096, 0x3333); + let a = gcs(&mon, &[(1, 30)]); + let b = gcs(&mon, &[(1, 30)]); + let mut cur = HashMap::new(); + cur.insert("a".to_string(), a.clone()); + cur.insert("b".to_string(), b); + mon.resync(&cur); + let base = mon.resync_count(); + // many small local updates: all silent (no resync triggered). + let mut ca = a; + for _ in 0..5 { + ca.update(1, 1); + assert!(mon.is_locally_safe("a", &ca)); + } + assert_eq!(mon.resync_count(), base, "silent updates must not resync"); + // a big jump trips the safe-zone → protocol resyncs. + let big = gcs(&mon, &[(1, 300)]); + assert!(!mon.is_locally_safe("a", &big)); + cur.insert("a".to_string(), big); + mon.resync(&cur); + assert_eq!(mon.resync_count(), base + 1); + } + + #[test] + fn geometric_alert_when_resync_global_exceeds_tau() { + // merged {1:100} ⇒ F2=10000 ≥ (1-0.1)·8000=7200 ⇒ Alert. + let mut mon = GeometricF2Monitor::new(8_000.0, 0.1, 9, 4096, 0x4444); + let mut cur = HashMap::new(); + cur.insert("a".to_string(), gcs(&mon, &[(1, 50)])); + cur.insert("b".to_string(), gcs(&mon, &[(1, 50)])); + let act = mon.resync(&cur).unwrap(); + assert!(matches!(act, F2Action::Alert { .. }), "global over τ ⇒ Alert, got {act:?}"); + } }