crates/signald/src/history.rs
176 lines · 6572 bytes
1//! # History store (spec §1.3, §4)
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
20use std::path::Path;
21
22use rusqlite::{params, Connection};
23use signal_schema::{wire, Signal};
24
25/// A handle to the SQLite history database.
26pub struct History {
27 conn: Connection,
28}
29
30impl History {
31 /// Open (creating if needed) the history database at `path` in WAL mode.
32 pub fn open(path: &Path) -> rusqlite::Result<History> {
33 let conn = Connection::open(path)?;
34 // WAL: concurrent reads while the daemon writes; survives restarts.
35 conn.pragma_update(None, "journal_mode", "WAL")?;
36 Self::init(conn)
37 }
38
39 /// An ephemeral in-memory store (used by tests).
40 pub fn open_in_memory() -> rusqlite::Result<History> {
41 Self::init(Connection::open_in_memory()?)
42 }
43
44 fn init(conn: Connection) -> rusqlite::Result<History> {
45 // Columns for querying (name, ts, value, tag); `frame` is the audited
46 // wire encoding, reused for faithful, validated reconstruction on read.
47 conn.execute(
48 "CREATE TABLE IF NOT EXISTS signals (
49 id INTEGER PRIMARY KEY AUTOINCREMENT,
50 schema_version INTEGER NOT NULL,
51 ts INTEGER NOT NULL,
52 source INTEGER NOT NULL,
53 name INTEGER NOT NULL,
54 value REAL NOT NULL,
55 tag TEXT,
56 frame BLOB NOT NULL
57 )",
58 [],
59 )?;
60 conn.execute(
61 "CREATE INDEX IF NOT EXISTS idx_signals_name_ts ON signals(name, ts)",
62 [],
63 )?;
64 Ok(History { conn })
65 }
66
67 /// Append one signal to the history.
68 pub fn record(&self, s: &Signal) -> rusqlite::Result<()> {
69 self.conn.execute(
70 "INSERT INTO signals (schema_version, ts, source, name, value, tag, frame)
71 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
72 params![
73 s.schema_version,
74 s.ts as i64, // SQLite integers are i64; unix-millis fits
75 s.source.to_u8(),
76 s.name.to_u8(),
77 s.value.0,
78 s.tag.as_ref().map(|t| t.as_str()),
79 wire::encode(s),
80 ],
81 )?;
82 Ok(())
83 }
84
85 /// The most recent `limit` signals, newest first. This is the query that
86 /// lets a renderer read recent history instead of only the live snapshot.
87 pub fn recent(&self, limit: usize) -> rusqlite::Result<Vec<Signal>> {
88 self.query("SELECT frame FROM signals ORDER BY id DESC LIMIT ?1", limit)
89 }
90
91 /// The most recent `limit` signals for one metric name, newest first.
92 pub fn recent_named(
93 &self,
94 name: signal_schema::SignalName,
95 limit: usize,
96 ) -> rusqlite::Result<Vec<Signal>> {
97 let mut stmt = self.conn.prepare(
98 "SELECT frame FROM signals WHERE name = ?1 ORDER BY id DESC LIMIT ?2",
99 )?;
100 let rows = stmt.query_map(params![name.to_u8(), limit as i64], |row| {
101 row.get::<_, Vec<u8>>(0)
102 })?;
103 Self::decode_rows(rows)
104 }
105
106 fn query(&self, sql: &str, limit: usize) -> rusqlite::Result<Vec<Signal>> {
107 let mut stmt = self.conn.prepare(sql)?;
108 let rows = stmt.query_map(params![limit as i64], |row| row.get::<_, Vec<u8>>(0))?;
109 Self::decode_rows(rows)
110 }
111
112 fn decode_rows(
113 rows: impl Iterator<Item = rusqlite::Result<Vec<u8>>>,
114 ) -> rusqlite::Result<Vec<Signal>> {
115 let mut out = Vec::new();
116 for frame in rows {
117 let frame = frame?;
118 if let Some(sig) = wire::decode(&frame) {
119 out.push(sig);
120 }
121 }
122 Ok(out)
123 }
124}
125
126#[cfg(test)]
127mod tests {
128 use super::*;
129 use signal_schema::{SignalName, Source, Tag, Value, SCHEMA_VERSION};
130
131 fn sig(name: SignalName, value: f64, tag: Option<Tag>) -> Signal {
132 Signal {
133 schema_version: SCHEMA_VERSION,
134 ts: 1_723_100_000_000,
135 source: Source::Git,
136 name,
137 value: Value(value),
138 tag,
139 }
140 }
141
142 #[test]
143 fn round_trip_write_then_query_recent() {
144 let h = History::open_in_memory().unwrap();
145 let repo = Tag::repo_path("/x/repo").unwrap();
146 h.record(&sig(SignalName::CommitsWindow, 5.0, Some(repo.clone())))
147 .unwrap();
148 h.record(&sig(SignalName::CommitsToday, 2.0, Some(repo.clone())))
149 .unwrap();
150 h.record(&sig(SignalName::KeysPerMin, 88.0, None)).unwrap();
151
152 let recent = h.recent(10).unwrap();
153 assert_eq!(recent.len(), 3);
154 // Newest first, and the aggregates survive the round trip exactly.
155 assert_eq!(recent[0].name, SignalName::KeysPerMin);
156 assert_eq!(recent[0].value, Value(88.0));
157 assert_eq!(recent[2].name, SignalName::CommitsWindow);
158 assert_eq!(recent[2].value, Value(5.0));
159 assert_eq!(recent[2].tag.as_ref().unwrap().as_str(), "/x/repo");
160 }
161
162 #[test]
163 fn recent_named_filters_by_metric() {
164 let h = History::open_in_memory().unwrap();
165 for v in [1.0, 2.0, 3.0] {
166 h.record(&sig(SignalName::KeysPerMin, v, None)).unwrap();
167 }
168 h.record(&sig(SignalName::SessionSeconds, 42.0, None))
169 .unwrap();
170
171 let keys = h.recent_named(SignalName::KeysPerMin, 10).unwrap();
172 assert_eq!(keys.len(), 3);
173 assert!(keys.iter().all(|s| s.name == SignalName::KeysPerMin));
174 assert_eq!(keys[0].value, Value(3.0)); // newest first
175 }
176}