diff --git a/v2/crates/wifi-densepose-sensing-server/src/engine_bridge.rs b/v2/crates/wifi-densepose-sensing-server/src/engine_bridge.rs index 5941a655..25ab95b1 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/engine_bridge.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/engine_bridge.rs @@ -41,7 +41,7 @@ use wifi_densepose_signal::ruvsense::fusion_quality::CalibrationId; use wifi_densepose_signal::ruvsense::multistatic::MultistaticConfig; use wifi_densepose_worldgraph::WorldId; -use super::multistatic_bridge::node_frames_from_states; +use super::multistatic_bridge::node_frames_from_states_with_guard; use super::NodeState; /// Minimum spacing between engine-error warn logs (errors are still counted @@ -53,6 +53,8 @@ const ENGINE_ERROR_WARN_INTERVAL: Duration = Duration::from_secs(10); /// live sensing loop publishes beliefs into. pub struct EngineBridge { engine: StreamingEngine, + /// Hard timestamp guard shared by cohort selection and engine validation. + guard_interval_us: u64, room: WorldId, /// Nodes already wired into the WorldGraph as sensors (by `node_id`). registered_nodes: HashMap, @@ -94,6 +96,11 @@ impl EngineBridge { room_name: &str, multistatic_cfg: Option, ) -> Self { + let guard_interval_us = multistatic_cfg + .as_ref() + .map_or_else(|| MultistaticConfig::default().guard_interval_us, |cfg| { + cfg.guard_interval_us + }); let mut engine = StreamingEngine::new(mode, model_version, GeoRegistration::default()); if let Some(cfg) = multistatic_cfg { engine.set_multistatic_config(cfg); @@ -101,6 +108,7 @@ impl EngineBridge { let room = engine.add_room(room_area_id, room_name); Self { engine, + guard_interval_us, room, registered_nodes: HashMap::new(), calibration: CalibrationId(0x5256_0001), // "RV\0\x01" — placeholder epoch @@ -165,7 +173,8 @@ impl EngineBridge { node_states: &HashMap, now_ms: i64, ) -> Option> { - let frames = node_frames_from_states(node_states); + let frames = + node_frames_from_states_with_guard(node_states, self.guard_interval_us); if frames.is_empty() { return None; } @@ -195,7 +204,15 @@ impl EngineBridge { node_states: &HashMap, now_ms: i64, ) -> Option { - match self.process_cycle_from_states(node_states, now_ms)? { + let result = self.process_cycle_from_states(node_states, now_ms)?; + self.record_cycle_result(result) + } + + fn record_cycle_result( + &mut self, + result: Result, + ) -> Option { + match result { Ok(trust) => { self.last_witness = Some(trust.witness); self.recalibration_recommended = trust.recalibration_recommended; @@ -419,45 +436,29 @@ mod tests { assert!(!bridge.suppress_raw_outputs()); } - /// 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. + /// Error-result accounting is independent from live cohort selection, so + /// a future engine error remains auditable without deliberately forwarding + /// temporally incoherent frames through the production bridge. #[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 { - 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", None); - let mismatched = mismatched_states(); + let error = || { + EngineError::Fusion( + wifi_densepose_signal::ruvsense::multistatic::MultistaticError::TimestampMismatch { + spread_us: 500_000, + guard_us: 60_000, + }, + ) + }; - assert!(bridge.observe_cycle(&mismatched, 1_000).is_none()); + assert!(bridge.record_cycle_result(Err(error())).is_none()); assert_eq!(bridge.engine_error_count(), 1); assert!( bridge.last_trust_witness().is_none(), "no witness from a failed cycle" ); - assert!(bridge.observe_cycle(&mismatched, 2_000).is_none()); + assert!(bridge.record_cycle_result(Err(error())).is_none()); assert_eq!(bridge.engine_error_count(), 2); // A later good cycle records trust state; the audit count is kept. @@ -467,11 +468,39 @@ mod tests { assert_eq!(bridge.engine_error_count(), 2); // And a subsequent failure keeps the last good witness readable. - assert!(bridge.observe_cycle(&mismatched, 4_000).is_none()); + assert!(bridge.record_cycle_result(Err(error())).is_none()); assert_eq!(bridge.engine_error_count(), 3); assert!(bridge.last_trust_witness().is_some()); } + #[test] + fn governed_cycles_prune_slow_mixed_width_nodes_without_errors() { + let mut bridge = EngineBridge::new(PrivacyMode::PrivateHome, 1, "r", "R", None); + + for tick in 0..50_i64 { + let now = Instant::now(); + let mut states = HashMap::new(); + for (node_id, age_ms, n_sub) in [ + (1, 0, 64), + (3, 10, 256), + (4, 50, 64), + (7, 1_000, 256), + ] { + let mut node = node_state_with_history(1.0 + node_id as f64 * 0.01, n_sub); + node.last_frame_time = Some(now - Duration::from_millis(age_ms)); + states.insert(node_id, node); + } + + assert!( + bridge.observe_cycle(&states, 1_000 + tick * 50).is_some(), + "governed cycle {tick}" + ); + } + + assert_eq!(bridge.engine_error_count(), 0); + assert!(bridge.last_trust_witness().is_some()); + } + /// ADR-141 mapping (review finding 1c): a cycle emitted at class /// Restricted flips `suppress_raw_outputs`, which `main.rs` uses to strip /// per-node raw amplitude vectors from the live publish — the same field diff --git a/v2/crates/wifi-densepose-sensing-server/src/multistatic_bridge.rs b/v2/crates/wifi-densepose-sensing-server/src/multistatic_bridge.rs index a84f8989..02b0bf4f 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/multistatic_bridge.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/multistatic_bridge.rs @@ -12,7 +12,9 @@ use std::time::{Duration, Instant}; use wifi_densepose_signal::hardware_norm::{CanonicalCsiFrame, HardwareNormalizer, HardwareType}; use wifi_densepose_signal::ruvsense::multiband::MultiBandCsiFrame; -use wifi_densepose_signal::ruvsense::multistatic::{FusedSensingFrame, MultistaticFuser}; +use wifi_densepose_signal::ruvsense::multistatic::{ + FusedSensingFrame, MultistaticConfig, MultistaticFuser, +}; use super::NodeState; @@ -44,6 +46,24 @@ static NORMALIZER: LazyLock = LazyLock::new(HardwareNormaliz /// `last_frame_time`. pub fn node_frame_from_state(node_id: u8, ns: &NodeState) -> Option { let last_time = ns.last_frame_time.as_ref()?; + let timestamp_us = ns + .mesh_aligned_us_for_latest_csi_frame() + .unwrap_or_else(|| host_arrival_timestamp_us(last_time)); + node_frame_from_state_at(node_id, ns, timestamp_us) +} + +fn host_arrival_timestamp_us(last_time: &Instant) -> u64 { + last_time + .checked_duration_since(*EPOCH) + .unwrap_or_default() + .as_micros() as u64 +} + +fn node_frame_from_state_at( + node_id: u8, + ns: &NodeState, + timestamp_us: u64, +) -> Option { let latest = ns.frame_history.back()?; if latest.is_empty() { return None; @@ -60,19 +80,6 @@ pub fn node_frame_from_state(node_id: u8, ns: &NodeState) -> Option Option) -> Vec { - let now = Instant::now(); - let mut frames = Vec::with_capacity(node_states.len()); + node_frames_from_states_with_guard( + node_states, + MultistaticConfig::default().guard_interval_us, + ) +} - for (&node_id, ns) in node_states { - // Skip stale nodes - if let Some(ref t) = ns.last_frame_time { - if now.duration_since(*t) > STALE_THRESHOLD { - continue; - } - } else { +/// Collect the freshest temporally coherent cohort of active node frames. +/// +/// Nodes can publish at very different rates (for example, a mixed S3/C6 +/// fleet). `STALE_THRESHOLD` determines whether a node is alive; it does not +/// mean its latest frame belongs to the current sensing cycle. After choosing +/// one timestamp domain for the whole cycle, retain only frames within the +/// fuser's hard guard of the freshest frame. This prevents a slow-but-live node +/// from turning every governed cycle into `TimestampMismatch`, while always +/// preserving at least the freshest node for the supported single-node path. +pub fn node_frames_from_states_with_guard( + node_states: &HashMap, + guard_interval_us: u64, +) -> Vec { + let now = Instant::now(); + let mut active: Vec<(u8, &NodeState)> = node_states + .iter() + .filter_map(|(&node_id, ns)| { + let last_time = ns.last_frame_time.as_ref()?; + (now.duration_since(*last_time) <= STALE_THRESHOLD).then_some((node_id, ns)) + }) + .collect(); + active.sort_unstable_by_key(|(node_id, _)| *node_id); + + if active.is_empty() { + return Vec::new(); + } + + let guard_interval_us = guard_interval_us.max(1); + + // Timestamp domains must be selected for the cycle as a whole. A CSI + // frame can legitimately arrive between periodic sync-marked frames. If + // that one node fell back to process-local host time while a peer retained + // mesh epoch time, the resulting hundreds-of-seconds spread made every + // governed fusion cycle fail. Use mesh time only when every active node + // can provide it; otherwise use host-arrival time consistently for all. + let mesh_times: Option> = active + .iter() + .map(|(_, ns)| ns.mesh_aligned_us_for_latest_csi_frame()) + .collect::>>() + .filter(|times| { + let Some(min) = times.iter().min() else { + return false; + }; + let Some(max) = times.iter().max() else { + return false; + }; + max.saturating_sub(*min) <= guard_interval_us + }); + + let mut timed = Vec::with_capacity(active.len()); + for (index, (node_id, ns)) in active.into_iter().enumerate() { + let timestamp_us = mesh_times.as_ref().map_or_else( + || { + host_arrival_timestamp_us( + ns.last_frame_time.as_ref().expect("active node has time"), + ) + }, + |times| times[index], + ); + timed.push((node_id, ns, timestamp_us)); + } + + let freshest_timestamp_us = timed + .iter() + .map(|(_, _, timestamp_us)| *timestamp_us) + .max() + .expect("non-empty active cohort"); + + let mut frames = Vec::with_capacity(timed.len()); + for (node_id, ns, timestamp_us) in timed { + if freshest_timestamp_us.saturating_sub(timestamp_us) > guard_interval_us { continue; } - - if let Some(frame) = node_frame_from_state(node_id, ns) { + if let Some(frame) = node_frame_from_state_at(node_id, ns, timestamp_us) { frames.push(frame); } } @@ -125,7 +198,8 @@ pub fn fuse_or_fallback( node_states: &HashMap, dedup_factor: f64, ) -> (Option, Option) { - let frames = node_frames_from_states(node_states); + let frames = + node_frames_from_states_with_guard(node_states, fuser.guard_interval_us()); if frames.is_empty() { return (None, Some(0)); } @@ -313,26 +387,127 @@ mod tests { } #[test] - fn unsynchronized_frames_keep_host_arrival_guard() { + fn partial_sync_uses_one_host_timestamp_domain_for_the_cycle() { let base = Instant::now() - Duration::from_millis(500); let mut states = HashMap::new(); - for (node_id, arrival) in [ - (1, base), - (2, base + Duration::from_millis(200)), + let mut synced_history = VecDeque::new(); + synced_history.push_back(vec![1.0; 64]); + let mut synced = make_node_state(synced_history, None, 0); + mark_mesh_timed_frame(&mut synced, 1, 100, 101, 500_000_000, base); + states.insert(1, synced); + + let mut unsynced_history = VecDeque::new(); + unsynced_history.push_back(vec![1.1; 64]); + states.insert( + 2, + make_node_state(unsynced_history, Some(base + Duration::from_millis(5)), 0), + ); + + let frames = node_frames_from_states(&states); + let spread = frames.iter().map(|f| f.timestamp_us).max().unwrap() + - frames.iter().map(|f| f.timestamp_us).min().unwrap(); + assert_eq!(spread, 5_000, "partial sync must fall back as one cycle"); + assert!( + MultistaticFuser::new().fuse(&frames).is_ok(), + "mixed sync validity must not mix mesh and host timestamp domains" + ); + } + + #[test] + fn incoherent_mesh_timestamps_fall_back_to_host_arrival_for_the_cycle() { + let base = Instant::now() - Duration::from_millis(500); + let mut states = HashMap::new(); + + for (node_id, mesh_epoch_us, arrival) in [ + (1, 1_000_000, base), + (2, 500_000_000, base + Duration::from_millis(5)), ] { + let mut history = VecDeque::new(); + history.push_back(vec![1.0; 64]); + let mut state = make_node_state(history, None, 0); + mark_mesh_timed_frame(&mut state, node_id, 100, 101, mesh_epoch_us, arrival); + states.insert(node_id, state); + } + + let frames = node_frames_from_states(&states); + let spread = frames.iter().map(|f| f.timestamp_us).max().unwrap() + - frames.iter().map(|f| f.timestamp_us).min().unwrap(); + assert_eq!(spread, 5_000, "incoherent mesh time must not reach fusion"); + assert!(MultistaticFuser::new().fuse(&frames).is_ok()); + } + + #[test] + fn unsynchronized_frames_prune_to_freshest_host_cohort() { + let base = Instant::now() - Duration::from_millis(500); + let mut states = HashMap::new(); + + for (node_id, arrival) in [(1, base), (2, base + Duration::from_millis(200))] { let mut history = VecDeque::new(); history.push_back(vec![1.0; 64]); states.insert(node_id, make_node_state(history, Some(arrival), 0)); } let frames = node_frames_from_states(&states); + assert_eq!(frames.len(), 1, "only the freshest host frame is coherent"); + assert_eq!(frames[0].node_id, 2); assert!( - MultistaticFuser::new().fuse(&frames).is_err(), - "without valid mesh time, 200 ms arrival skew must still trip the 60 ms guard" + MultistaticFuser::new().fuse(&frames).is_ok(), + "an asynchronous slow node must not fail the live cycle" ); } + #[test] + fn slow_live_node_is_excluded_from_fresh_cohort() { + let now = Instant::now(); + let mut states = HashMap::new(); + for (node_id, age_ms, n_sub) in [ + (1, 0, 64), + (3, 10, 256), + (4, 50, 64), + (7, 1_000, 256), + ] { + let mut history = VecDeque::new(); + history.push_back(vec![1.0 + node_id as f64 * 0.01; n_sub]); + states.insert( + node_id, + make_node_state( + history, + Some(now - Duration::from_millis(age_ms)), + 1, + ), + ); + } + + let frames = node_frames_from_states_with_guard(&states, 60_000); + let ids: Vec = frames.iter().map(|frame| frame.node_id).collect(); + assert_eq!(ids, vec![1, 3, 4]); + assert!(MultistaticFuser::new().fuse(&frames).is_ok()); + } + + #[test] + fn configured_guard_is_shared_with_cohort_selection() { + let base = Instant::now() - Duration::from_millis(500); + let mut states = HashMap::new(); + for (node_id, arrival) in [ + (1, base), + (2, base + Duration::from_millis(150)), + ] { + let mut history = VecDeque::new(); + history.push_back(vec![1.0; 64]); + states.insert(node_id, make_node_state(history, Some(arrival), 0)); + } + + let cfg = MultistaticConfig { + guard_interval_us: 200_000, + ..MultistaticConfig::default() + }; + let fuser = MultistaticFuser::with_config(cfg); + let (fused, fallback) = fuse_or_fallback(&fuser, &states, 3.0); + assert_eq!(fused.as_ref().map(|frame| frame.active_nodes), Some(2)); + assert!(fallback.is_none()); + } + #[test] fn heterogeneous_node_counts_canonicalize_and_fuse() { // Issue #1170 regression: a mixed mesh with HT20 (64-bin) and HT40 diff --git a/v2/crates/wifi-densepose-signal/src/ruvsense/multistatic.rs b/v2/crates/wifi-densepose-signal/src/ruvsense/multistatic.rs index d4ca9f89..372056bc 100644 --- a/v2/crates/wifi-densepose-signal/src/ruvsense/multistatic.rs +++ b/v2/crates/wifi-densepose-signal/src/ruvsense/multistatic.rs @@ -272,6 +272,17 @@ impl MultistaticFuser { self.node_positions = positions; } + /// Return the configured hard timestamp guard in microseconds. + /// + /// Callers that assemble frames before invoking [`Self::fuse`] use this + /// to select one temporally coherent sensing cohort. Keeping selection and + /// validation on the same guard prevents stale low-rate nodes from making + /// every otherwise-live fusion cycle fail. + #[must_use] + pub fn guard_interval_us(&self) -> u64 { + self.config.guard_interval_us + } + /// Return the current node positions. pub fn node_positions(&self) -> &[[f32; 3]] { &self.node_positions