From 8c6d29bf830c877641fb62bc133147eafa341913 Mon Sep 17 00:00:00 2001 From: Mathew Beane Date: Fri, 2 Oct 2026 22:15:23 -0400 Subject: [PATCH] fix(sensing-server): report measured stream fps and real WebSocket client count /api/v1/stream/status returned "fps": 10 whenever tick > 1, whatever the real rate, and both it and /health counted every broadcast subscriber (recorder, MQTT bridge) as a client. The dashboard Streaming card read "IDLE, 0 client(s)" for up to 30 s while the page was streaming, because its first /health poll lands before the page's socket connects; with auth on, the ticket round-trip widens that gap. - New stream_stats module: UpdateRateMeter (sliding 5 s window, bounded at 1024 samples) and WsClientCounter with an RAII guard. - AppStateInner::advance_tick() replaces the five `s.tick += 1` sites and records each sensing update in the meter. - /api/v1/stream/status now returns the measured update rate as `fps` (0 when nothing flows), `active` from that rate, `clients` from the WebSocket counter, `nodes` with each fresh node's measured csi_fps (null until its EMA has 5 samples) and `csi_fps_total`. - /health and the /health/health stream component count WebSocket clients on /ws/sensing and /api/v1/stream/pose; the stream component also carries the measured rate. - DashboardTab re-reads health 500 ms after the sensing socket connects or drops, so the card follows the socket instead of waiting for the next 30 s poll. Fixes #2087 Co-Authored-By: claude-flow --- .github/workflows/ci.yml | 2 +- ui/components/DashboardTab.js | 23 +- ui/components/DashboardTab.test.mjs | 70 ++++++ .../wifi-densepose-sensing-server/src/lib.rs | 2 + .../wifi-densepose-sensing-server/src/main.rs | 204 ++++++++++++++++-- .../src/stream_stats.rs | 187 ++++++++++++++++ 6 files changed, 464 insertions(+), 24 deletions(-) create mode 100644 ui/components/DashboardTab.test.mjs create mode 100644 v2/crates/wifi-densepose-sensing-server/src/stream_stats.rs diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 32cc2a6a7..4e46997e3 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -204,7 +204,7 @@ jobs: node-version: '22' - name: Run UI unit tests - run: node --test ui/sw.test.mjs ui/services/ws-ticket.test.mjs ui/services/websocket.service.test.mjs v2/crates/wifi-densepose-desktop/ui/build-config.test.mjs + run: node --test ui/sw.test.mjs ui/services/ws-ticket.test.mjs ui/services/websocket.service.test.mjs ui/components/DashboardTab.test.mjs v2/crates/wifi-densepose-desktop/ui/build-config.test.mjs # Unit and Integration Tests # Python pytest matrix — runs against the archived v1 Python tree. diff --git a/ui/components/DashboardTab.js b/ui/components/DashboardTab.js index 5e505fa99..b3a541464 100644 --- a/ui/components/DashboardTab.js +++ b/ui/components/DashboardTab.js @@ -5,6 +5,8 @@ import { poseService } from '../services/pose.service.js'; import { sensingService } from '../services/sensing.service.js'; export class DashboardTab { + static STREAM_HEALTH_REFRESH_DELAY_MS = 500; + constructor(containerElement) { this.container = containerElement; this.statsElements = {}; @@ -65,8 +67,9 @@ export class DashboardTab { }); // Subscribe to sensing service state changes for data source indicator - this._sensingUnsub = sensingService.onStateChange(() => { + this._sensingUnsub = sensingService.onStateChange((state) => { this.updateDataSourceIndicator(); + this._refreshStreamHealthOn(state); }); // Also update on data — catches source changes mid-stream this._sensingDataUnsub = sensingService.onData(() => { @@ -84,6 +87,23 @@ export class DashboardTab { healthService.startHealthMonitoring(30000); } + // #2087: the Streaming card comes from /health, polled every 30 s. The first + // poll runs before this page's own socket has connected (with auth on, the + // ticket round-trip makes that gap wider), so the card read "IDLE, 0 + // client(s)" for up to 30 s while data was streaming. Re-read health when + // the socket opens or drops. The short delay lets the server register the + // socket before it is counted. + _refreshStreamHealthOn(state) { + if (state === this._lastSensingState) return; + const wasConnected = this._lastSensingState === 'connected'; + this._lastSensingState = state; + if (state !== 'connected' && !wasConnected) return; + clearTimeout(this._streamHealthTimer); + this._streamHealthTimer = setTimeout(() => { + healthService.getSystemHealth().catch(() => { /* next poll retries */ }); + }, DashboardTab.STREAM_HEALTH_REFRESH_DELAY_MS); + } + // Update the data source indicator on the dashboard updateDataSourceIndicator() { const el = this.container.querySelector('#dashboard-datasource'); @@ -427,6 +447,7 @@ export class DashboardTab { this.healthSubscription(); } if (this._sensingUnsub) this._sensingUnsub(); + clearTimeout(this._streamHealthTimer); if (this._sensingDataUnsub) this._sensingDataUnsub(); if (this.statsInterval) { diff --git a/ui/components/DashboardTab.test.mjs b/ui/components/DashboardTab.test.mjs new file mode 100644 index 000000000..68fcb199f --- /dev/null +++ b/ui/components/DashboardTab.test.mjs @@ -0,0 +1,70 @@ +// Executed regression test for #2087: the dashboard Streaming card read +// "IDLE, 0 client(s)" while the page's own socket was streaming, because the +// card is fed by a 30 s /health poll whose first read lands before the socket +// connects. +// +// Run: node --test ui/components/DashboardTab.test.mjs +// +// This EXECUTES DashboardTab's refresh wiring in Node with fake timers and a +// stubbed health service. It does not render the card in a browser. + +import { test, beforeEach, mock } from 'node:test'; +import assert from 'node:assert/strict'; + +globalThis.localStorage = { getItem: () => null, setItem() {}, removeItem() {} }; +globalThis.fetch = async () => ({ ok: true, status: 200, json: async () => ({}) }); + +const { DashboardTab } = await import('./DashboardTab.js'); +const { healthService } = await import('../services/health.service.js'); + +let healthCalls; +let tab; + +beforeEach(() => { + healthCalls = 0; + healthService.getSystemHealth = async () => { healthCalls++; return {}; }; + mock.timers.enable({ apis: ['setTimeout'] }); + tab = new DashboardTab({ querySelector: () => null }); +}); + +const advance = () => mock.timers.tick(DashboardTab.STREAM_HEALTH_REFRESH_DELAY_MS); + +test('health is re-read once the sensing socket connects', () => { + tab._refreshStreamHealthOn('connecting'); + advance(); + assert.equal(healthCalls, 0, 'nothing to report until the socket is up'); + + tab._refreshStreamHealthOn('connected'); + assert.equal(healthCalls, 0, 'waits for the server to register the socket'); + advance(); + assert.equal(healthCalls, 1); + mock.timers.reset(); +}); + +test('health is re-read when a connected socket drops', () => { + tab._refreshStreamHealthOn('connected'); + advance(); + tab._refreshStreamHealthOn('reconnecting'); + advance(); + assert.equal(healthCalls, 2); + mock.timers.reset(); +}); + +test('repeated notifications of the same state do not refetch', () => { + // onStateChange also fires on dataSource changes with an unchanged state. + tab._refreshStreamHealthOn('connected'); + tab._refreshStreamHealthOn('connected'); + tab._refreshStreamHealthOn('connected'); + advance(); + assert.equal(healthCalls, 1); + mock.timers.reset(); +}); + +test('retry churn while never connected does not poll health', () => { + for (const s of ['connecting', 'reconnecting', 'connecting', 'reconnecting']) { + tab._refreshStreamHealthOn(s); + advance(); + } + assert.equal(healthCalls, 0); + mock.timers.reset(); +}); diff --git a/v2/crates/wifi-densepose-sensing-server/src/lib.rs b/v2/crates/wifi-densepose-sensing-server/src/lib.rs index 0cf0214f1..31102f92a 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/lib.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/lib.rs @@ -38,6 +38,8 @@ pub mod rufield_surface; pub mod rvf_container; pub mod rvf_pipeline; pub mod semconv; +/// #2087: measured stream rate and WebSocket client count. +pub mod stream_stats; pub mod telemetry; #[allow(dead_code)] pub mod trainer; diff --git a/v2/crates/wifi-densepose-sensing-server/src/main.rs b/v2/crates/wifi-densepose-sensing-server/src/main.rs index 180ae6f53..57e569998 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/main.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/main.rs @@ -1873,6 +1873,12 @@ struct AppStateInner { // (not replacing) the window-aggregated `tx` / `/ws/sensing` pipeline. intro: wifi_densepose_sensing_server::introspection::IntrospectionState, intro_tx: broadcast::Sender, + /// #2087: browser WebSocket clients currently connected. `tx` receivers + /// also include the recorder and the MQTT bridge, so they are not a count + /// of clients. + ws_clients: wifi_densepose_sensing_server::stream_stats::WsClientCounter, + /// #2087: measured rate of sensing updates (one per `tick` advance). + update_rate: wifi_densepose_sensing_server::stream_stats::UpdateRateMeter, total_detections: u64, start_time: std::time::Instant, /// Vital sign detector (processes CSI frames to estimate HR/RR). @@ -2157,6 +2163,14 @@ fn calibration_sequence_order(previous: Option, sequence: u32) -> Calibrati } impl AppStateInner { + /// Advance the sensing tick and record the update for the measured + /// stream rate (#2087). Returns the new tick. + fn advance_tick(&mut self) -> u64 { + self.tick += 1; + self.update_rate.observe(std::time::Instant::now()); + self.tick + } + /// Admit at most one raw CSI observation per forward node sequence into /// the current calibration session. This stateful boundary prevents a /// caller from advancing the model with a replayed history tail. @@ -2577,6 +2591,8 @@ impl AppStateInner { tx: broadcast::channel::(16).0, intro: wifi_densepose_sensing_server::introspection::IntrospectionState::new(), intro_tx: broadcast::channel::(16).0, + ws_clients: Default::default(), + update_rate: Default::default(), total_detections: 0, start_time: std::time::Instant::now(), vital_detector: VitalSignDetector::new(10.0), @@ -4289,8 +4305,7 @@ async fn wifi_task(state: SharedState, tick_ms: u64) { s.rssi_history.pop_front(); } - s.tick += 1; - let tick = s.tick; + let tick = s.advance_tick(); let motion_score = if classification.motion_level == "active" { 0.8 @@ -4455,8 +4470,7 @@ async fn windows_wifi_fallback_tick(state: &SharedState, seq: u32) { s.rssi_history.pop_front(); } - s.tick += 1; - let tick = s.tick; + let tick = s.advance_tick(); let motion_score = if classification.motion_level == "active" { 0.8 @@ -4838,9 +4852,9 @@ async fn ws_sensing_handler( } async fn handle_ws_client(mut socket: WebSocket, state: SharedState) { - let mut rx = { + let (mut rx, _client) = { let s = state.read().await; - s.tx.subscribe() + (s.tx.subscribe(), s.ws_clients.enter()) }; info!("WebSocket client connected (sensing)"); @@ -4947,9 +4961,9 @@ async fn ws_pose_handler( } async fn handle_ws_pose_client(mut socket: WebSocket, state: SharedState) { - let mut rx = { + let (mut rx, _client) = { let s = state.read().await; - s.tx.subscribe() + (s.tx.subscribe(), s.ws_clients.enter()) }; info!("WebSocket client connected (pose)"); @@ -5101,7 +5115,7 @@ async fn health(State(state): State) -> Json { "status": "ok", "source": s.effective_source(), "tick": s.tick, - "clients": s.tx.receiver_count(), + "clients": s.ws_clients.get(), })) } @@ -6567,8 +6581,7 @@ async fn health_system(State(state): State) -> Json 0 { "healthy" } else { "idle" }, - "message": format!("{} client(s)", s.tx.receiver_count()) }, + "stream": stream_health_component(&s, std::time::Instant::now()), }, "metrics": { "cpu_percent": 2.5, @@ -6728,12 +6741,160 @@ async fn pose_zones_summary(State(state): State) -> Json) -> Json { let s = state.read().await; - Json(serde_json::json!({ - "active": true, - "clients": s.tx.receiver_count(), - "fps": if s.tick > 1 { 10u64 } else { 0u64 }, + Json(stream_status_json(&s, std::time::Instant::now())) +} + +/// #2087: measured stream status. `fps` is the rate of sensing updates over +/// the last few seconds (0 when nothing is flowing), `nodes` carries each +/// fresh node's measured CSI frame rate, and `clients` counts browser +/// WebSockets rather than every internal broadcast subscriber. +fn stream_status_json(s: &AppStateInner, now: std::time::Instant) -> serde_json::Value { + use wifi_densepose_sensing_server::stream_stats::round1; + let fps = s.update_rate.rate_hz(now); + let mut ids: Vec = s + .node_states + .iter() + .filter(|(_, n)| node_is_fresh(n, now)) + .map(|(id, _)| *id) + .collect(); + ids.sort_unstable(); + let nodes: Vec = ids + .iter() + .map(|id| { + let n = &s.node_states[id]; + serde_json::json!({ + "node_id": id, + // null until the node's EMA has warmed up (5 frames). + "csi_fps": n.measured_sample_rate_hz().map(round1), + "csi_fps_samples": n.csi_fps_samples, + }) + }) + .collect(); + // fold from +0.0: an empty f64 `sum()` is -0.0, which serializes as "-0.0". + let csi_fps_total = ids + .iter() + .filter_map(|id| s.node_states[id].measured_sample_rate_hz()) + .fold(0.0, |acc, fps| acc + fps); + serde_json::json!({ + "active": fps > 0.0, + "clients": s.ws_clients.get(), + "fps": round1(fps), + "csi_fps_total": round1(csi_fps_total), + "nodes": nodes, "source": s.effective_source(), - })) + }) +} + +/// `stream` component of `/health/system`: healthy while updates are flowing +/// to at least one browser client. +fn stream_health_component(s: &AppStateInner, now: std::time::Instant) -> serde_json::Value { + let clients = s.ws_clients.get(); + let fps = wifi_densepose_sensing_server::stream_stats::round1(s.update_rate.rate_hz(now)); + let status = if clients > 0 && fps > 0.0 { + "healthy" + } else { + "idle" + }; + serde_json::json!({ + "status": status, + "message": format!("{clients} client(s), {fps} updates/s"), + "clients": clients, + "fps": fps, + }) +} + +#[cfg(test)] +mod stream_status_tests { + //! #2087: `/api/v1/stream/status` reported a literal 10 fps and counted + //! every broadcast subscriber as a client. + + use super::*; + + fn warmed_node(fps: f64, seen: std::time::Instant) -> NodeState { + let mut n = NodeState::new(); + n.csi_fps_ema = fps; + n.csi_fps_samples = 50; + n.last_frame_time = Some(seen); + n + } + + #[test] + fn idle_server_reports_zero_fps_and_no_clients() { + let s = AppStateInner::minimal(); + let _recorder = s.tx.subscribe(); // internal subscriber, not a client + let v = stream_status_json(&s, std::time::Instant::now()); + assert_eq!(v["fps"], 0.0); + assert_eq!(v["active"], false); + assert_eq!(v["clients"], 0); + assert_eq!(v["nodes"].as_array().unwrap().len(), 0); + assert_eq!( + stream_health_component(&s, std::time::Instant::now())["status"], + "idle" + ); + } + + #[test] + fn fps_is_the_measured_update_rate() { + let mut s = AppStateInner::minimal(); + let t0 = std::time::Instant::now(); + for i in 0..=40u64 { + s.tick += 1; + s.update_rate.observe(t0 + Duration::from_millis(50 * i)); + } + let v = stream_status_json(&s, t0 + Duration::from_millis(2000)); + assert_eq!(v["fps"], 20.0, "a 50 ms cadence must read 20, not 10"); + assert_eq!(v["active"], true); + } + + #[test] + fn per_node_fps_and_aggregate_cover_fresh_nodes_only() { + let mut s = AppStateInner::minimal(); + let now = std::time::Instant::now(); + s.node_states.insert(2, warmed_node(46.0, now)); + s.node_states.insert(1, warmed_node(41.0, now)); + s.node_states.insert( + 3, + warmed_node(30.0, now - Duration::from_millis(NODE_STALE_AFTER_MS + 1)), + ); + let mut cold = NodeState::new(); + cold.last_frame_time = Some(now); + s.node_states.insert(4, cold); + + let v = stream_status_json(&s, now); + let nodes = v["nodes"].as_array().unwrap(); + let ids: Vec = nodes + .iter() + .map(|n| n["node_id"].as_u64().unwrap()) + .collect(); + assert_eq!(ids, vec![1, 2, 4], "stale node 3 excluded, sorted by id"); + assert_eq!(nodes[0]["csi_fps"], 41.0); + assert_eq!(nodes[1]["csi_fps"], 46.0); + assert!( + nodes[2]["csi_fps"].is_null(), + "unwarmed EMA is not reported" + ); + assert_eq!(v["csi_fps_total"], 87.0); + } + + #[test] + fn clients_count_websockets_not_internal_subscribers() { + let mut s = AppStateInner::minimal(); + let _recorder = s.tx.subscribe(); + let _mqtt = s.tx.subscribe(); + let browser = s.ws_clients.enter(); + let now = std::time::Instant::now(); + for i in 0..10u64 { + s.update_rate + .observe(now - Duration::from_millis(100 * (10 - i))); + } + assert_eq!(stream_status_json(&s, now)["clients"], 1); + let health = stream_health_component(&s, now); + assert_eq!(health["status"], "healthy"); + assert_eq!(health["clients"], 1); + drop(browser); + assert_eq!(stream_status_json(&s, now)["clients"], 0); + assert_eq!(stream_health_component(&s, now)["status"], "idle"); + } } // ── Model Management Endpoints ────────────────────────────────────────────── @@ -9890,8 +10051,7 @@ async fn udp_receiver_task( update_edge_node_classification(ns, &vitals); - s.tick += 1; - let tick = s.tick; + let tick = s.advance_tick(); let motion_score = if vitals.motion { 0.8 @@ -10404,8 +10564,7 @@ async fn udp_receiver_task( // Cross-node fusion: combine features from all active nodes. let fused_features = fuse_multi_node_features(&features, &s.node_states); - s.tick += 1; - let tick = s.tick; + let tick = s.advance_tick(); let motion_score = if classification.motion_level == "active" { 0.8 @@ -10709,8 +10868,7 @@ async fn simulated_data_task(state: SharedState, tick_ms: u64) { continue; } - s.tick += 1; - let tick = s.tick; + let tick = s.advance_tick(); let frame = generate_simulated_frame(tick); @@ -12127,6 +12285,8 @@ async fn main() { tx, intro: wifi_densepose_sensing_server::introspection::IntrospectionState::new(), intro_tx, + ws_clients: Default::default(), + update_rate: Default::default(), total_detections: 0, start_time: std::time::Instant::now(), vital_detector: VitalSignDetector::new(vital_sample_rate), diff --git a/v2/crates/wifi-densepose-sensing-server/src/stream_stats.rs b/v2/crates/wifi-densepose-sensing-server/src/stream_stats.rs new file mode 100644 index 000000000..da98f70da --- /dev/null +++ b/v2/crates/wifi-densepose-sensing-server/src/stream_stats.rs @@ -0,0 +1,187 @@ +//! Measured streaming statistics for `/api/v1/stream/status` and the +//! `stream` component of `/health/system` (#2087). +//! +//! Two things used to be guessed: the stream rate (a literal 10 fps) and the +//! client count (`broadcast::Sender::receiver_count`, which also counts the +//! recorder and the MQTT bridge and so is not a count of browsers). This module +//! measures both. + +use std::collections::VecDeque; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +/// Window over which the sensing-update rate is averaged. +pub const UPDATE_RATE_WINDOW: Duration = Duration::from_secs(5); + +/// Hard bound on retained timestamps. At 50 fps per node this covers a +/// 5-node mesh for the whole window; beyond that the rate is taken over a +/// shorter span, which is still a measurement and never an allocation spike. +const UPDATE_RATE_MAX_SAMPLES: usize = 1024; + +/// Sliding-window rate of sensing updates (one per `tick` advance). +#[derive(Debug, Default)] +pub struct UpdateRateMeter { + times: VecDeque, +} + +impl UpdateRateMeter { + pub fn new() -> Self { + Self::default() + } + + /// Record one sensing update at `now`. + pub fn observe(&mut self, now: Instant) { + if self.times.len() == UPDATE_RATE_MAX_SAMPLES { + self.times.pop_front(); + } + self.times.push_back(now); + self.prune(now); + } + + fn prune(&mut self, now: Instant) { + while let Some(&oldest) = self.times.front() { + match now.checked_duration_since(oldest) { + Some(age) if age > UPDATE_RATE_WINDOW => { + self.times.pop_front(); + } + _ => break, + } + } + } + + /// Updates per second over the window ending at `now`. Zero when fewer + /// than two updates fall inside the window, so a stalled source reads 0 + /// instead of holding its last rate. + pub fn rate_hz(&self, now: Instant) -> f64 { + let mut in_window = self.times.iter().filter(|t| { + now.checked_duration_since(**t) + .is_some_and(|age| age <= UPDATE_RATE_WINDOW) + }); + let Some(first) = in_window.next() else { + return 0.0; + }; + let (count, last) = in_window.fold((1usize, first), |(n, _), t| (n + 1, t)); + let span = last.duration_since(*first).as_secs_f64(); + if count < 2 || span <= 0.0 { + return 0.0; + } + (count - 1) as f64 / span + } +} + +/// Live count of browser-facing WebSocket clients (`/ws/sensing` and +/// `/api/v1/stream/pose`). Ticketed and unticketed sockets count the same: +/// the count is taken after the upgrade, so it only includes sockets the auth +/// layer admitted. +#[derive(Debug, Clone, Default)] +pub struct WsClientCounter(Arc); + +impl WsClientCounter { + pub fn new() -> Self { + Self::default() + } + + pub fn get(&self) -> usize { + self.0.load(Ordering::Relaxed) + } + + /// Count one client until the returned guard is dropped. + pub fn enter(&self) -> WsClientGuard { + self.0.fetch_add(1, Ordering::Relaxed); + WsClientGuard(self.0.clone()) + } +} + +/// Decrements the client count when the socket task ends, on every exit path. +#[derive(Debug)] +pub struct WsClientGuard(Arc); + +impl Drop for WsClientGuard { + fn drop(&mut self) { + self.0.fetch_sub(1, Ordering::Relaxed); + } +} + +/// Round to one decimal place for JSON output. +pub fn round1(x: f64) -> f64 { + (x * 10.0).round() / 10.0 +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn rate_is_zero_without_samples() { + let m = UpdateRateMeter::new(); + assert_eq!(m.rate_hz(Instant::now()), 0.0); + } + + #[test] + fn rate_measures_a_steady_25_hz_stream() { + let mut m = UpdateRateMeter::new(); + let t0 = Instant::now(); + for i in 0..=100 { + m.observe(t0 + Duration::from_millis(40 * i)); + } + let now = t0 + Duration::from_millis(4000); + let r = m.rate_hz(now); + assert!((r - 25.0).abs() < 0.01, "expected 25 Hz, got {r}"); + } + + #[test] + fn rate_is_not_a_constant() { + // The old endpoint said 10 fps for every source. A 40 ms tick must + // read 25 and a 200 ms tick must read 5. + let t0 = Instant::now(); + let mut fast = UpdateRateMeter::new(); + let mut slow = UpdateRateMeter::new(); + for i in 0..=20 { + fast.observe(t0 + Duration::from_millis(40 * i)); + slow.observe(t0 + Duration::from_millis(200 * i)); + } + assert!((fast.rate_hz(t0 + Duration::from_millis(800)) - 25.0).abs() < 0.01); + assert!((slow.rate_hz(t0 + Duration::from_millis(4000)) - 5.0).abs() < 0.01); + } + + #[test] + fn rate_decays_to_zero_when_the_source_stalls() { + let mut m = UpdateRateMeter::new(); + let t0 = Instant::now(); + for i in 0..50 { + m.observe(t0 + Duration::from_millis(100 * i)); + } + let later = t0 + Duration::from_millis(100 * 49) + UPDATE_RATE_WINDOW * 2; + assert_eq!(m.rate_hz(later), 0.0); + } + + #[test] + fn meter_memory_is_bounded() { + let mut m = UpdateRateMeter::new(); + let t0 = Instant::now(); + for i in 0..(UPDATE_RATE_MAX_SAMPLES as u64 * 3) { + m.observe(t0 + Duration::from_micros(i)); + } + assert!(m.times.len() <= UPDATE_RATE_MAX_SAMPLES); + } + + #[test] + fn client_counter_tracks_guards() { + let c = WsClientCounter::new(); + assert_eq!(c.get(), 0); + let a = c.enter(); + let b = c.clone().enter(); + assert_eq!(c.get(), 2); + drop(a); + assert_eq!(c.get(), 1); + drop(b); + assert_eq!(c.get(), 0); + } + + #[test] + fn round1_rounds_to_one_decimal() { + assert_eq!(round1(24.96), 25.0); + assert_eq!(round1(41.04), 41.0); + } +}