From 1a50162598b84a7cb16ce18b1373caa36f457535 Mon Sep 17 00:00:00 2001 From: Mathew Beane Date: Fri, 2 Oct 2026 19:22:26 -0400 Subject: [PATCH] fix(sensing-server): stop serving a frozen /sensing/latest as current latest_update is only rewritten when a source produces a new update. When every source went quiet, /api/v1/sensing/latest kept returning the last classification indefinitely, with no marker that it was old. A one-off bogus frame could leave "absent 0.97" frozen there. Past 10 s, the endpoint now returns {"status":"stale","age_ms":...,"source":effective_source()} instead of the old payload. Fresh payloads gain an age_ms field. "no data yet" is unchanged. This is separate from #2060, which excludes stale nodes from the room vote: that changes what a new update says, while this bounds how long an old update is served. Developed in whitsentry, a RuView-based project. Co-Authored-By: claude-flow --- .../wifi-densepose-sensing-server/src/main.rs | 114 +++++++++++++++++- 1 file changed, 113 insertions(+), 1 deletion(-) diff --git a/v2/crates/wifi-densepose-sensing-server/src/main.rs b/v2/crates/wifi-densepose-sensing-server/src/main.rs index 180ae6f53..c6e60cfc2 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/main.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/main.rs @@ -5105,14 +5105,126 @@ async fn health(State(state): State) -> Json { })) } +/// How old `latest_update` may be before `/api/v1/sensing/latest` stops +/// serving it as current. `latest_update` is only rewritten when a source +/// produces a new update, so when every source goes quiet (frames stop, a +/// receiver is unplugged, a bogus node's one-off frame was the last write) +/// the endpoint kept serving the last classification indefinitely, with +/// nothing marking it old. `effective_source()` already reports an offline +/// source after `ESP32_OFFLINE_TIMEOUT`; this bounds the payload itself. +const LATEST_UPDATE_STALE_AFTER: Duration = Duration::from_secs(10); + async fn latest(State(state): State) -> Json { let s = state.read().await; match &s.latest_update { - Some(update) => Json(serde_json::to_value(update).unwrap_or_default()), + Some(update) => { + // Every SensingUpdate is stamped with wall-clock seconds at + // creation, so its age is measured against the same clock. + let now_ms = chrono::Utc::now().timestamp_millis() as f64; + let age_ms = (now_ms - update.timestamp * 1000.0).max(0.0); + if age_ms > LATEST_UPDATE_STALE_AFTER.as_millis() as f64 { + return Json(serde_json::json!({ + "status": "stale", + "age_ms": age_ms, + "source": s.effective_source(), + })); + } + let mut value = serde_json::to_value(update).unwrap_or_default(); + if let Some(obj) = value.as_object_mut() { + obj.insert("age_ms".to_string(), serde_json::json!(age_ms)); + } + Json(value) + } None => Json(serde_json::json!({"status": "no data yet"})), } } +#[cfg(test)] +mod latest_update_staleness_tests { + //! `/api/v1/sensing/latest` must age out a frozen update instead of + //! serving it as current forever. + use super::*; + + fn update_at(timestamp: f64) -> SensingUpdate { + SensingUpdate { + msg_type: "sensing_update".to_string(), + timestamp, + source: "esp32".to_string(), + tick: 1, + nodes: vec![], + features: FeatureInfo { + mean_rssi: 0.0, + variance: 0.0, + motion_band_power: 0.0, + breathing_band_power: 0.0, + dominant_freq_hz: 0.0, + change_points: 0, + spectral_power: 0.0, + }, + classification: ClassificationInfo { + motion_level: "absent".to_string(), + presence: false, + confidence: 0.97, + }, + signal_field: SignalField { + grid_size: [0, 0, 0], + values: vec![], + }, + vital_signs: None, + calibrated_presence_evidence: None, + enhanced_motion: None, + enhanced_breathing: None, + posture: None, + signal_quality_score: None, + quality_verdict: None, + bssid_count: None, + pose_keypoints: None, + model_status: None, + persons: None, + estimated_persons: None, + node_features: None, + room_inference: None, + } + } + + fn now_s() -> f64 { + chrono::Utc::now().timestamp_millis() as f64 / 1000.0 + } + + #[tokio::test] + async fn fresh_update_is_served_with_age_ms() { + let state: SharedState = Arc::new(RwLock::new(AppStateInner::minimal())); + state.write().await.latest_update = Some(update_at(now_s())); + let Json(value) = latest(State(state)).await; + assert_eq!(value["classification"]["motion_level"], "absent"); + let age = value["age_ms"] + .as_f64() + .expect("fresh payload carries age_ms"); + assert!(age < 1000.0, "age_ms was {age}"); + } + + #[tokio::test] + async fn update_older_than_cutoff_is_reported_stale_not_served() { + let state: SharedState = Arc::new(RwLock::new(AppStateInner::minimal())); + let old = now_s() - (LATEST_UPDATE_STALE_AFTER.as_secs_f64() + 60.0); + state.write().await.latest_update = Some(update_at(old)); + let Json(value) = latest(State(state)).await; + assert_eq!(value["status"], "stale"); + assert!(value["age_ms"].as_f64().unwrap() > LATEST_UPDATE_STALE_AFTER.as_millis() as f64); + assert!( + value.get("classification").is_none(), + "a frozen classification must not be served as current" + ); + } + + #[tokio::test] + async fn no_update_yet_is_unchanged() { + let state: SharedState = Arc::new(RwLock::new(AppStateInner::minimal())); + let Json(value) = latest(State(state)).await; + assert_eq!(value["status"], "no data yet"); + } +} + async fn latest_realtek_radar(State(state): State) -> Json { let s = state.read().await; match &s.latest_realtek_radar {