crates/signald/src/history.rs
308 lines · 11991 bytes
1//! # History store
2//!
3//! A local SQLite (WAL) store of a rolling history of signals. History is what
4//! lets a renderer read *recent aggregates* rather than only a live snapshot —
5//! it is what lets the garden survive a restart and (later) the aging pet know
6//! its true age.
7//!
8//! ## Same privacy constraints as the wire
9//!
10//! The store holds **aggregate scalars only**. Each row carries the schema
11//! discriminants, the `f64` `value`, and the audited non-content `tag` — exactly
12//! the fields of a [`Signal`], nothing wider. There is no content column. As a
13//! belt-and-braces measure the audited wire frame is also stored as a BLOB and
14//! is the source of truth on read, so reconstruction reuses the same validated
15//! [`wire::decode`] the socket uses — a row can decode to nothing but a
16//! well-formed, `f64`-only signal. The differential secret-typing test writes a
17//! planted secret through the collector into this store and asserts it never
18//! appears here in any encoding.
19//!
20//! ## Retention
21//!
22//! The store is rolling: rows older than the retention are deleted on open and
23//! every [`PRUNE_EVERY`] records, so the file stays bounded for a daemon that
24//! runs for months.
25
26use std::path::Path;
27use std::time::Duration;
28
29use rusqlite::{params, Connection};
30use signal_schema::{wire, Signal};
31
32use crate::now_millis;
33
34/// Retention for [`History::open`].
35pub const DEFAULT_RETENTION: Duration = Duration::from_secs(7 * 86_400);
36
37/// [`History::record`] prunes after this many inserts.
38pub const PRUNE_EVERY: u64 = 1000;
39
40/// A handle to the SQLite history database.
41pub struct History {
42 conn: Connection,
43 /// `None` for an in-memory store: nothing is ever pruned.
44 retention_ms: Option<u64>,
45 since_prune: u64,
46}
47
48impl History {
49 /// Open (creating if needed) the history database at `path` in WAL mode
50 /// with [`DEFAULT_RETENTION`].
51 pub fn open(path: &Path) -> rusqlite::Result<History> {
52 Self::open_with_retention(path, DEFAULT_RETENTION)
53 }
54
55 /// Open with an explicit retention. Rows older than it are pruned now and
56 /// periodically as records arrive.
57 pub fn open_with_retention(path: &Path, retention: Duration) -> rusqlite::Result<History> {
58 let conn = Connection::open(path)?;
59 // WAL: concurrent reads while the daemon writes; survives restarts.
60 conn.pragma_update(None, "journal_mode", "WAL")?;
61 let h = Self::init(conn, Some(retention.as_millis() as u64))?;
62 h.drop_foreign_schema_rows()?;
63 h.prune()?;
64 Ok(h)
65 }
66
67 /// An ephemeral in-memory store (used by tests). No retention.
68 pub fn open_in_memory() -> rusqlite::Result<History> {
69 Self::init(Connection::open_in_memory()?, None)
70 }
71
72 fn init(conn: Connection, retention_ms: Option<u64>) -> rusqlite::Result<History> {
73 // Columns for querying (name, ts, value, tag); `frame` is the audited
74 // wire encoding, reused for faithful, validated reconstruction on read.
75 conn.execute(
76 "CREATE TABLE IF NOT EXISTS signals (
77 id INTEGER PRIMARY KEY AUTOINCREMENT,
78 schema_version INTEGER NOT NULL,
79 ts INTEGER NOT NULL,
80 source INTEGER NOT NULL,
81 name INTEGER NOT NULL,
82 value REAL NOT NULL,
83 tag TEXT,
84 frame BLOB NOT NULL
85 )",
86 [],
87 )?;
88 conn.execute(
89 "CREATE INDEX IF NOT EXISTS idx_signals_name_ts ON signals(name, ts)",
90 [],
91 )?;
92 // For the retention delete.
93 conn.execute("CREATE INDEX IF NOT EXISTS idx_signals_ts ON signals(ts)", [])?;
94 Ok(History {
95 conn,
96 retention_ms,
97 since_prune: 0,
98 })
99 }
100
101 /// Delete rows written by a different schema version. Returns the number
102 /// deleted.
103 ///
104 /// `query` rebuilds each row from its stored `frame`, and `wire::decode`
105 /// refuses a frame whose version is not [`signal_schema::SCHEMA_VERSION`].
106 /// Rows from an older daemon would therefore linger, unreadable and
107 /// silently skipped, until retention caught up with them. Dropping them on
108 /// open keeps the store to rows it can actually return.
109 pub fn drop_foreign_schema_rows(&self) -> rusqlite::Result<usize> {
110 self.conn.execute(
111 "DELETE FROM signals WHERE schema_version != ?1",
112 params![signal_schema::SCHEMA_VERSION],
113 )
114 }
115
116 /// Delete rows older than the retention. Returns the number deleted.
117 pub fn prune(&self) -> rusqlite::Result<usize> {
118 let Some(retention_ms) = self.retention_ms else {
119 return Ok(0);
120 };
121 let cutoff = now_millis().saturating_sub(retention_ms) as i64;
122 self.conn
123 .execute("DELETE FROM signals WHERE ts < ?1", params![cutoff])
124 }
125
126 /// Append one signal to the history, pruning every [`PRUNE_EVERY`] calls.
127 pub fn record(&mut self, s: &Signal) -> rusqlite::Result<()> {
128 self.since_prune += 1;
129 if self.since_prune >= PRUNE_EVERY {
130 self.since_prune = 0;
131 self.prune()?;
132 }
133 self.conn.execute(
134 "INSERT INTO signals (schema_version, ts, source, name, value, tag, frame)
135 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
136 params![
137 s.schema_version,
138 s.ts as i64, // SQLite integers are i64; unix-millis fits
139 s.source.to_u8(),
140 s.name.to_u8(),
141 s.value.0,
142 s.tag.as_ref().map(|t| t.as_str()),
143 wire::encode(s),
144 ],
145 )?;
146 Ok(())
147 }
148
149 /// The most recent `limit` signals, newest first. This is the query that
150 /// lets a renderer read recent history instead of only the live snapshot.
151 pub fn recent(&self, limit: usize) -> rusqlite::Result<Vec<Signal>> {
152 self.query("SELECT frame FROM signals ORDER BY id DESC LIMIT ?1", limit)
153 }
154
155 /// The most recent `limit` signals for one metric name, newest first.
156 pub fn recent_named(
157 &self,
158 name: signal_schema::SignalName,
159 limit: usize,
160 ) -> rusqlite::Result<Vec<Signal>> {
161 let mut stmt = self.conn.prepare(
162 "SELECT frame FROM signals WHERE name = ?1 ORDER BY id DESC LIMIT ?2",
163 )?;
164 let rows = stmt.query_map(params![name.to_u8(), limit as i64], |row| {
165 row.get::<_, Vec<u8>>(0)
166 })?;
167 Self::decode_rows(rows)
168 }
169
170 fn query(&self, sql: &str, limit: usize) -> rusqlite::Result<Vec<Signal>> {
171 let mut stmt = self.conn.prepare(sql)?;
172 let rows = stmt.query_map(params![limit as i64], |row| row.get::<_, Vec<u8>>(0))?;
173 Self::decode_rows(rows)
174 }
175
176 fn decode_rows(
177 rows: impl Iterator<Item = rusqlite::Result<Vec<u8>>>,
178 ) -> rusqlite::Result<Vec<Signal>> {
179 let mut out = Vec::new();
180 for frame in rows {
181 let frame = frame?;
182 if let Some(sig) = wire::decode(&frame) {
183 out.push(sig);
184 }
185 }
186 Ok(out)
187 }
188}
189
190#[cfg(test)]
191mod tests {
192 use super::*;
193 use signal_schema::{SignalName, Source, Tag, Value, SCHEMA_VERSION};
194
195 fn sig(name: SignalName, value: f64, tag: Option<Tag>) -> Signal {
196 Signal {
197 schema_version: SCHEMA_VERSION,
198 ts: 1_723_100_000_000,
199 source: Source::Git,
200 name,
201 value: Value(value),
202 tag,
203 }
204 }
205
206 #[test]
207 fn round_trip_write_then_query_recent() {
208 let mut h = History::open_in_memory().unwrap();
209 let repo = Tag::repo_path("/x/repo").unwrap();
210 h.record(&sig(SignalName::CommitsWindow, 5.0, Some(repo.clone())))
211 .unwrap();
212 h.record(&sig(SignalName::CommitsToday, 2.0, Some(repo.clone())))
213 .unwrap();
214 h.record(&sig(SignalName::KeysPerMin, 88.0, None)).unwrap();
215
216 let recent = h.recent(10).unwrap();
217 assert_eq!(recent.len(), 3);
218 // Newest first, and the aggregates survive the round trip exactly.
219 assert_eq!(recent[0].name, SignalName::KeysPerMin);
220 assert_eq!(recent[0].value, Value(88.0));
221 assert_eq!(recent[2].name, SignalName::CommitsWindow);
222 assert_eq!(recent[2].value, Value(5.0));
223 assert_eq!(recent[2].tag.as_ref().unwrap().as_str(), "/x/repo");
224 }
225
226 #[test]
227 fn recent_named_filters_by_metric() {
228 let mut h = History::open_in_memory().unwrap();
229 for v in [1.0, 2.0, 3.0] {
230 h.record(&sig(SignalName::KeysPerMin, v, None)).unwrap();
231 }
232 h.record(&sig(SignalName::SessionSeconds, 42.0, None))
233 .unwrap();
234
235 let keys = h.recent_named(SignalName::KeysPerMin, 10).unwrap();
236 assert_eq!(keys.len(), 3);
237 assert!(keys.iter().all(|s| s.name == SignalName::KeysPerMin));
238 assert_eq!(keys[0].value, Value(3.0)); // newest first
239 }
240
241 #[test]
242 fn retention_prunes_old_rows_on_open_and_periodically() {
243 let dir = std::env::temp_dir().join(format!("signald-history-{}", std::process::id()));
244 std::fs::create_dir_all(&dir).unwrap();
245 let db = dir.join("h.sqlite");
246 let retention = Duration::from_secs(3600);
247 let ancient = Signal {
248 ts: 1,
249 ..sig(SignalName::KeysPerMin, 1.0, None)
250 };
251 let fresh = Signal {
252 ts: now_millis(),
253 ..sig(SignalName::KeysPerMin, 2.0, None)
254 };
255 {
256 let mut h = History::open_with_retention(&db, retention).unwrap();
257 h.record(&ancient).unwrap();
258 h.record(&fresh).unwrap();
259 assert_eq!(h.recent(10).unwrap().len(), 2);
260 assert_eq!(h.prune().unwrap(), 1);
261 assert_eq!(h.recent(10).unwrap().len(), 1);
262 // Left behind for the reopen to prune.
263 h.record(&ancient).unwrap();
264 }
265 let h = History::open_with_retention(&db, retention).unwrap();
266 let rows = h.recent(10).unwrap();
267 assert_eq!(rows.len(), 1, "reopen prunes");
268 assert_eq!(rows[0].value, Value(2.0));
269 let _ = std::fs::remove_dir_all(&dir);
270 }
271
272 /// A row written under an older schema cannot be rebuilt from its frame, so
273 /// reopening drops it rather than leaving it to be skipped silently.
274 #[test]
275 fn reopen_drops_rows_from_a_foreign_schema_version() {
276 let dir = std::env::temp_dir().join(format!("signald-schema-{}", std::process::id()));
277 std::fs::create_dir_all(&dir).unwrap();
278 let db = dir.join("history.db");
279
280 let old_version = signal_schema::SCHEMA_VERSION - 1;
281 {
282 let h = History::open(&db).unwrap();
283 // A row exactly as an older daemon left it: the version in the
284 // column and in the stored frame both predate this build.
285 let s = sig(SignalName::KeysPerMin, 1.0, None);
286 let mut frame = wire::encode(&s);
287 frame[4..6].copy_from_slice(&old_version.to_le_bytes());
288 h.conn
289 .execute(
290 "INSERT INTO signals (schema_version, ts, source, name, value, tag, frame)
291 VALUES (?1, ?2, ?3, ?4, ?5, NULL, ?6)",
292 params![old_version, s.ts as i64, s.source.to_u8(), s.name.to_u8(), s.value.0, frame],
293 )
294 .unwrap();
295 assert_eq!(h.recent(10).unwrap().len(), 0, "unreadable while present");
296 }
297
298 let h = History::open(&db).unwrap();
299 assert_eq!(
300 h.conn
301 .query_row("SELECT COUNT(*) FROM signals", [], |r| r.get::<_, i64>(0))
302 .unwrap(),
303 0,
304 "the foreign-schema row is gone, not merely unreadable"
305 );
306 let _ = std::fs::remove_dir_all(&dir);
307 }
308}