From 1beedb4114d82ec21ccbd0fa4e47419d2a573d74 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 16 Jun 2026 10:56:52 -0600 Subject: [PATCH] fix(monitor): coordinated sampling allocates on per-sid (agg-id) freq f_i, not total rate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit coordinator.rs::allocate_p() passed the edge `rate` vector as BOTH the rate and the freq argument to allocate_sample_rates(), so √(f_i/rate_i)=1 for every edge and the KKT allocation collapsed to a uniform p — all differentiation was left to the ε-floor (rate-only). The MonitorReport already carries `known_value`: the monitored functional's per-edge value, i.e. the frequency of the monitored series id (the cms_point sid — internally the CMS `key`) / the edge's sum contribution, identified by the report's aggregation_id. That, not the total update rate, is the correct f_i. Feed it as the freq vector so the allocation is genuinely p_i ∝ √(f_i/rate_i), evaluated per-aggregation per-edge: an edge carrying more of the monitored sid at equal total rate now keeps a higher p. Edges with no monitored-sid mass (known_value=0) fall back to p=1. New test per_key_freq_differentiates_at_equal_rate pins the freq-driven branch (equal rate, different known_value ⇒ different p, both above the equal rate-floor); the existing skewed-rate (floor-driven) test still passes. 13/13 monitor tests green. Co-Authored-By: Claude Opus 4.8 --- data_plane/src/monitor/coordinator.rs | 46 +++++++++++++++++++++++++-- 1 file changed, 43 insertions(+), 3 deletions(-) diff --git a/data_plane/src/monitor/coordinator.rs b/data_plane/src/monitor/coordinator.rs index 5ad71ef14..8c7a16f1f 100644 --- a/data_plane/src/monitor/coordinator.rs +++ b/data_plane/src/monitor/coordinator.rs @@ -235,15 +235,27 @@ impl Monitor { /// 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. + // Stable edge order so the rate / freq vectors align 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). + // Per-key frequency `f_i` = the monitored functional's value this edge + // knows this epoch (`known_value`: the cms_point key frequency, or the + // edge's contribution to the monitored sum), which is DISTINCT from the + // edge's total update `rate`. Feeding `f_i` (not `rates`) into the KKT + // allocation is what makes `p_i ∝ √(f_i/rate_i)` differentiate by the + // monitored key's *share* of each edge's stream. Previously this passed + // `rates` as both vectors → `√(rate_i/rate_i)=1` for every edge → a + // uniform allocation, with ALL differentiation left to the ε-floor; + // now the allocation itself is freq-driven (an edge that carries more of + // the monitored key at equal total rate keeps a higher p). An edge with + // no monitored-key mass yet (`known_value=0`) falls back to `p_i=1` + // (allocate_sample_rates: `freqs_i ≤ 0 ⇒ p_i=1`), i.e. nothing to sample. + let freqs: Vec = ids.iter().map(|id| self.edges[*id].known_value).collect(); let var_budget = { let band = self.cfg.epsilon * self.cfg.tau; band * band }; - let mut p_vec = allocate_sample_rates(&rates, &rates, var_budget); + let mut p_vec = allocate_sample_rates(&rates, &freqs, 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() { @@ -501,6 +513,34 @@ mod tests { assert_eq!(slack_in(&a, "e1"), slack_in(&a, "e2")); } + #[test] + fn per_key_freq_differentiates_at_equal_rate() { + // EQUAL total rate, but different monitored-key frequency (`known_value`): + // the edge carrying more of the monitored key keeps a HIGHER p (sampled + // less), since p_i ∝ √(f_i/rate_i) and rate cancels. Under the old + // freqs==rates this returned a UNIFORM p (same rate ⇒ same floor ⇒ tie), + // so this test fails on the pre-fix code — it pins the freq-driven branch. + // High rate ⇒ low ε-floor, modest τ ⇒ tight var_budget, so the allocation + // (not the floor) is the operative differentiator. + let mut m = Monitor::new(cfg(2_000.0)); // epsilon 0.05 ⇒ var_budget=(100)^2 + m.on_register("e1", 60_000, 0); + m.on_register("e2", 60_000, 0); + // same rate (100k/win); e1 sees the monitored key 10× more than e2. + m.on_report("e1", 0, 900.0, 1, 100_000.0); // known_value=900 + let a = m.on_report("e2", 0, 90.0, 1, 100_000.0); // known_value=90 + let p_e1 = sample_p_in(&a, "e1"); + let p_e2 = sample_p_in(&a, "e2"); + let floor = epsilon_sample_floor(m.cfg.epsilon, 100_000.0); + assert!( + p_e1 > p_e2, + "high-freq edge p {p_e1} should exceed low-freq p {p_e2} (freq-driven, not floor)" + ); + // Both sit ABOVE the (equal) rate-floor, proving the allocation — not the + // floor — produced the spread. + assert!(p_e1 > floor + 1e-6 && p_e2 > floor + 1e-6, + "both p ({p_e1}, {p_e2}) should exceed the equal rate-floor {floor}"); + } + #[test] fn epoch_advance_resets_estimate() { let mut m = Monitor::new(cfg(100.0));