smolkik_adm/RuView
1
0
Fork 0
mirror of https://github.com/ruvnet/RuView.git synced 2026-08-26 02:04:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: align multistatic CSI time and clear Rust advisories (#1669)

Browse Source 
Use mesh-aligned capture timestamps, remediate Rust advisories, harden the audit gate, and correct deployment claims. Includes the live MQTT subscriber lifetime fix verified against Mosquitto.
This commit is contained in:
rUv
2026-08-22 14:59:22 -04:00
committed by GitHub
parent a3b6e1d500
commit f3c361efd1
22 changed files with 579 additions and 908 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
v2/.cargo/audit.toml
View File

@@ -1,13 +1,9 @@
# cargo-audit configuration — v2 workspace
# Managed by security audit (fix/security-audit-rustsec-clippy branch).
#
# This file suppresses advisories in two categories:
# A) CVE-bearing advisories in TRANSITIVE deps we cannot upgrade directly
# because the parent published crate (ruvector-core 2.2.0) has not yet
# published a version with the fix. These are tracked as issues.
# B) UNMAINTAINED-only advisories (no CVE) flowing through dependencies
# that are purely transitive / build-time and have no user-facing attack
# surface in this workspace.
# This file suppresses UNMAINTAINED-only advisories (no CVE) flowing through
# dependencies that are purely transitive / build-time and have no
# user-facing attack surface in this workspace.
# Each entry documents the root cause and the mitigation path.
[advisories]
@@ -24,26 +20,6 @@
# Mitigation: Accept transitively until Tauri v2 drops GTK3 or a workspace
# override path becomes available.
ignore = [
# -----------------------------------------------------------------------
# CATEGORY A — transitive CVEs from ruvector-core 2.2.0 → reqwest 0.11
# ruvector-core 2.2.0 (latest on crates.io) depends on reqwest 0.11.27,
# which pulls in rustls 0.21 / rustls-webpki 0.101.7. We cannot upgrade
# this without a new ruvector-core release. Tracked in issue #812.
# The workspace's own TLS stack uses rustls-webpki 0.103.13 (patched);
# the vulnerable 0.101.7 instance is not reachable from our TLS code.
"RUSTSEC-2026-0098", # rustls-webpki 0.101.7: URI name constraint bypass
"RUSTSEC-2026-0099", # rustls-webpki 0.101.7: wildcard name constraint bypass
"RUSTSEC-2026-0104", # rustls-webpki 0.101.7: reachable panic in CRL parsing
# quinn-proto 0.11.13 is also pulled through midstreamer-quic 0.3 (now
# upgraded). The remaining 0.11.13 instance comes from the same
# ruvector-core transitive chain. Tracked in issue #812.
"RUSTSEC-2026-0037", # quinn-proto 0.11.13: DoS in Quinn endpoints
# CRL Distribution Point matching bug — same ruvector-core / reqwest 0.11
# transitive chain; rustls-webpki 0.101.7 also affected.
"RUSTSEC-2026-0049", # rustls-webpki <0.103.10: CRL authority matching
# -----------------------------------------------------------------------
# CATEGORY B — unmaintained / no CVE
"RUSTSEC-2024-0411", # gdkwayland-sys: unmaintained
"RUSTSEC-2024-0412", # gdk: unmaintained
"RUSTSEC-2024-0413", # atk: unmaintained

987
v2/Cargo.lock generated
View File

File diff suppressed because it is too large Load Diff

2
v2/Cargo.toml
View File

@@ -255,7 +255,7 @@ midstreamer-attractor = "0.2"
# ruvector integration (published on crates.io)
# Vendored at origin/main (a083bd77f) in vendor/ruvector; using crates.io versions
# until published. Bumps per ADR-152 §2.6 (2026-06-10 vendor sync survey).
ruvector-core = "2.2.0"
ruvector-core = "2.3.0"
ruvector-mincut = "2.0.6"
ruvector-attn-mincut = "2.0.4"
ruvector-temporal-tensor = "2.0.6"

14
v2/crates/homecore-recorder/Cargo.toml
View File

@@ -31,10 +31,16 @@ homecore = { path = "../homecore", version = "0.1.0-alpha.0" }
# Async runtime
tokio = { version = "1", features = ["sync", "rt", "rt-multi-thread", "time", "macros"] }
# SQLite via sqlx — only the lite feature set; no postgres, no tls
sqlx = { version = "0.8.1", default-features = false, features = [
"runtime-tokio-native-tls",
"sqlite",
# SQLite-only SQLx crates, pinned in lockstep because their direct APIs are
# semver-exempt. Depending on the umbrella `sqlx` package also resolves its
# unused MySQL backend (and vulnerable `rsa`) into Cargo.lock.
sqlx-core = { version = "=0.8.6", default-features = false, features = [
"_rt-tokio",
"chrono",
"uuid",
] }
sqlx-sqlite = { version = "=0.8.6", default-features = false, features = [
"bundled",
"chrono",
"uuid",
] }

13
v2/crates/homecore-recorder/src/db.rs
View File

@@ -26,6 +26,19 @@ use homecore::StateMachine;
use crate::dedup::fnv64a_hash;
use crate::schema::ALL_DDL;
// Preserve the narrow `sqlx::*` call surface used in this module while
// depending only on SQLx core + SQLite. The umbrella crate resolves unused
// database backends into Cargo.lock, including MySQL's vulnerable RSA stack.
mod sqlx {
pub use sqlx_core::error::Error;
pub use sqlx_core::query::query;
pub use sqlx_core::query_as::query_as;
pub mod sqlite {
pub use sqlx_sqlite::{SqliteConnectOptions, SqlitePool, SqlitePoolOptions};
}
}
type SearchStateRecord = (
i64,
String,

2
v2/crates/ruview-swarm

Submodule v2/crates/ruview-swarm updated: 267aba5be2...5cc4b8625f

2
v2/crates/wifi-densepose-bfld/Cargo.toml
View File

@@ -39,7 +39,7 @@ serde = { workspace = true, features = ["derive"], optional = true }
serde_json = { workspace = true, optional = true }
# MQTT publisher backend (optional). Matches the `rumqttc` choice already in
# `wifi-densepose-sensing-server` so both crates share TLS / version posture.
rumqttc = { version = "0.24", default-features = false, features = ["use-rustls"], optional = true }
rumqttc = { package = "rumqttc-v4-next", version = "0.34", default-features = false, features = ["use-rustls-ring"], optional = true }
wifi-veil = { workspace = true, optional = true }
[dev-dependencies]

2
v2/crates/wifi-densepose-bfld/src/ha_discovery.rs
View File

@@ -31,7 +31,7 @@ use crate::PrivacyClass;
/// };
/// use rumqttc::MqttOptions;
///
/// let opts = MqttOptions::new("seed-01", "broker.local", 1883);
/// let opts = MqttOptions::new("seed-01", ("broker.local", 1883));
/// let (retained_pub, _conn) = RumqttPublisher::connect(opts.clone(), 64);
/// let mut retained_pub = retained_pub.with_retain(true);
/// publish_discovery(&mut retained_pub, "seed-01", PrivacyClass::Anonymous)?;

14
v2/crates/wifi-densepose-bfld/src/rumqttc_publisher.rs
View File

@@ -13,7 +13,7 @@
//! use wifi_densepose_bfld::{publish_event, RumqttPublisher};
//! use rumqttc::MqttOptions;
//!
//! let opts = MqttOptions::new("seed-01", "broker.local", 1883);
//! let opts = MqttOptions::new("seed-01", ("broker.local", 1883));
//! let (mut publisher, mut connection) = RumqttPublisher::connect(opts, 100);
//! thread::spawn(move || for _ in connection.iter() { /* drain */ });
//! // ... build BfldEvent ...
@@ -22,7 +22,7 @@
#![cfg(feature = "mqtt")]
use rumqttc::{Client, Connection, LastWill, MqttOptions, QoS};
use rumqttc::{Client, Connection, LastWill, MqttOptions, PublishOptions, QoS};
use crate::availability::{availability_topic, PAYLOAD_NOT_AVAILABLE};
use crate::mqtt_topics::{Publish, TopicMessage};
@@ -60,7 +60,7 @@ impl RumqttPublisher {
/// shown in the module-level doc example).
#[must_use]
pub fn connect(opts: MqttOptions, capacity: usize) -> (Self, Connection) {
let (client, connection) = Client::new(opts, capacity);
let (client, connection) = Client::builder(opts).capacity(capacity).build();
(Self::new(client, QoS::AtLeastOnce), connection)
}
@@ -87,7 +87,7 @@ impl RumqttPublisher {
/// opt in to the LWT without using `connect_with_lwt`.
#[must_use]
pub fn with_lwt(mut opts: MqttOptions, node_id: &str) -> MqttOptions {
// rumqttc 0.24 LastWill::new takes (topic, message, qos, retain).
// LastWill::new takes (topic, message, qos, retain).
// retain = true so HA sees "offline" on next start even if the session
// dropped while HA was down.
let will = LastWill::new(
@@ -105,6 +105,10 @@ impl Publish for RumqttPublisher {
fn publish(&mut self, msg: &TopicMessage) -> Result<(), Self::Error> {
self.client
.publish(&msg.topic, self.qos, self.retain, msg.payload.as_bytes())
.publish(
&msg.topic,
msg.payload.as_bytes(),
PublishOptions::new(self.qos).retain(self.retain),
)
}
}

25
v2/crates/wifi-densepose-bfld/tests/mosquitto_integration.rs
View File

@@ -23,9 +23,7 @@ use std::thread;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use rumqttc::{Client, Event, Incoming, MqttOptions, Packet, QoS};
use wifi_densepose_bfld::{
publish_event, BfldEvent, PrivacyClass, RumqttPublisher,
};
use wifi_densepose_bfld::{publish_event, BfldEvent, PrivacyClass, RumqttPublisher};
const SUBSCRIBE_TIMEOUT: Duration = Duration::from_secs(5);
const RECEIVE_TIMEOUT: Duration = Duration::from_secs(10);
@@ -69,9 +67,9 @@ fn spawn_subscriber(
port: u16,
topic_filter: &str,
) -> (Receiver<(String, String)>, Receiver<()>) {
let mut opts = MqttOptions::new(unique_client_id("bfld-sub"), host, port);
opts.set_keep_alive(Duration::from_secs(5));
let (client, mut connection) = Client::new(opts, 64);
let mut opts = MqttOptions::new(unique_client_id("bfld-sub"), (host, port));
opts.set_keep_alive(5);
let (client, mut connection) = Client::builder(opts).capacity(64).build();
client
.subscribe(topic_filter, QoS::AtLeastOnce)
.expect("subscribe enqueue");
@@ -79,13 +77,18 @@ fn spawn_subscriber(
let (incoming_tx, incoming_rx) = channel();
let (suback_tx, suback_rx) = channel();
thread::spawn(move || {
// rumqttc-v4-next stops the connection once every request sender is
// dropped. Keep the subscriber client alive for as long as its pump
// thread runs; otherwise the broker sees a clean disconnect directly
// after SUBACK and no subsequent publications can be delivered.
let _client_guard = client;
for notification in connection.iter() {
match notification {
Ok(Event::Incoming(Packet::SubAck(_))) => {
let _ = suback_tx.send(());
}
Ok(Event::Incoming(Incoming::Publish(p))) => {
let topic = p.topic.clone();
let topic = String::from_utf8_lossy(&p.topic).to_string();
let payload = String::from_utf8_lossy(&p.payload).to_string();
if incoming_tx.send((topic, payload)).is_err() {
break;
@@ -141,8 +144,8 @@ fn live_broker_anonymous_event_roundtrips_all_six_topics() {
// Publisher with its own connection. Spawn a thread iterating the
// Connection so publishes actually reach the broker.
let mut opts = MqttOptions::new(unique_client_id("bfld-pub"), &host, port);
opts.set_keep_alive(Duration::from_secs(5));
let mut opts = MqttOptions::new(unique_client_id("bfld-pub"), (host.as_str(), port));
opts.set_keep_alive(5);
let (mut publisher, mut pub_connection) = RumqttPublisher::connect(opts, 64);
thread::spawn(move || {
for _ in pub_connection.iter() { /* drain protocol events */ }
@@ -197,8 +200,8 @@ fn live_broker_restricted_event_omits_identity_risk() {
.recv_timeout(SUBSCRIBE_TIMEOUT)
.expect("SubAck within 5s");
let mut opts = MqttOptions::new(unique_client_id("bfld-pub-r"), &host, port);
opts.set_keep_alive(Duration::from_secs(5));
let mut opts = MqttOptions::new(unique_client_id("bfld-pub-r"), (host.as_str(), port));
opts.set_keep_alive(5);
let (mut publisher, mut pub_connection) = RumqttPublisher::connect(opts, 64);
thread::spawn(move || for _ in pub_connection.iter() {});
thread::sleep(Duration::from_millis(200));

4
v2/crates/wifi-densepose-bfld/tests/rumqttc_lwt.rs
View File

@@ -9,7 +9,7 @@ use wifi_densepose_bfld::{
};
fn unreachable_opts(client_id: &str) -> MqttOptions {
MqttOptions::new(client_id, "127.0.0.1", 1)
MqttOptions::new(client_id, ("127.0.0.1", 1))
}
#[test]
@@ -89,7 +89,7 @@ fn caller_built_options_can_opt_in_via_with_lwt_then_pass_to_connect() {
// Operators with custom MqttOptions (e.g., TLS, credentials) build their
// own opts, then call with_lwt before passing to RumqttPublisher::connect.
let mut opts = unreachable_opts("bfld-lwt-6");
opts.set_keep_alive(std::time::Duration::from_secs(30));
opts.set_keep_alive(30);
let opts = with_lwt(opts, "seed-01");
let (_publisher, _connection) = RumqttPublisher::connect(opts, 16);
}

2
v2/crates/wifi-densepose-bfld/tests/rumqttc_publisher_smoke.rs
View File

@@ -12,7 +12,7 @@ use wifi_densepose_bfld::{publish_event, BfldEvent, PrivacyClass, Publish, Rumqt
fn unreachable_opts() -> MqttOptions {
// Port 1 is reserved (RFC 1700) and the loopback address will refuse
// immediately — perfect for a construction smoke test that must not block.
MqttOptions::new("bfld-smoke-iter23", "127.0.0.1", 1)
MqttOptions::new("bfld-smoke-iter23", ("127.0.0.1", 1))
}
fn sample_event() -> BfldEvent {

2
v2/crates/wifi-densepose-sensing-server/Cargo.toml
View File

@@ -110,7 +110,7 @@ rand = "0.8"
# client (ADR-115 §10 references). `rustls` is preferred over openssl on
# Windows to keep parity with the rest of the workspace (`ureq` above also
# uses rustls).
rumqttc = { version = "0.24", default-features = false, features = ["use-rustls"], optional = true }
rumqttc = { package = "rumqttc-v4-next", version = "0.34", default-features = false, features = ["use-rustls-ring"], optional = true }
# `otel` feature — OTLP log export (`telemetry` module). Same gating
# principle as `mqtt`: the heavy exporter stack (opentelemetry SDK +

92
v2/crates/wifi-densepose-sensing-server/src/main.rs
View File

@@ -286,6 +286,9 @@ struct Esp32Frame {
/// ADR-110 byte 18: PPDU type the CSI was sampled from. Pre-ADR-110
/// firmware sends 0 ⇒ `PpduType::HtLegacy`.
ppdu_type: wifi_densepose_hardware::PpduType,
/// ADR-110 byte 19 metadata, including whether this frame was captured
/// while the node had a valid IEEE 802.15.4 mesh-time solution.
adr018_flags: wifi_densepose_hardware::Adr018Flags,
amplitudes: Vec<f64>,
phases: Vec<f64>,
}
@@ -675,6 +678,12 @@ struct NodeState {
latest_sync: Option<wifi_densepose_hardware::SyncPacket>,
/// Last time a sync packet from this node was received (for staleness).
latest_sync_at: Option<std::time::Instant>,
/// Sequence number of the newest CSI frame admitted to `frame_history`.
/// Kept alongside the history so multistatic fusion can timestamp the
/// exact sample it consumes, rather than the host's UDP arrival time.
latest_csi_sequence: Option<u32>,
/// Whether byte 19 bit 4 marked that newest admitted CSI frame as synced.
latest_csi_sync_valid: bool,
/// ADR-110 iter 18: EMA-tracked CSI frame rate for this node.
/// Replaces the hardcoded 20 Hz fallback in
/// `mesh_aligned_us_for_csi_frame` once `csi_fps_samples ≥ 5`.
@@ -832,6 +841,9 @@ impl NodeState {
/// staleness gate).
pub(crate) fn mesh_aligned_us(&self, local_at_frame_us: u64) -> Option<u64> {
let sync = self.latest_sync.as_ref()?;
if !sync.flags.is_valid {
return None;
}
let seen_at = self.latest_sync_at?;
// Drop stale syncs — firmware emits at ~0.5 Hz default, anything
// older than 9 s likely means the mesh transport dropped.
@@ -850,10 +862,20 @@ impl NodeState {
/// no fresh sync has been observed for this node.
pub(crate) fn mesh_aligned_us_for_csi_frame(&self, frame_sequence: u32) -> Option<u64> {
let sync = self.latest_sync.as_ref()?;
if !sync.flags.is_valid {
return None;
}
let seen_at = self.latest_sync_at?;
if seen_at.elapsed() > std::time::Duration::from_secs(9) {
return None;
}
// A recently-received sync datagram can overtake an older CSI
// datagram in UDP delivery order. Only extrapolate forward (including
// a genuine u32 wrap); otherwise fall back to host arrival time.
let delta_frames = frame_sequence.wrapping_sub(sync.sequence);
if delta_frames > i32::MAX as u32 {
return None;
}
// Iter 18: use the measured per-node fps once we have ≥5 inter-frame
// samples; until then fall back to the 20 Hz firmware ceiling. The
// §A0.12 capture showed real bench fps ≈ 10, so the measured value
@@ -862,6 +884,16 @@ impl NodeState {
Some(sync.mesh_aligned_us_for_sequence(frame_sequence, fps))
}
/// Mesh timestamp for the newest CSI frame admitted to `frame_history`.
/// Both the frame-level sync-valid bit and a fresh, valid sync packet are
/// required; callers retain their existing host-arrival fallback.
pub(crate) fn mesh_aligned_us_for_latest_csi_frame(&self) -> Option<u64> {
if !self.latest_csi_sync_valid {
return None;
}
self.mesh_aligned_us_for_csi_frame(self.latest_csi_sequence?)
}
/// ADR-110 iter 18 — update the per-node observed-fps EMA from a fresh
/// CSI frame arrival. Call once per accepted CSI frame from
/// `udp_receiver_task`. Uses `last_frame_time` as the previous-frame
@@ -927,6 +959,21 @@ impl NodeState {
first_sensing_frame
}
/// Record an accepted CSI sample and preserve the wire metadata needed by
/// the multistatic bridge to recover capture time. Grid-rejected frames
/// intentionally use `observe_csi_frame_arrival` directly because they do
/// not replace the sample at the back of `frame_history`.
pub(crate) fn observe_accepted_csi_frame(
&mut self,
sequence: u32,
sync_valid: bool,
now: std::time::Instant,
) -> bool {
self.latest_csi_sequence = Some(sequence);
self.latest_csi_sync_valid = sync_valid;
self.observe_csi_frame_arrival(now)
}
pub(crate) fn new() -> Self {
Self {
frame_history: VecDeque::new(),
@@ -951,6 +998,8 @@ impl NodeState {
edge_vitals: None,
latest_sync: None,
latest_sync_at: None,
latest_csi_sequence: None,
latest_csi_sync_valid: false,
csi_fps_ema: 20.0,
csi_fps_samples: 0,
latest_features: None,
@@ -1945,7 +1994,8 @@ fn parse_esp32_frame(buf: &[u8]) -> Option<Esp32Frame> {
// [12..15] sequence (u32 LE)
// [16] rssi (i8)
// [17] noise_floor (i8)
// [18..19] reserved
// [18] PPDU type
// [19] ADR-018 flags (bit 4 = IEEE 802.15.4 sync valid)
// [20..] I/Q data
// Issue #1005: until 2026-06 this code read n_subcarriers from byte 6
// alone (an ESP32-C6 HE-SU frame's 256 = 0x0100 LE decoded as 0 — the
@@ -1966,6 +2016,7 @@ fn parse_esp32_frame(buf: &[u8]) -> Option<Esp32Frame> {
};
let noise_floor = buf[17] as i8;
let ppdu_type = wifi_densepose_hardware::PpduType::from_byte(buf[18]);
let adr018_flags = wifi_densepose_hardware::Adr018Flags::from_byte(buf[19]);
let iq_start = 20;
let n_pairs = n_antennas as usize * n_subcarriers as usize;
@@ -1995,6 +2046,7 @@ fn parse_esp32_frame(buf: &[u8]) -> Option<Esp32Frame> {
rssi,
noise_floor,
ppdu_type,
adr018_flags,
amplitudes,
phases,
})
@@ -2024,7 +2076,7 @@ mod issue_1009_n_subcarriers_u16_tests {
buf[16] = (-40i8) as u8; // rssi
buf[17] = (-90i8) as u8; // noise_floor
buf[18] = 0; // ppdu_type
buf[19] = 0;
buf[19] = 0x10; // ADR-018: IEEE 802.15.4 sync valid
for k in 0..n_subcarriers as usize {
buf[20 + k * 2] = (5 + (k % 40) as i8) as u8; // i
buf[20 + k * 2 + 1] = (k % 30) as u8; // q
@@ -2047,6 +2099,7 @@ mod issue_1009_n_subcarriers_u16_tests {
assert_eq!(frame.node_id, 7);
assert_eq!(frame.rssi, -40);
assert_eq!(frame.sequence, 42);
assert!(frame.adr018_flags.ieee802154_sync_valid);
}
#[test]
@@ -2941,6 +2994,7 @@ async fn windows_wifi_task(state: SharedState, tick_ms: u64) {
rssi: first_rssi.clamp(-128.0, 127.0) as i8,
noise_floor: -90,
ppdu_type: wifi_densepose_hardware::PpduType::HtLegacy,
adr018_flags: wifi_densepose_hardware::Adr018Flags::default(),
amplitudes: multi_ap_frame.amplitudes.clone(),
phases: multi_ap_frame.phases.clone(),
};
@@ -3129,6 +3183,7 @@ async fn windows_wifi_fallback_tick(state: &SharedState, seq: u32) {
rssi: rssi_dbm as i8,
noise_floor: -90,
ppdu_type: wifi_densepose_hardware::PpduType::HtLegacy,
adr018_flags: wifi_densepose_hardware::Adr018Flags::default(),
amplitudes: vec![signal_pct],
phases: vec![0.0],
};
@@ -3504,6 +3559,7 @@ fn generate_simulated_frame(tick: u64) -> Esp32Frame {
rssi: (-40.0 + 5.0 * (t * 0.2).sin()) as i8,
noise_floor: -90,
ppdu_type: wifi_densepose_hardware::PpduType::HtLegacy,
adr018_flags: wifi_densepose_hardware::Adr018Flags::default(),
amplitudes,
phases,
}
@@ -6702,8 +6758,11 @@ async fn udp_receiver_task(
// ADR-110 iter 19 — feed the per-node fps EMA from real
// CSI arrivals. The helper sets `last_frame_time` as a
// side effect, so the previous bare assignment is gone.
let first_sensing_frame =
ns.observe_csi_frame_arrival(std::time::Instant::now());
let first_sensing_frame = ns.observe_accepted_csi_frame(
frame.sequence,
frame.adr018_flags.ieee802154_sync_valid,
std::time::Instant::now(),
);
if first_sensing_frame && telemetry::curated_events_enabled() {
info!(name: semconv::EVENT_RUVIEW_NODE_ONLINE, { "ruview.node.id" = node_id }, "node {node_id} online (CSI)");
}
@@ -9327,6 +9386,31 @@ mod sync_snapshot_helper_tests {
"10 s old sync must trigger the 9 s staleness gate");
}
#[test]
fn latest_csi_mesh_time_requires_both_validity_signals() {
let now = std::time::Instant::now();
let mut ns = NodeState::new();
ns.apply_sync_packet(populated_sync(9), now);
ns.observe_accepted_csi_frame(21, false, now);
assert!(
ns.mesh_aligned_us_for_latest_csi_frame().is_none(),
"an unsynchronized CSI capture must use the host-time fallback"
);
ns.observe_accepted_csi_frame(21, true, now + std::time::Duration::from_millis(50));
assert_eq!(
ns.mesh_aligned_us_for_latest_csi_frame(),
Some(27_684_885)
);
ns.latest_sync.as_mut().unwrap().flags.is_valid = false;
assert!(
ns.mesh_aligned_us_for_latest_csi_frame().is_none(),
"an invalid sync packet must not timestamp even a flagged CSI frame"
);
}
#[test]
fn snapshot_reflects_leader_state() {
// Same data shape that /api/v1/mesh emits for a leader node.

40
v2/crates/wifi-densepose-sensing-server/src/mqtt/publisher.rs
View File

@@ -30,7 +30,10 @@
use std::sync::Arc;
use std::time::{Duration, Instant};
use rumqttc::{AsyncClient, ClientError, EventLoop, MqttOptions, QoS, Transport, TlsConfiguration};
use rumqttc::{
AsyncClient, ClientError, EventLoop, MqttOptions, PublishOptions, QoS, Transport,
TlsConfiguration,
};
use tokio::sync::broadcast;
use tokio::task::JoinHandle;
use tracing::{error, info, warn};
@@ -70,14 +73,14 @@ const NODE_SNAPSHOT_STALE_AFTER: Duration = Duration::from_secs(10);
/// Build a `rumqttc::MqttOptions` from validated [`MqttConfig`].
fn build_mqtt_options(cfg: &MqttConfig) -> MqttOptions {
let mut opts = MqttOptions::new(&cfg.client_id, &cfg.host, cfg.port);
opts.set_keep_alive(Duration::from_secs(30));
let mut opts = MqttOptions::new(&cfg.client_id, (cfg.host.as_str(), cfg.port));
opts.set_keep_alive(30);
opts.set_clean_session(true);
if let (Some(u), Some(p)) = (cfg.username.as_deref(), cfg.password.as_deref()) {
opts.set_credentials(u, p);
opts.set_credentials(u.to_owned(), p.as_bytes().to_vec());
} else if let Some(u) = cfg.username.as_deref() {
opts.set_credentials(u, "");
opts.set_credentials(u.to_owned(), Vec::<u8>::new());
}
opts.set_transport(build_transport(&cfg.tls));
@@ -223,7 +226,8 @@ async fn run(
mut state_rx: broadcast::Receiver<VitalsSnapshot>,
) {
let opts = build_mqtt_options(&cfg);
let (client, mut eventloop): (AsyncClient, EventLoop) = AsyncClient::new(opts, 256);
let (client, mut eventloop): (AsyncClient, EventLoop) =
AsyncClient::builder(opts).capacity(256).build();
let entities = DiscoveryBuilder::enabled_entities(
cfg.privacy_mode,
@@ -369,7 +373,13 @@ async fn publish_all_discovery(
let cfg = b.build(e);
let topic = b.config_topic(e);
let payload = serde_json::to_string(&cfg).expect("discovery payload always serialises");
client.publish(&topic, QoS::AtLeastOnce, true, payload).await?;
client
.publish(
&topic,
payload,
PublishOptions::new(QoS::AtLeastOnce).retained(),
)
.await?;
}
Ok(())
}
@@ -380,7 +390,13 @@ async fn publish_availability(
state: &str,
) -> Result<(), ClientError> {
for topic in &avail.online_topics {
client.publish(topic, QoS::AtLeastOnce, true, state).await?;
client
.publish(
topic,
state,
PublishOptions::new(QoS::AtLeastOnce).retained(),
)
.await?;
}
Ok(())
}
@@ -441,7 +457,13 @@ async fn publish_state(client: &AsyncClient, m: &StateMessage) -> Result<(), Cli
1 => QoS::AtLeastOnce,
_ => QoS::ExactlyOnce,
};
client.publish(&m.topic, qos, m.retain, m.payload.clone()).await
client
.publish(
&m.topic,
m.payload.clone(),
PublishOptions::new(qos).retain(m.retain),
)
.await
}
#[cfg(test)]

118
v2/crates/wifi-densepose-sensing-server/src/multistatic_bridge.rs
View File

@@ -24,7 +24,13 @@ const DEFAULT_FREQ_MHZ: u32 = 2437; // Channel 6
/// Monotonic reference point for timestamp generation. All node timestamps
/// are relative to this instant, avoiding wall-clock/monotonic mixing issues.
static EPOCH: LazyLock<Instant> = LazyLock::new(Instant::now);
/// Backdate the lazy initialization beyond the active-node window so frames
/// recorded just before the first bridge call retain their arrival-time skew.
static EPOCH: LazyLock<Instant> = LazyLock::new(|| {
Instant::now()
.checked_sub(STALE_THRESHOLD + STALE_THRESHOLD)
.unwrap_or_else(Instant::now)
});
/// Shared length-only canonicalizer (issue #1170). The default 56-tone grid
/// matches what `MultistaticFuser` (ADR-154) expects. Stateless and immutable,
@@ -54,10 +60,18 @@ pub fn node_frame_from_state(node_id: u8, ns: &NodeState) -> Option<MultiBandCsi
let n_sub = amplitude.len();
let phase = vec![0.0_f32; n_sub];
// Monotonic timestamp: microseconds since a shared process-local epoch.
// All nodes use the same reference so the fuser's guard_interval_us check
// compares apples to apples. No wall-clock mixing (immune to NTP jumps).
let timestamp_us = last_time.duration_since(*EPOCH).as_micros() as u64;
// Prefer the capture timestamp recovered from the node's mesh sync. This
// keeps UDP scheduling jitter out of the fuser's cross-node guard. Older
// firmware, stale sync state, and frames without the sync-valid bit retain
// the process-local host-arrival fallback.
let timestamp_us = ns
.mesh_aligned_us_for_latest_csi_frame()
.unwrap_or_else(|| {
last_time
.checked_duration_since(*EPOCH)
.unwrap_or_default()
.as_micros() as u64
});
let canonical = CanonicalCsiFrame {
amplitude,
@@ -173,6 +187,7 @@ pub fn compute_person_score_from_amplitudes(amplitudes: &[f32]) -> f64 {
mod tests {
use super::*;
use std::collections::VecDeque;
use wifi_densepose_hardware::{SyncPacket, SyncPacketFlags};
/// Helper: build a minimal NodeState for testing. Uses `NodeState::new()`
/// then mutates the `pub(crate)` fields the bridge needs.
@@ -225,6 +240,99 @@ mod tests {
assert_eq!(ch.hardware_type, HardwareType::Esp32S3);
}
fn mark_mesh_timed_frame(
ns: &mut NodeState,
node_id: u8,
sync_sequence: u32,
frame_sequence: u32,
mesh_epoch_us: u64,
host_arrival: Instant,
) {
ns.apply_sync_packet(
SyncPacket {
node_id,
proto_ver: 1,
flags: SyncPacketFlags {
is_leader: node_id == 1,
is_valid: true,
smoothed_used: node_id != 1,
},
local_us: 10_000_000,
epoch_us: mesh_epoch_us,
sequence: sync_sequence,
},
Instant::now(),
);
ns.observe_accepted_csi_frame(frame_sequence, true, host_arrival);
}
#[test]
fn mesh_timestamp_replaces_skewed_host_arrival_time() {
let mut history = VecDeque::new();
history.push_back(vec![10.0, 20.0, 30.0]);
let host_arrival = Instant::now();
let mut ns = make_node_state(history, None, 0);
mark_mesh_timed_frame(&mut ns, 1, 100, 101, 1_000_000, host_arrival);
let frame = node_frame_from_state(1, &ns).expect("mesh-timed frame");
assert_eq!(frame.timestamp_us, 1_050_000);
}
#[test]
fn mesh_time_allows_fusion_despite_udp_arrival_skew() {
let base = Instant::now() - Duration::from_millis(500);
let mut states = HashMap::new();
let mut first_history = VecDeque::new();
first_history.push_back(vec![1.0; 64]);
let mut first = make_node_state(first_history, None, 0);
mark_mesh_timed_frame(&mut first, 1, 100, 101, 1_000_000, base);
states.insert(1, first);
let mut second_history = VecDeque::new();
second_history.push_back(vec![1.1; 64]);
let mut second = make_node_state(second_history, None, 0);
mark_mesh_timed_frame(
&mut second,
2,
200,
201,
1_005_000,
base + Duration::from_millis(200),
);
states.insert(2, second);
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, "mesh capture spread, not 200 ms UDP skew");
assert!(
MultistaticFuser::new().fuse(&frames).is_ok(),
"mesh-aligned frames inside the 60 ms guard must fuse"
);
}
#[test]
fn unsynchronized_frames_keep_host_arrival_guard() {
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!(
MultistaticFuser::new().fuse(&frames).is_err(),
"without valid mesh time, 200 ms arrival skew must still trip the 60 ms guard"
);
}
#[test]
fn heterogeneous_node_counts_canonicalize_and_fuse() {
// Issue #1170 regression: a mixed mesh with HT20 (64-bin) and HT40

13
v2/crates/wifi-densepose-sensing-server/tests/mqtt_integration.rs
View File

@@ -104,12 +104,11 @@ async fn subscribe_client(port: u16, topics: &[&str]) -> (AsyncClient, EventLoop
.unwrap_or(0);
let mut opts = MqttOptions::new(
format!("ruview-test-sub-{}-{}", std::process::id(), suffix),
"127.0.0.1",
port,
("127.0.0.1", port),
);
opts.set_keep_alive(Duration::from_secs(10));
opts.set_keep_alive(10);
opts.set_clean_session(true);
let (client, mut eventloop) = AsyncClient::new(opts, 256);
let (client, mut eventloop) = AsyncClient::builder(opts).capacity(256).build();
for t in topics {
client.subscribe(*t, QoS::AtLeastOnce).await.unwrap();
}
@@ -147,7 +146,11 @@ async fn collect_published(
let remain = until - tokio::time::Instant::now();
match timeout(remain, eventloop.poll()).await {
Ok(Ok(Event::Incoming(Packet::Publish(p)))) => {
out.push((p.topic, p.payload.to_vec(), p.retain));
out.push((
String::from_utf8_lossy(&p.topic).to_string(),
p.payload.to_vec(),
p.retain,
));
}
Ok(Ok(_)) => {} // ignore other events
Ok(Err(e)) => {