crates/signald/src/hub.rs

  1//! # The signal hub — last-value cache + live fan-out
  2//!
  3//! The in-memory heart of the publish side. Collectors [`Hub::publish`] signals;
  4//! subscribers [`Hub::subscribe`] and receive the current value of every cached
  5//! metric immediately, then a live stream of updates.
  6//!
  7//! Subscription semantics: *last-value-cache + live stream.* On
  8//! subscribe, the caller is handed the latest value of each `name` (keyed by
  9//! name + audited tag), so a renderer paints correct state at once; thereafter
 10//! it receives updates as they are published. Channels are unbounded, so a slow
 11//! subscriber never blocks a publisher.
 12//!
 13//! When a `Hub` is built [`Hub::with_history`], every published signal is also
 14//! persisted to the SQLite history store, so live and persisted paths share one
 15//! write point.
 16
 17use std::collections::BTreeMap;
 18use std::sync::mpsc::{channel, Receiver, Sender};
 19use std::sync::{Arc, Mutex};
 20
 21use signal_schema::Signal;
 22
 23use crate::history::History;
 24
 25/// Cache key: metric name discriminant + optional audited tag. Distinct
 26/// per-repo / per-core signals coexist; a new value for the same key replaces
 27/// the old one (keep-latest).
 28type Key = (u8, Option<String>);
 29
 30/// A cloneable handle to the shared hub. Clones share one cache + subscriber
 31/// set behind a mutex.
 32#[derive(Clone)]
 33pub struct Hub {
 34    inner: Arc<Mutex<Inner>>,
 35}
 36
 37struct Inner {
 38    cache: BTreeMap<Key, Signal>,
 39    subs: Vec<Sender<Signal>>,
 40    history: Option<History>,
 41}
 42
 43impl Hub {
 44    /// A hub with no history (live-only). Useful for tests and for a renderer
 45    /// that never persists.
 46    pub fn new() -> Hub {
 47        Hub::build(None)
 48    }
 49
 50    /// A hub that also persists every published signal to `history`.
 51    pub fn with_history(history: History) -> Hub {
 52        Hub::build(Some(history))
 53    }
 54
 55    fn build(history: Option<History>) -> Hub {
 56        Hub {
 57            inner: Arc::new(Mutex::new(Inner {
 58                cache: BTreeMap::new(),
 59                subs: Vec::new(),
 60                history,
 61            })),
 62        }
 63    }
 64
 65    /// Publish one signal: persist it (if a history is attached), update the
 66    /// last-value cache, and fan it out to every live subscriber. Dead
 67    /// subscribers (whose receiver was dropped) are pruned here.
 68    pub fn publish(&self, s: Signal) {
 69        let mut guard = self.inner.lock().unwrap();
 70        let Inner {
 71            cache,
 72            subs,
 73            history,
 74        } = &mut *guard;
 75
 76        if let Some(h) = history {
 77            // A history write failing must not stop the live stream.
 78            let _ = h.record(&s);
 79        }
 80        let key = (s.name.to_u8(), s.tag.as_ref().map(|t| t.as_str().to_string()));
 81        cache.insert(key, s.clone());
 82        subs.retain(|tx| tx.send(s.clone()).is_ok());
 83    }
 84
 85    /// Subscribe: return the current last-value snapshot plus a receiver that
 86    /// yields every signal published after this call.
 87    pub fn subscribe(&self) -> (Vec<Signal>, Receiver<Signal>) {
 88        let mut guard = self.inner.lock().unwrap();
 89        let snapshot = guard.cache.values().cloned().collect();
 90        let (tx, rx) = channel();
 91        guard.subs.push(tx);
 92        (snapshot, rx)
 93    }
 94
 95    /// The current last-value snapshot without subscribing.
 96    pub fn snapshot(&self) -> Vec<Signal> {
 97        self.inner.lock().unwrap().cache.values().cloned().collect()
 98    }
 99}
100
101impl Default for Hub {
102    fn default() -> Self {
103        Hub::new()
104    }
105}
106
107#[cfg(test)]
108mod tests {
109    use super::*;
110    use signal_schema::{SignalName, Source, Value, SCHEMA_VERSION};
111
112    fn sig(name: SignalName, value: f64) -> Signal {
113        Signal {
114            schema_version: SCHEMA_VERSION,
115            ts: 1,
116            source: Source::Terminal,
117            name,
118            value: Value(value),
119            tag: None,
120        }
121    }
122
123    #[test]
124    fn subscriber_gets_cached_value_then_live_update() {
125        let hub = Hub::new();
126        // A value published before anyone subscribes must still be seen (cache).
127        hub.publish(sig(SignalName::KeysPerMin, 10.0));
128
129        let (snapshot, rx) = hub.subscribe();
130        assert_eq!(snapshot.len(), 1);
131        assert_eq!(snapshot[0].value, Value(10.0));
132
133        // A change after subscribing arrives live.
134        hub.publish(sig(SignalName::KeysPerMin, 25.0));
135        let update = rx.recv().expect("live update");
136        assert_eq!(update.name, SignalName::KeysPerMin);
137        assert_eq!(update.value, Value(25.0));
138    }
139
140    #[test]
141    fn cache_keeps_latest_per_name() {
142        let hub = Hub::new();
143        hub.publish(sig(SignalName::KeysPerMin, 1.0));
144        hub.publish(sig(SignalName::KeysPerMin, 2.0));
145        let snap = hub.snapshot();
146        assert_eq!(snap.len(), 1, "same name collapses to keep-latest");
147        assert_eq!(snap[0].value, Value(2.0));
148    }
149}