crates/signald/src/history.rs
428 lines · 17136 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, OpenFlags};
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 /// Open an existing database without writing to it.
68 ///
69 /// [`History::open`] prunes, creates and migrates. A reader that polls —
70 /// the menu-bar pet asks every few seconds when activity last happened —
71 /// must not mutate the daemon's store as a side effect of looking at it.
72 /// Returns an error if the database does not exist yet.
73 pub fn open_read_only(path: &Path) -> rusqlite::Result<History> {
74 let conn = Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_ONLY)?;
75 Ok(History {
76 conn,
77 retention_ms: None,
78 since_prune: 0,
79 })
80 }
81
82 /// When activity was last seen, as unix millis, or `None` if there is
83 /// none on record.
84 ///
85 /// Activity is a keystroke rate or a commit-count above zero. Deliberately
86 /// not `SessionSeconds`: a shell left open is not attention, and counting
87 /// it would let a forgotten terminal stand in for a person indefinitely.
88 /// Deliberately not the hardware signals either — a sleeping machine still
89 /// reports a battery percentage.
90 pub fn last_activity_ms(&self) -> rusqlite::Result<Option<u64>> {
91 let ts: Option<i64> = self.conn.query_row(
92 "SELECT MAX(ts) FROM signals WHERE name IN (?1, ?2) AND value > 0.0",
93 params![
94 signal_schema::SignalName::KeysPerMin.to_u8(),
95 signal_schema::SignalName::CommitsToday.to_u8(),
96 ],
97 |row| row.get(0),
98 )?;
99 Ok(ts.map(|t| t as u64))
100 }
101
102 /// An ephemeral in-memory store (used by tests). No retention.
103 pub fn open_in_memory() -> rusqlite::Result<History> {
104 Self::init(Connection::open_in_memory()?, None)
105 }
106
107 fn init(conn: Connection, retention_ms: Option<u64>) -> rusqlite::Result<History> {
108 // Columns for querying (name, ts, value, tag); `frame` is the audited
109 // wire encoding, reused for faithful, validated reconstruction on read.
110 conn.execute(
111 "CREATE TABLE IF NOT EXISTS signals (
112 id INTEGER PRIMARY KEY AUTOINCREMENT,
113 schema_version INTEGER NOT NULL,
114 ts INTEGER NOT NULL,
115 source INTEGER NOT NULL,
116 name INTEGER NOT NULL,
117 value REAL NOT NULL,
118 tag TEXT,
119 frame BLOB NOT NULL
120 )",
121 [],
122 )?;
123 conn.execute(
124 "CREATE INDEX IF NOT EXISTS idx_signals_name_ts ON signals(name, ts)",
125 [],
126 )?;
127 // For the retention delete.
128 conn.execute("CREATE INDEX IF NOT EXISTS idx_signals_ts ON signals(ts)", [])?;
129 Ok(History {
130 conn,
131 retention_ms,
132 since_prune: 0,
133 })
134 }
135
136 /// Delete rows written by a different schema version. Returns the number
137 /// deleted.
138 ///
139 /// `query` rebuilds each row from its stored `frame`, and `wire::decode`
140 /// refuses a frame whose version is not [`signal_schema::SCHEMA_VERSION`].
141 /// Rows from an older daemon would therefore linger, unreadable and
142 /// silently skipped, until retention caught up with them. Dropping them on
143 /// open keeps the store to rows it can actually return.
144 pub fn drop_foreign_schema_rows(&self) -> rusqlite::Result<usize> {
145 self.conn.execute(
146 "DELETE FROM signals WHERE schema_version != ?1",
147 params![signal_schema::SCHEMA_VERSION],
148 )
149 }
150
151 /// Delete rows older than the retention. Returns the number deleted.
152 pub fn prune(&self) -> rusqlite::Result<usize> {
153 let Some(retention_ms) = self.retention_ms else {
154 return Ok(0);
155 };
156 let cutoff = now_millis().saturating_sub(retention_ms) as i64;
157 self.conn
158 .execute("DELETE FROM signals WHERE ts < ?1", params![cutoff])
159 }
160
161 /// Append one signal to the history, pruning every [`PRUNE_EVERY`] calls.
162 pub fn record(&mut self, s: &Signal) -> rusqlite::Result<()> {
163 self.since_prune += 1;
164 if self.since_prune >= PRUNE_EVERY {
165 self.since_prune = 0;
166 self.prune()?;
167 }
168 self.conn.execute(
169 "INSERT INTO signals (schema_version, ts, source, name, value, tag, frame)
170 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
171 params![
172 s.schema_version,
173 s.ts as i64, // SQLite integers are i64; unix-millis fits
174 s.source.to_u8(),
175 s.name.to_u8(),
176 s.value.0,
177 s.tag.as_ref().map(|t| t.as_str()),
178 wire::encode(s),
179 ],
180 )?;
181 Ok(())
182 }
183
184 /// The most recent `limit` signals, newest first. This is the query that
185 /// lets a renderer read recent history instead of only the live snapshot.
186 pub fn recent(&self, limit: usize) -> rusqlite::Result<Vec<Signal>> {
187 self.query("SELECT frame FROM signals ORDER BY id DESC LIMIT ?1", limit)
188 }
189
190 /// The most recent `limit` signals for one metric name, newest first.
191 pub fn recent_named(
192 &self,
193 name: signal_schema::SignalName,
194 limit: usize,
195 ) -> rusqlite::Result<Vec<Signal>> {
196 let mut stmt = self.conn.prepare(
197 "SELECT frame FROM signals WHERE name = ?1 ORDER BY id DESC LIMIT ?2",
198 )?;
199 let rows = stmt.query_map(params![name.to_u8(), limit as i64], |row| {
200 row.get::<_, Vec<u8>>(0)
201 })?;
202 Self::decode_rows(rows)
203 }
204
205 fn query(&self, sql: &str, limit: usize) -> rusqlite::Result<Vec<Signal>> {
206 let mut stmt = self.conn.prepare(sql)?;
207 let rows = stmt.query_map(params![limit as i64], |row| row.get::<_, Vec<u8>>(0))?;
208 Self::decode_rows(rows)
209 }
210
211 fn decode_rows(
212 rows: impl Iterator<Item = rusqlite::Result<Vec<u8>>>,
213 ) -> rusqlite::Result<Vec<Signal>> {
214 let mut out = Vec::new();
215 for frame in rows {
216 let frame = frame?;
217 if let Some(sig) = wire::decode(&frame) {
218 out.push(sig);
219 }
220 }
221 Ok(out)
222 }
223}
224
225#[cfg(test)]
226mod tests {
227 use super::*;
228 use signal_schema::{SignalName, Source, Tag, Value, SCHEMA_VERSION};
229
230 fn sig(name: SignalName, value: f64, tag: Option<Tag>) -> Signal {
231 Signal {
232 schema_version: SCHEMA_VERSION,
233 ts: 1_723_100_000_000,
234 source: Source::Git,
235 name,
236 value: Value(value),
237 tag,
238 }
239 }
240
241 #[test]
242 fn round_trip_write_then_query_recent() {
243 let mut h = History::open_in_memory().unwrap();
244 let repo = Tag::repo_path("/x/repo").unwrap();
245 h.record(&sig(SignalName::CommitsWindow, 5.0, Some(repo.clone())))
246 .unwrap();
247 h.record(&sig(SignalName::CommitsToday, 2.0, Some(repo.clone())))
248 .unwrap();
249 h.record(&sig(SignalName::KeysPerMin, 88.0, None)).unwrap();
250
251 let recent = h.recent(10).unwrap();
252 assert_eq!(recent.len(), 3);
253 // Newest first, and the aggregates survive the round trip exactly.
254 assert_eq!(recent[0].name, SignalName::KeysPerMin);
255 assert_eq!(recent[0].value, Value(88.0));
256 assert_eq!(recent[2].name, SignalName::CommitsWindow);
257 assert_eq!(recent[2].value, Value(5.0));
258 assert_eq!(recent[2].tag.as_ref().unwrap().as_str(), "/x/repo");
259 }
260
261 #[test]
262 fn recent_named_filters_by_metric() {
263 let mut h = History::open_in_memory().unwrap();
264 for v in [1.0, 2.0, 3.0] {
265 h.record(&sig(SignalName::KeysPerMin, v, None)).unwrap();
266 }
267 h.record(&sig(SignalName::SessionSeconds, 42.0, None))
268 .unwrap();
269
270 let keys = h.recent_named(SignalName::KeysPerMin, 10).unwrap();
271 assert_eq!(keys.len(), 3);
272 assert!(keys.iter().all(|s| s.name == SignalName::KeysPerMin));
273 assert_eq!(keys[0].value, Value(3.0)); // newest first
274 }
275
276 #[test]
277 fn retention_prunes_old_rows_on_open_and_periodically() {
278 let dir = std::env::temp_dir().join(format!("signald-history-{}", std::process::id()));
279 std::fs::create_dir_all(&dir).unwrap();
280 let db = dir.join("h.sqlite");
281 let retention = Duration::from_secs(3600);
282 let ancient = Signal {
283 ts: 1,
284 ..sig(SignalName::KeysPerMin, 1.0, None)
285 };
286 let fresh = Signal {
287 ts: now_millis(),
288 ..sig(SignalName::KeysPerMin, 2.0, None)
289 };
290 {
291 let mut h = History::open_with_retention(&db, retention).unwrap();
292 h.record(&ancient).unwrap();
293 h.record(&fresh).unwrap();
294 assert_eq!(h.recent(10).unwrap().len(), 2);
295 assert_eq!(h.prune().unwrap(), 1);
296 assert_eq!(h.recent(10).unwrap().len(), 1);
297 // Left behind for the reopen to prune.
298 h.record(&ancient).unwrap();
299 }
300 let h = History::open_with_retention(&db, retention).unwrap();
301 let rows = h.recent(10).unwrap();
302 assert_eq!(rows.len(), 1, "reopen prunes");
303 assert_eq!(rows[0].value, Value(2.0));
304 let _ = std::fs::remove_dir_all(&dir);
305 }
306
307 /// Activity is what the menu-bar pet lives on, so what counts is exact.
308 #[test]
309 fn last_activity_is_typing_and_commits_only() {
310 let mut h = History::open_in_memory().unwrap();
311 // A shell sitting open, and a machine reporting hardware while asleep.
312 // Neither is a person showing up.
313 for (name, value, ts) in [
314 (SignalName::SessionSeconds, 9000.0, 500_u64),
315 (SignalName::BatteryPct, 80.0, 600),
316 (SignalName::CpuLoad, 0.4, 700),
317 ] {
318 h.record(&Signal { ts, ..sig(name, value, None) }).unwrap();
319 }
320 assert_eq!(h.last_activity_ms().unwrap(), None, "none of that is attention");
321
322 h.record(&Signal { ts: 1_000, ..sig(SignalName::KeysPerMin, 30.0, None) })
323 .unwrap();
324 assert_eq!(h.last_activity_ms().unwrap(), Some(1_000));
325
326 h.record(&Signal { ts: 2_000, ..sig(SignalName::CommitsToday, 3.0, None) })
327 .unwrap();
328 assert_eq!(h.last_activity_ms().unwrap(), Some(2_000), "the most recent wins");
329 }
330
331 /// A quiet day publishes zeroes rather than nothing, so a zero must not
332 /// read as activity or the pet would never age.
333 #[test]
334 fn a_zero_is_not_activity() {
335 let mut h = History::open_in_memory().unwrap();
336 h.record(&Signal { ts: 1_000, ..sig(SignalName::KeysPerMin, 0.0, None) })
337 .unwrap();
338 h.record(&Signal { ts: 1_100, ..sig(SignalName::CommitsToday, 0.0, None) })
339 .unwrap();
340 assert_eq!(h.last_activity_ms().unwrap(), None);
341 }
342
343 /// The reason this exists: `open` prunes on the way in. A pet polling every
344 /// five seconds must not be quietly deleting the daemon's history.
345 #[test]
346 fn read_only_open_does_not_prune() {
347 let dir = std::env::temp_dir().join(format!("signald-ro-{}", std::process::id()));
348 let _ = std::fs::remove_dir_all(&dir);
349 std::fs::create_dir_all(&dir).unwrap();
350 let db = dir.join("history.db");
351
352 {
353 // Retention of an hour, and a row far older than that.
354 let mut h =
355 History::open_with_retention(&db, std::time::Duration::from_secs(3600)).unwrap();
356 h.record(&Signal { ts: 1, ..sig(SignalName::KeysPerMin, 5.0, None) })
357 .unwrap();
358 }
359
360 let ro = History::open_read_only(&db).unwrap();
361 assert_eq!(ro.last_activity_ms().unwrap(), Some(1), "the old row is readable");
362 drop(ro);
363
364 // Still there: a read-only open left it alone. A normal open would have
365 // pruned it on the way in.
366 let ro = History::open_read_only(&db).unwrap();
367 assert_eq!(
368 ro.conn
369 .query_row("SELECT COUNT(*) FROM signals", [], |r| r.get::<_, i64>(0))
370 .unwrap(),
371 1,
372 "read-only open must not have deleted anything"
373 );
374 drop(ro);
375
376 // And prove the contrast: a normal open does prune it.
377 let h = History::open_with_retention(&db, std::time::Duration::from_secs(3600)).unwrap();
378 assert_eq!(h.recent(10).unwrap().len(), 0, "a writing open prunes");
379 let _ = std::fs::remove_dir_all(&dir);
380 }
381
382 #[test]
383 fn read_only_open_refuses_a_database_that_is_not_there() {
384 let missing = std::env::temp_dir().join(format!("signald-absent-{}.db", std::process::id()));
385 let _ = std::fs::remove_file(&missing);
386 assert!(
387 History::open_read_only(&missing).is_err(),
388 "read-only must not conjure a database"
389 );
390 }
391
392 /// A row written under an older schema cannot be rebuilt from its frame, so
393 /// reopening drops it rather than leaving it to be skipped silently.
394 #[test]
395 fn reopen_drops_rows_from_a_foreign_schema_version() {
396 let dir = std::env::temp_dir().join(format!("signald-schema-{}", std::process::id()));
397 std::fs::create_dir_all(&dir).unwrap();
398 let db = dir.join("history.db");
399
400 let old_version = signal_schema::SCHEMA_VERSION - 1;
401 {
402 let h = History::open(&db).unwrap();
403 // A row exactly as an older daemon left it: the version in the
404 // column and in the stored frame both predate this build.
405 let s = sig(SignalName::KeysPerMin, 1.0, None);
406 let mut frame = wire::encode(&s);
407 frame[4..6].copy_from_slice(&old_version.to_le_bytes());
408 h.conn
409 .execute(
410 "INSERT INTO signals (schema_version, ts, source, name, value, tag, frame)
411 VALUES (?1, ?2, ?3, ?4, ?5, NULL, ?6)",
412 params![old_version, s.ts as i64, s.source.to_u8(), s.name.to_u8(), s.value.0, frame],
413 )
414 .unwrap();
415 assert_eq!(h.recent(10).unwrap().len(), 0, "unreadable while present");
416 }
417
418 let h = History::open(&db).unwrap();
419 assert_eq!(
420 h.conn
421 .query_row("SELECT COUNT(*) FROM signals", [], |r| r.get::<_, i64>(0))
422 .unwrap(),
423 0,
424 "the foreign-schema row is gone, not merely unreadable"
425 );
426 let _ = std::fs::remove_dir_all(&dir);
427 }
428}