crates/signald/src/hub.rs
177 lines · 6123 bytes
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, source discriminant, and optional
26/// audited tag. Distinct per-repo signals coexist, and so do two signals that
27/// share a name but come from different collectors — `SignalName::CollectorUp`
28/// is published once per `Source`, and without `source` in the key those
29/// would overwrite each other. A new value for the same key replaces the old
30/// one (keep-latest).
31type Key = (u8, u8, Option<String>);
32
33/// A cloneable handle to the shared hub. Clones share one cache + subscriber
34/// set behind a mutex.
35#[derive(Clone)]
36pub struct Hub {
37 inner: Arc<Mutex<Inner>>,
38}
39
40struct Inner {
41 cache: BTreeMap<Key, Signal>,
42 subs: Vec<Sender<Signal>>,
43 history: Option<History>,
44}
45
46impl Hub {
47 /// A hub with no history (live-only). Useful for tests and for a renderer
48 /// that never persists.
49 pub fn new() -> Hub {
50 Hub::build(None)
51 }
52
53 /// A hub that also persists every published signal to `history`.
54 pub fn with_history(history: History) -> Hub {
55 Hub::build(Some(history))
56 }
57
58 fn build(history: Option<History>) -> Hub {
59 Hub {
60 inner: Arc::new(Mutex::new(Inner {
61 cache: BTreeMap::new(),
62 subs: Vec::new(),
63 history,
64 })),
65 }
66 }
67
68 /// Publish one signal: persist it (if a history is attached), update the
69 /// last-value cache, and fan it out to every live subscriber. Dead
70 /// subscribers (whose receiver was dropped) are pruned here.
71 pub fn publish(&self, s: Signal) {
72 let mut guard = self.inner.lock().unwrap();
73 let Inner {
74 cache,
75 subs,
76 history,
77 } = &mut *guard;
78
79 if let Some(h) = history {
80 // A history write failing must not stop the live stream.
81 let _ = h.record(&s);
82 }
83 let key = (
84 s.name.to_u8(),
85 s.source.to_u8(),
86 s.tag.as_ref().map(|t| t.as_str().to_string()),
87 );
88 cache.insert(key, s.clone());
89 subs.retain(|tx| tx.send(s.clone()).is_ok());
90 }
91
92 /// Subscribe: return the current last-value snapshot plus a receiver that
93 /// yields every signal published after this call.
94 pub fn subscribe(&self) -> (Vec<Signal>, Receiver<Signal>) {
95 let mut guard = self.inner.lock().unwrap();
96 let snapshot = guard.cache.values().cloned().collect();
97 let (tx, rx) = channel();
98 guard.subs.push(tx);
99 (snapshot, rx)
100 }
101
102 /// The current last-value snapshot without subscribing.
103 pub fn snapshot(&self) -> Vec<Signal> {
104 self.inner.lock().unwrap().cache.values().cloned().collect()
105 }
106}
107
108impl Default for Hub {
109 fn default() -> Self {
110 Hub::new()
111 }
112}
113
114#[cfg(test)]
115mod tests {
116 use super::*;
117 use signal_schema::{SignalName, Source, Value, SCHEMA_VERSION};
118
119 fn sig(name: SignalName, value: f64) -> Signal {
120 Signal {
121 schema_version: SCHEMA_VERSION,
122 ts: 1,
123 source: Source::Terminal,
124 name,
125 value: Value(value),
126 tag: None,
127 }
128 }
129
130 #[test]
131 fn subscriber_gets_cached_value_then_live_update() {
132 let hub = Hub::new();
133 // A value published before anyone subscribes must still be seen (cache).
134 hub.publish(sig(SignalName::KeysPerMin, 10.0));
135
136 let (snapshot, rx) = hub.subscribe();
137 assert_eq!(snapshot.len(), 1);
138 assert_eq!(snapshot[0].value, Value(10.0));
139
140 // A change after subscribing arrives live.
141 hub.publish(sig(SignalName::KeysPerMin, 25.0));
142 let update = rx.recv().expect("live update");
143 assert_eq!(update.name, SignalName::KeysPerMin);
144 assert_eq!(update.value, Value(25.0));
145 }
146
147 #[test]
148 fn cache_keeps_latest_per_name_and_source() {
149 let hub = Hub::new();
150 hub.publish(sig(SignalName::KeysPerMin, 1.0));
151 hub.publish(sig(SignalName::KeysPerMin, 2.0));
152 let snap = hub.snapshot();
153 assert_eq!(snap.len(), 1, "same name and source collapses to keep-latest");
154 assert_eq!(snap[0].value, Value(2.0));
155 }
156
157 /// `CollectorUp` is published once per source, so the cache must not treat
158 /// the name alone as the identity. Without `source` in the key the second
159 /// publish would evict the first and a subscriber would see one collector's
160 /// health standing in for all three.
161 #[test]
162 fn same_name_from_two_sources_coexists() {
163 let hub = Hub::new();
164 for (source, value) in [(Source::Git, 1.0), (Source::Hardware, 0.0)] {
165 hub.publish(Signal {
166 source,
167 ..sig(SignalName::CollectorUp, value)
168 });
169 }
170 let snap = hub.snapshot();
171 assert_eq!(snap.len(), 2, "one entry per source");
172 let git = snap.iter().find(|s| s.source == Source::Git).expect("git health");
173 let hw = snap.iter().find(|s| s.source == Source::Hardware).expect("hardware health");
174 assert_eq!(git.value, Value(1.0));
175 assert_eq!(hw.value, Value(0.0));
176 }
177}