crates/signald/src/history.rs
255 lines · 9618 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//!
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.prune()?;
63 Ok(h)
64 }
65
66 /// An ephemeral in-memory store (used by tests). No retention.
67 pub fn open_in_memory() -> rusqlite::Result<History> {
68 Self::init(Connection::open_in_memory()?, None)
69 }
70
71 fn init(conn: Connection, retention_ms: Option<u64>) -> rusqlite::Result<History> {
72 // Columns for querying (name, ts, value, tag); `frame` is the audited
73 // wire encoding, reused for faithful, validated reconstruction on read.
74 conn.execute(
75 "CREATE TABLE IF NOT EXISTS signals (
76 id INTEGER PRIMARY KEY AUTOINCREMENT,
77 schema_version INTEGER NOT NULL,
78 ts INTEGER NOT NULL,
79 source INTEGER NOT NULL,
80 name INTEGER NOT NULL,
81 value REAL NOT NULL,
82 tag TEXT,
83 frame BLOB NOT NULL
84 )",
85 [],
86 )?;
87 conn.execute(
88 "CREATE INDEX IF NOT EXISTS idx_signals_name_ts ON signals(name, ts)",
89 [],
90 )?;
91 // For the retention delete.
92 conn.execute("CREATE INDEX IF NOT EXISTS idx_signals_ts ON signals(ts)", [])?;
93 Ok(History {
94 conn,
95 retention_ms,
96 since_prune: 0,
97 })
98 }
99
100 /// Delete rows older than the retention. Returns the number deleted.
101 pub fn prune(&self) -> rusqlite::Result<usize> {
102 let Some(retention_ms) = self.retention_ms else {
103 return Ok(0);
104 };
105 let cutoff = now_millis().saturating_sub(retention_ms) as i64;
106 self.conn
107 .execute("DELETE FROM signals WHERE ts < ?1", params![cutoff])
108 }
109
110 /// Append one signal to the history, pruning every [`PRUNE_EVERY`] calls.
111 pub fn record(&mut self, s: &Signal) -> rusqlite::Result<()> {
112 self.since_prune += 1;
113 if self.since_prune >= PRUNE_EVERY {
114 self.since_prune = 0;
115 self.prune()?;
116 }
117 self.conn.execute(
118 "INSERT INTO signals (schema_version, ts, source, name, value, tag, frame)
119 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
120 params![
121 s.schema_version,
122 s.ts as i64, // SQLite integers are i64; unix-millis fits
123 s.source.to_u8(),
124 s.name.to_u8(),
125 s.value.0,
126 s.tag.as_ref().map(|t| t.as_str()),
127 wire::encode(s),
128 ],
129 )?;
130 Ok(())
131 }
132
133 /// The most recent `limit` signals, newest first. This is the query that
134 /// lets a renderer read recent history instead of only the live snapshot.
135 pub fn recent(&self, limit: usize) -> rusqlite::Result<Vec<Signal>> {
136 self.query("SELECT frame FROM signals ORDER BY id DESC LIMIT ?1", limit)
137 }
138
139 /// The most recent `limit` signals for one metric name, newest first.
140 pub fn recent_named(
141 &self,
142 name: signal_schema::SignalName,
143 limit: usize,
144 ) -> rusqlite::Result<Vec<Signal>> {
145 let mut stmt = self.conn.prepare(
146 "SELECT frame FROM signals WHERE name = ?1 ORDER BY id DESC LIMIT ?2",
147 )?;
148 let rows = stmt.query_map(params![name.to_u8(), limit as i64], |row| {
149 row.get::<_, Vec<u8>>(0)
150 })?;
151 Self::decode_rows(rows)
152 }
153
154 fn query(&self, sql: &str, limit: usize) -> rusqlite::Result<Vec<Signal>> {
155 let mut stmt = self.conn.prepare(sql)?;
156 let rows = stmt.query_map(params![limit as i64], |row| row.get::<_, Vec<u8>>(0))?;
157 Self::decode_rows(rows)
158 }
159
160 fn decode_rows(
161 rows: impl Iterator<Item = rusqlite::Result<Vec<u8>>>,
162 ) -> rusqlite::Result<Vec<Signal>> {
163 let mut out = Vec::new();
164 for frame in rows {
165 let frame = frame?;
166 if let Some(sig) = wire::decode(&frame) {
167 out.push(sig);
168 }
169 }
170 Ok(out)
171 }
172}
173
174#[cfg(test)]
175mod tests {
176 use super::*;
177 use signal_schema::{SignalName, Source, Tag, Value, SCHEMA_VERSION};
178
179 fn sig(name: SignalName, value: f64, tag: Option<Tag>) -> Signal {
180 Signal {
181 schema_version: SCHEMA_VERSION,
182 ts: 1_723_100_000_000,
183 source: Source::Git,
184 name,
185 value: Value(value),
186 tag,
187 }
188 }
189
190 #[test]
191 fn round_trip_write_then_query_recent() {
192 let mut h = History::open_in_memory().unwrap();
193 let repo = Tag::repo_path("/x/repo").unwrap();
194 h.record(&sig(SignalName::CommitsWindow, 5.0, Some(repo.clone())))
195 .unwrap();
196 h.record(&sig(SignalName::CommitsToday, 2.0, Some(repo.clone())))
197 .unwrap();
198 h.record(&sig(SignalName::KeysPerMin, 88.0, None)).unwrap();
199
200 let recent = h.recent(10).unwrap();
201 assert_eq!(recent.len(), 3);
202 // Newest first, and the aggregates survive the round trip exactly.
203 assert_eq!(recent[0].name, SignalName::KeysPerMin);
204 assert_eq!(recent[0].value, Value(88.0));
205 assert_eq!(recent[2].name, SignalName::CommitsWindow);
206 assert_eq!(recent[2].value, Value(5.0));
207 assert_eq!(recent[2].tag.as_ref().unwrap().as_str(), "/x/repo");
208 }
209
210 #[test]
211 fn recent_named_filters_by_metric() {
212 let mut h = History::open_in_memory().unwrap();
213 for v in [1.0, 2.0, 3.0] {
214 h.record(&sig(SignalName::KeysPerMin, v, None)).unwrap();
215 }
216 h.record(&sig(SignalName::SessionSeconds, 42.0, None))
217 .unwrap();
218
219 let keys = h.recent_named(SignalName::KeysPerMin, 10).unwrap();
220 assert_eq!(keys.len(), 3);
221 assert!(keys.iter().all(|s| s.name == SignalName::KeysPerMin));
222 assert_eq!(keys[0].value, Value(3.0)); // newest first
223 }
224
225 #[test]
226 fn retention_prunes_old_rows_on_open_and_periodically() {
227 let dir = std::env::temp_dir().join(format!("signald-history-{}", std::process::id()));
228 std::fs::create_dir_all(&dir).unwrap();
229 let db = dir.join("h.sqlite");
230 let retention = Duration::from_secs(3600);
231 let ancient = Signal {
232 ts: 1,
233 ..sig(SignalName::KeysPerMin, 1.0, None)
234 };
235 let fresh = Signal {
236 ts: now_millis(),
237 ..sig(SignalName::KeysPerMin, 2.0, None)
238 };
239 {
240 let mut h = History::open_with_retention(&db, retention).unwrap();
241 h.record(&ancient).unwrap();
242 h.record(&fresh).unwrap();
243 assert_eq!(h.recent(10).unwrap().len(), 2);
244 assert_eq!(h.prune().unwrap(), 1);
245 assert_eq!(h.recent(10).unwrap().len(), 1);
246 // Left behind for the reopen to prune.
247 h.record(&ancient).unwrap();
248 }
249 let h = History::open_with_retention(&db, retention).unwrap();
250 let rows = h.recent(10).unwrap();
251 assert_eq!(rows.len(), 1, "reopen prunes");
252 assert_eq!(rows[0].value, Value(2.0));
253 let _ = std::fs::remove_dir_all(&dir);
254 }
255}