sensing-server (mqtt): RateLimiter keyed by EntityKind only — multi-node meshes starve every node except the batch winner

[jkuscevic]

Summary

In wifi-densepose-sensing-server (mqtt feature), the publish RateLimiter is keyed by EntityKind only — one shared instance gates the snapshots of every node plus the aggregate. In a multi-node mesh, the first snapshot of each engine batch consumes the per-entity slot and all other nodes' numeric entities (presence_score, motion_level, motion_energy, person_count, rssi, vitals) are dropped. Because the batch order is fixed, the same node wins for hours; the winner only rotates on restart.

Environment

• Docker image ruvnet/wifi-densepose:v2051 (Rust sensing-server, --features mqtt), host networking
• 5× M5Stack Atom S3 Lite running esp32-csi-node, streaming UDP CSI to one server
• Home Assistant consuming the MQTT discovery entities

Measured symptom (5-node mesh, before fix)

With --mqtt-rate-motion 12 (raised 6× precisely to work around this): a 60 s mosquitto_sub count on +/presence_score/state gave ~640 messages for one node and ~4 for each of the other four (~1 sample per 15–30 s). Raising rates does not help — the winner just consumes more of the shared budget. Binary (change-only) entities are unaffected, which matches the code path.

Mechanism

src/mqtt/state.rs:

pub struct RateLimiter {
    last: HashMap<EntityKind, Duration>,   // no node identity
}

src/mqtt/publisher.rs creates one RateLimiter for the connection and calls publish_snapshot(..., &mut rate_limiter, ...) for each node snapshot and the aggregate; all seven numeric rl.allow(entity, ...) calls therefore compete across nodes for the same last[entity] slot.

Fix

Key the budget by (node_id, EntityKind). Patch against the v2051 tag (also updates the existing tests and adds a regression test rate_limiter_per_node_independent that fails on the old code):

 #[derive(Debug, Default)]
 pub struct RateLimiter {
-    last: HashMap<EntityKind, Duration>,
+    last: HashMap<String, HashMap<EntityKind, Duration>>,
 }

-    pub fn allow(&mut self, entity: EntityKind, now: Duration, rates: &PublishRates) -> bool {
+    pub fn allow(&mut self, node_id: &str, entity: EntityKind, now: Duration, rates: &PublishRates) -> bool {
         let min_gap = match rate_hz_for(entity, rates) {
             rate if rate <= 0.0 => return true,
             rate => Duration::from_secs_f64(1.0 / rate),
         };
-        match self.last.get(&entity) {
+        if !self.last.contains_key(node_id) {
+            self.last.insert(node_id.to_string(), HashMap::new());
+        }
+        let per_node = self.last.get_mut(node_id).expect("inserted above");
+        match per_node.get(&entity) {
             Some(&prev) if now.saturating_sub(prev) < min_gap => false,
             _ => {
-                self.last.insert(entity, now);
+                per_node.insert(entity, now);
                 true
             }
         }
     }

…and in publisher.rs, pass &snap.node_id at the seven call sites in publish_snapshot.

Verification (same 5-node mesh, after fix)

• 60 s topic count: exactly 119 presence_score messages for each of the five nodes (~2 Hz, the configured rate) plus the aggregate at its own cadence — zero starvation.
• Stable overnight: ~82,000 score samples per node at a uniform 1.97/s for 11+ hours.
• cargo test -p wifi-densepose-sensing-server --features mqtt: all green (rust:1.89-bookworm).

Happy to open a PR with the patch (single commit on top of the v2051 tag) if useful.