mirror of
https://github.com/ruvnet/RuView
synced 2026-08-04 19:31:42 +00:00
fix: multistatic canonicalization, csi_fps burst inflation, control-packet starvation (#1170, #1180, #1183) (#1193)
#1170 — live multistatic bridge fed raw, un-canonicalized per-node CSI (64/128/192 bins) to MultistaticFuser, tripping DimensionMismatch every cycle and silently disabling fusion on mixed HT20/HT40 meshes. Add HardwareNormalizer::resample_to_canonical (resample-only, no z-score) and canonicalize every node frame onto the 56-tone grid before fusion. #1180 — update_csi_fps_ema only rejected dt<=0 or >=1s, so sub-ms UDP-burst arrivals (36us -> ~27kHz) inflated csi_fps_ema 40-840x. Add a 5ms plausibility floor and stop re-anchoring observe_csi_frame_arrival on burst deltas. #1183 — global ENOMEM backoff (CSI flood) starved <=48B/<=1Hz control packets. Add stream_sender_send_priority() bypassing the backoff gate without touching the streak; route feature_state/HEALTH/sync through it. Fix the misleading "HEALTH sent" log that printed even on rv_mesh_send failure. Verified: signal 501, sensing-server 677 tests (0 failed); firmware builds clean. Claude-Session: https://claude.ai/code/session_01AgpTcBLRJ32hUsKWxDXf36
This commit is contained in:
@@ -402,17 +402,36 @@ mod tests {
|
||||
assert!(!bridge.suppress_raw_outputs());
|
||||
}
|
||||
|
||||
/// Error wiring (review finding 1a): two live nodes with mismatched
|
||||
/// subcarrier counts make fusion return a `DimensionMismatch` →
|
||||
/// `EngineError` — previously dropped by `if let Some(Ok(..))` at the
|
||||
/// Error wiring (review finding 1a): a live cycle that fails fusion yields
|
||||
/// an `EngineError` — previously dropped by `if let Some(Ok(..))` at the
|
||||
/// call sites. The counter must increment and the last good trust state
|
||||
/// must survive a later failure.
|
||||
///
|
||||
/// Originally this forced the failure with a 56-vs-30 subcarrier mismatch
|
||||
/// (`DimensionMismatch`). Since #1170 the live bridge canonicalizes every
|
||||
/// node onto the 56-tone grid, so heterogeneous counts now fuse cleanly —
|
||||
/// a frame-timestamp spread wider than the fuser's 60 ms guard interval is
|
||||
/// the remaining deterministic way to provoke a fusion error here.
|
||||
#[test]
|
||||
fn observe_cycle_counts_engine_errors() {
|
||||
// Both nodes are 56-subcarrier (canonicalization-clean), but their
|
||||
// frame timestamps are 500 ms apart — far beyond the 60 ms guard —
|
||||
// so the fuser rejects the cycle with TimestampMismatch. Future
|
||||
// offsets keep both instants safely after the bridge's lazy EPOCH.
|
||||
fn mismatched_states() -> HashMap<u8, NodeState> {
|
||||
let now = Instant::now();
|
||||
let mut a = node_state_with_history(1.0, 56);
|
||||
a.last_frame_time = Some(now + std::time::Duration::from_millis(600));
|
||||
let mut b = node_state_with_history(1.05, 56);
|
||||
b.last_frame_time = Some(now + std::time::Duration::from_millis(100));
|
||||
let mut m = HashMap::new();
|
||||
m.insert(0u8, a);
|
||||
m.insert(1u8, b);
|
||||
m
|
||||
}
|
||||
|
||||
let mut bridge = EngineBridge::new(PrivacyMode::PrivateHome, 1, "r", "R");
|
||||
let mut mismatched = HashMap::new();
|
||||
mismatched.insert(0u8, node_state_with_history(1.0, 56));
|
||||
mismatched.insert(1u8, node_state_with_history(1.05, 30)); // 30 ≠ 56 subcarriers
|
||||
let mismatched = mismatched_states();
|
||||
|
||||
assert!(bridge.observe_cycle(&mismatched, 1_000).is_none());
|
||||
assert_eq!(bridge.engine_error_count(), 1);
|
||||
|
||||
@@ -518,17 +518,31 @@ const NOVELTY_HISTORY_CAPACITY: usize = 64;
|
||||
/// subcarrier ordering / normalisation so banks reject stale data.
|
||||
const NOVELTY_SKETCH_VERSION: u16 = 1;
|
||||
|
||||
/// Lower plausibility floor (seconds) for a CSI inter-frame delta.
|
||||
///
|
||||
/// The firmware caps CSI sends at `CSI_MIN_SEND_INTERVAL_US = 20 ms`
|
||||
/// (`csi_collector.c`), so a single node cannot physically produce frames
|
||||
/// faster than 50 fps. UDP/OS buffering, however, delivers frames in tight
|
||||
/// bursts whose intra-burst arrival deltas are tens of microseconds apart —
|
||||
/// a 36 µs delta yields `1/dt ≈ 27 kHz`, which the old `< 1 s` guard let
|
||||
/// straight into the EMA and inflated `csi_fps_ema` by 1–3 orders of
|
||||
/// magnitude (issue #1180). We reject any delta implying more than 200 fps
|
||||
/// (4× the physical ceiling, leaving slack for benign arrival jitter); such
|
||||
/// deltas are burst artifacts, not distinct production intervals.
|
||||
pub(crate) const MIN_PLAUSIBLE_CSI_DT_SEC: f64 = 0.005;
|
||||
|
||||
/// ADR-110 iter 18 — EMA update for per-node CSI fps tracking.
|
||||
///
|
||||
/// Returns the new EMA value, or `None` if the delta is implausible
|
||||
/// (≤ 0, or > 1 second — likely a connection gap, not a real frame
|
||||
/// rate sample). α = 1/8 fixed shift, ~8-sample effective window,
|
||||
/// matching the firmware-side ESP-NOW offset smoother in §A0.10.
|
||||
/// (below [`MIN_PLAUSIBLE_CSI_DT_SEC`] — a sub-ms burst artifact, see
|
||||
/// issue #1180 — or `> 1 second`, likely a connection gap rather than a
|
||||
/// real frame-rate sample). α = 1/8 fixed shift, ~8-sample effective
|
||||
/// window, matching the firmware-side ESP-NOW offset smoother in §A0.10.
|
||||
///
|
||||
/// Free function for testability — every transformation that doesn't
|
||||
/// touch the rest of `NodeState` lives outside the `impl` block.
|
||||
pub(crate) fn update_csi_fps_ema(prev_fps: f64, dt_sec: f64) -> Option<f64> {
|
||||
if !(dt_sec > 0.0 && dt_sec < 1.0) {
|
||||
if !(dt_sec >= MIN_PLAUSIBLE_CSI_DT_SEC && dt_sec < 1.0) {
|
||||
return None;
|
||||
}
|
||||
let instantaneous = 1.0 / dt_sec;
|
||||
@@ -569,6 +583,35 @@ mod fps_ema_tests {
|
||||
fn long_gap_rejected_as_implausible() {
|
||||
assert!(update_csi_fps_ema(20.0, 2.0).is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn subms_burst_delta_rejected() {
|
||||
// Issue #1180: a 36 µs intra-burst delta implies ~27 kHz and must
|
||||
// not enter the EMA. Anything below the 5 ms floor is rejected.
|
||||
assert!(update_csi_fps_ema(40.0, 0.000_036).is_none());
|
||||
assert!(update_csi_fps_ema(40.0, 0.001).is_none());
|
||||
// Just above the floor is accepted.
|
||||
assert!(update_csi_fps_ema(40.0, 0.005).is_some());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn burst_interleaved_with_nominal_stays_in_band() {
|
||||
// A true ~40 fps node whose frames arrive in sub-ms bursts: feeding
|
||||
// only the plausible (nominal-cadence) deltas keeps the EMA near the
|
||||
// ground truth instead of blowing up. Burst deltas are rejected by
|
||||
// the caller (see NodeState::observe_csi_frame_arrival), so the EMA
|
||||
// only ever sees the ~25 ms inter-group gaps.
|
||||
let mut fps = 40.0;
|
||||
for _ in 0..40 {
|
||||
// nominal 25 ms gap (40 fps); intervening sub-ms bursts skipped
|
||||
fps = update_csi_fps_ema(fps, 0.025).unwrap();
|
||||
assert!(update_csi_fps_ema(fps, 0.000_040).is_none());
|
||||
}
|
||||
assert!(
|
||||
(fps - 40.0).abs() < 1.0,
|
||||
"EMA should stay within ~1 Hz of the 40 fps ground truth, got {fps}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
impl NodeState {
|
||||
@@ -653,6 +696,15 @@ impl NodeState {
|
||||
pub(crate) fn observe_csi_frame_arrival(&mut self, now: std::time::Instant) {
|
||||
if let Some(prev) = self.last_frame_time {
|
||||
let dt = now.duration_since(prev).as_secs_f64();
|
||||
// Burst arrivals (sub-floor dt, issue #1180): do NOT re-anchor on
|
||||
// them. Keeping the previous anchor means the next genuine
|
||||
// inter-frame gap measures the true cadence across the whole
|
||||
// burst instead of intra-burst jitter — so a 50 fps node whose
|
||||
// frames arrive in 36 µs bursts every 25 ms still reads ~40 fps,
|
||||
// not 27 kHz.
|
||||
if dt < MIN_PLAUSIBLE_CSI_DT_SEC {
|
||||
return;
|
||||
}
|
||||
if let Some(new_ema) = update_csi_fps_ema(self.csi_fps_ema, dt) {
|
||||
self.csi_fps_ema = new_ema;
|
||||
self.csi_fps_samples = self.csi_fps_samples.saturating_add(1);
|
||||
@@ -8037,6 +8089,36 @@ mod sync_snapshot_helper_tests {
|
||||
assert_eq!(snap.csi_fps_samples, 42);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn observe_csi_frame_arrival_ignores_subms_bursts() {
|
||||
// Issue #1180 regression: a ~40 fps node whose frames are delivered
|
||||
// in tight UDP bursts (sub-ms intra-burst deltas) must still report
|
||||
// ~40 fps, not tens of kHz. Synthesize the arrival stream by adding
|
||||
// Durations to a base Instant.
|
||||
use std::time::Duration;
|
||||
let base = std::time::Instant::now();
|
||||
let mut ns = NodeState::new();
|
||||
ns.csi_fps_ema = 40.0; // pretend already warmed up
|
||||
ns.csi_fps_samples = 10;
|
||||
|
||||
// 30 nominal 25 ms groups, each preceded by a 3-frame sub-ms burst.
|
||||
for g in 0..30u64 {
|
||||
let group_t = base + Duration::from_millis(25 * g);
|
||||
ns.observe_csi_frame_arrival(group_t);
|
||||
// burst: two extra arrivals 40 µs and 80 µs later — must be
|
||||
// ignored for rate purposes (anchor must not advance to them).
|
||||
ns.observe_csi_frame_arrival(group_t + Duration::from_micros(40));
|
||||
ns.observe_csi_frame_arrival(group_t + Duration::from_micros(80));
|
||||
}
|
||||
|
||||
assert!(
|
||||
(ns.csi_fps_ema - 40.0).abs() < 2.0,
|
||||
"csi_fps_ema must stay near the 40 fps ground truth despite \
|
||||
sub-ms bursts, got {}",
|
||||
ns.csi_fps_ema
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn apply_sync_packet_populates_a_fresh_node() {
|
||||
// Mirrors what udp_receiver_task does on the very first sync
|
||||
|
||||
@@ -10,7 +10,7 @@ use std::collections::HashMap;
|
||||
use std::sync::LazyLock;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use wifi_densepose_signal::hardware_norm::{CanonicalCsiFrame, HardwareType};
|
||||
use wifi_densepose_signal::hardware_norm::{CanonicalCsiFrame, HardwareNormalizer, HardwareType};
|
||||
use wifi_densepose_signal::ruvsense::multiband::MultiBandCsiFrame;
|
||||
use wifi_densepose_signal::ruvsense::multistatic::{FusedSensingFrame, MultistaticFuser};
|
||||
|
||||
@@ -26,6 +26,11 @@ const DEFAULT_FREQ_MHZ: u32 = 2437; // Channel 6
|
||||
/// are relative to this instant, avoiding wall-clock/monotonic mixing issues.
|
||||
static EPOCH: LazyLock<Instant> = LazyLock::new(Instant::now);
|
||||
|
||||
/// Shared length-only canonicalizer (issue #1170). The default 56-tone grid
|
||||
/// matches what `MultistaticFuser` (ADR-154) expects. Stateless and immutable,
|
||||
/// so a single process-wide instance is safe to share across nodes.
|
||||
static NORMALIZER: LazyLock<HardwareNormalizer> = LazyLock::new(HardwareNormalizer::new);
|
||||
|
||||
/// Convert a single `NodeState` into a `MultiBandCsiFrame` suitable for
|
||||
/// multistatic fusion.
|
||||
///
|
||||
@@ -38,7 +43,14 @@ pub fn node_frame_from_state(node_id: u8, ns: &NodeState) -> Option<MultiBandCsi
|
||||
return None;
|
||||
}
|
||||
|
||||
let amplitude: Vec<f32> = latest.iter().map(|&v| v as f32).collect();
|
||||
// Issue #1170: resample the raw amplitude onto the canonical 56-tone grid
|
||||
// BEFORE fusion. ESP32 nodes in mixed HT20/HT40 capture modes report
|
||||
// different subcarrier counts (64 / 128 / 192); feeding those raw into
|
||||
// `MultistaticFuser::fuse` tripped `DimensionMismatch` on every cycle and
|
||||
// silently disabled real multistatic fusion. Length-only canonicalization
|
||||
// (no z-score) keeps the amplitude scale the person-score relies on.
|
||||
let canonical_amp = NORMALIZER.resample_to_canonical(latest);
|
||||
let amplitude: Vec<f32> = canonical_amp.iter().map(|&v| v as f32).collect();
|
||||
let n_sub = amplitude.len();
|
||||
let phase = vec![0.0_f32; n_sub];
|
||||
|
||||
@@ -201,15 +213,58 @@ mod tests {
|
||||
assert_eq!(frame.channel_frames.len(), 1);
|
||||
|
||||
let ch = &frame.channel_frames[0];
|
||||
assert_eq!(ch.amplitude.len(), 3);
|
||||
assert!((ch.amplitude[0] - 10.0_f32).abs() < f32::EPSILON);
|
||||
assert!((ch.amplitude[1] - 20.0_f32).abs() < f32::EPSILON);
|
||||
assert!((ch.amplitude[2] - 30.5_f32).abs() < f32::EPSILON);
|
||||
// Issue #1170: amplitude is now resampled onto the canonical 56-tone
|
||||
// grid regardless of the raw count.
|
||||
assert_eq!(ch.amplitude.len(), 56);
|
||||
// resample_cubic preserves the endpoints (no z-scoring), so the scale
|
||||
// the person-score relies on is intact.
|
||||
assert!((ch.amplitude[0] - 10.0_f32).abs() < 1e-3);
|
||||
assert!((ch.amplitude[55] - 30.5_f32).abs() < 1e-3);
|
||||
// Phase should be all zeros
|
||||
assert!(ch.phase.iter().all(|&p| p == 0.0));
|
||||
assert_eq!(ch.hardware_type, HardwareType::Esp32S3);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn heterogeneous_node_counts_canonicalize_and_fuse() {
|
||||
// Issue #1170 regression: a mixed mesh with HT20 (64-bin) and HT40
|
||||
// (192-bin) nodes must canonicalize to a uniform 56 tones and fuse,
|
||||
// instead of tripping DimensionMismatch on every cycle.
|
||||
let mut states: HashMap<u8, NodeState> = HashMap::new();
|
||||
|
||||
let mut h64 = VecDeque::new();
|
||||
h64.push_back((0..64).map(|i| 1.0 + 0.1 * i as f64).collect::<Vec<f64>>());
|
||||
states.insert(1, make_node_state(h64, Some(Instant::now()), 1));
|
||||
|
||||
let mut h192 = VecDeque::new();
|
||||
h192.push_back((0..192).map(|i| 2.0 + 0.05 * i as f64).collect::<Vec<f64>>());
|
||||
states.insert(3, make_node_state(h192, Some(Instant::now()), 1));
|
||||
|
||||
let frames = node_frames_from_states(&states);
|
||||
assert_eq!(frames.len(), 2, "both nodes should produce frames");
|
||||
for f in &frames {
|
||||
assert_eq!(
|
||||
f.channel_frames[0].amplitude.len(),
|
||||
56,
|
||||
"every node must present the canonical 56-tone dimension"
|
||||
);
|
||||
}
|
||||
|
||||
// The fuser must now accept the cycle (no DimensionMismatch).
|
||||
let fuser = MultistaticFuser::new();
|
||||
let result = fuser.fuse(&frames);
|
||||
assert!(
|
||||
result.is_ok(),
|
||||
"heterogeneous mesh should fuse after canonicalization, got {result:?}"
|
||||
);
|
||||
|
||||
// And the higher-level fallback path returns the fused frame, not the
|
||||
// sum/dedup fallback.
|
||||
let (fused, fallback) = fuse_or_fallback(&fuser, &states, 3.0);
|
||||
assert!(fused.is_some(), "fusion should succeed");
|
||||
assert!(fallback.is_none(), "no fallback when fusion succeeds");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_stale_node_excluded() {
|
||||
let mut states: HashMap<u8, NodeState> = HashMap::new();
|
||||
|
||||
@@ -167,6 +167,22 @@ impl HardwareNormalizer {
|
||||
hardware_type: hw,
|
||||
})
|
||||
}
|
||||
|
||||
/// Resample a raw 1-D CSI vector onto the canonical subcarrier grid
|
||||
/// **without** z-score normalization (length-only canonicalization).
|
||||
///
|
||||
/// Used by the live multistatic bridge (issue #1170): heterogeneous
|
||||
/// ESP32 capture modes report different subcarrier counts (HT20 ≈ 64,
|
||||
/// HT40 ≈ 128/192), and [`MultistaticFuser`] requires every node frame
|
||||
/// to share one dimension. Full [`Self::normalize`] would z-score the
|
||||
/// amplitude (mean → 0), which saturates the downstream person-score
|
||||
/// (a squared coefficient of variation `variance / mean²`); resampling
|
||||
/// alone makes frames fusable while preserving amplitude scale.
|
||||
///
|
||||
/// [`MultistaticFuser`]: crate::ruvsense::multistatic::MultistaticFuser
|
||||
pub fn resample_to_canonical(&self, raw: &[f64]) -> Vec<f64> {
|
||||
resample_cubic(raw, self.canonical_subcarriers)
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for HardwareNormalizer {
|
||||
@@ -344,6 +360,25 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn resample_to_canonical_is_length_only_no_zscore() {
|
||||
// Issue #1170: resample_to_canonical must change length to 56 but
|
||||
// NOT z-score (mean must be preserved, not driven to ~0). A raw
|
||||
// amplitude vector with a large positive mean keeps that mean.
|
||||
let norm = HardwareNormalizer::new();
|
||||
let raw: Vec<f64> = (0..192).map(|i| 50.0 + 0.1 * i as f64).collect();
|
||||
let out = norm.resample_to_canonical(&raw);
|
||||
assert_eq!(out.len(), 56, "must resample onto the 56-tone grid");
|
||||
let mean = out.iter().sum::<f64>() / out.len() as f64;
|
||||
assert!(
|
||||
mean > 40.0,
|
||||
"resample-only must preserve amplitude scale (mean ~60), got {mean}"
|
||||
);
|
||||
// Endpoints preserved.
|
||||
assert!((out[0] - raw[0]).abs() < 1e-6);
|
||||
assert!((out[55] - raw[191]).abs() < 0.5);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn zscore_produces_zero_mean_unit_std() {
|
||||
let data: Vec<f64> = (0..100)
|
||||
|
||||
Reference in New Issue
Block a user