From 04475db7d64f87a55e9da4873e97584bf9f526cc Mon Sep 17 00:00:00 2001 From: Heng-Yi Wu <2316687+henry40408@users.noreply.github.com> Date: Mon, 14 Sep 2026 00:01:13 +0800 Subject: [PATCH] feat(db): maintain statistics rollups of query_logs Every dashboard and Statistics reading is a count, sum or histogram over a window. With the default retention that window is the whole table, so answering it from query_logs reads an index entry per logged query whichever index it uses. Migration 16 adds five rollup tables in which one row stands for every query sharing a key within a unit of time: query_stats_quarter, and per hour domain, client, upstream and metrics. Nothing reads them yet; the dashboard and the Statistics page move onto them in follow-up changes. - Inserts are maintained by an AFTER INSERT trigger, so rows written outside the logger (the e2e fixtures use the sqlite3 CLI) are counted too. Replaying 1.48 M queries in 500-row batches, the trigger writes 886 480 pages, against 885 858 for one grouped upsert per batch and 846 079 with no rollups. - prune_logs_before unwinds the rollups in its transaction: whole units before the cutoff are dropped, and the rows before the cutoff in the quarter and hour it falls inside are recounted and subtracted, so the prune keeps its exact cutoff. A one-day prune writes 5 151 pages against 5 020. - Clear All empties the rollups. - The migration fills them from existing rows (101 420 pages read, 3 714 written on the 1.48 M-row database) with a replacing upsert, so an interrupted migration can run again. Co-Authored-By: Claude Opus 5 (1M context) --- ARCHITECTURE.md | 13 ++ CLAUDE.md | 1 + src/db.rs | 415 +++++++++++++++++++++++++++++++++++++++++++++++- 3 files changed, 428 insertions(+), 1 deletion(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 310fd76..4b1b907 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -237,6 +237,7 @@ Everything is in a single SQLite file (`noadd.sqlite3` by default; a legacy `noa | `users` | Operator accounts (username, Argon2 password hash) | | `sessions` | Active admin sessions (token, user_id, ip, user agent, timestamps) | | `api_keys` | Programmatic API keys (BLAKE2b hash, owning user_id, `ON DELETE CASCADE`) | +| `query_stats_quarter`, `query_stats_{domain,client,upstream,metrics}_hour` | Rollups of `query_logs`: counts per quarter hour or hour and grouping key — see *Rollups* below | `query_logs` carries five indexes, all of them shaped by the statistics queries: @@ -260,6 +261,18 @@ The upstream index is partial because blocked and cached answers never reach an Indexes are not free here. On that same 103 MiB database `dbstat` attributes 20 MiB to `(domain, timestamp)`, 18 MiB to `(client_ip, doh_token, timestamp)`, 9 MiB to the metrics index and 7 MiB to `timestamp` — the two composites added for statistics cost about a quarter of the file. (Those are the figures from before version 15 replaced the client index.) Measuring an index by the file-size delta of `CREATE INDEX` understates it whenever the database is carrying a freelist, since the new pages come out of that first; `dbstat` reports the real figure. +### Rollups + +Every index above is still read one entry per logged query. With the default seven-day retention the Statistics page's windows, and the dashboard's 30-day summary, span the whole table, so no index choice can take those readings below the size of the index they read — 9 475 pages for the summary on a 1.48 M-row database, every dashboard tick. Version 16 adds five rollup tables in which one row stands for every query sharing a key within a unit of time: `query_stats_quarter` (blocked, cached; count and summed response time per quarter hour), and per hour `query_stats_domain_hour`, `query_stats_client_hour`, `query_stats_upstream_hour` (count and summed response time) and `query_stats_metrics_hour` (the grain the outcome, query-type and latency folds read). They are `WITHOUT ROWID`, with the unit first in the key, so a window is one range and the newest unit is where every write lands. On that database they total 3 631 pages of a 111 282-page file. + +The quarter hour is there because it is the finest bucket any chart draws and the unit every UTC offset in use is a whole number of; nothing else needs finer than the hour. `doh_token` is stored as `''` for plain DNS, since a key column cannot hold `NULL`. + +**The rollups must always equal a recount of `query_logs`**, because a reader folding them answers for the table. Inserts keep them there through an `AFTER INSERT` trigger (`query_logs_maintain_stats`) rather than through the logger, so rows that reach the table any other way — the e2e fixtures write theirs with the `sqlite3` CLI — are counted too. Replaying the same 1.48 M queries in the logger's 500-row batches, the trigger writes 886 480 pages against 885 858 for one grouped upsert per batch and 846 079 with no rollups at all. + +Deletes are not a trigger. A `DELETE` trigger would unwind a prune row by row and turn off SQLite's truncate optimisation for Clear All, so both do it in their own transaction instead: Clear All empties the five tables, and `prune_logs_before` calls `unwind_stats_rollups` before its delete. That drops whole units before the cutoff and, for the one quarter and one hour the cutoff falls inside, recounts the rows about to go and subtracts them — so the prune keeps its exact cutoff rather than rounding retention to the hour. Pruning a day from that database writes 5 151 pages and misses 13 971, against 5 020 and 12 282 without rollups. `rollups_follow_every_write_that_changes_query_logs` (`src/db.rs`) is the guard: it compares every table with its recount after batches, a direct SQL insert, prunes inside and on a unit boundary, and Clear All. + +Version 16 fills the rollups from the rows already logged, which on that database reads 101 420 pages and writes 3 714. The fill replaces rather than adds, so a migration interrupted before `user_version` moved is safe to run again. Nothing reads the rollups yet; the dashboard and the Statistics page move onto them in later changes. + ### Measuring these queries Index work here is measured in **page misses** — `SQLITE_DBSTATUS_CACHE_MISS`, the 4 KiB database pages SQLite has to fetch from the file — and not in milliseconds. Development happens on an SSD and the appliance runs off an SD card, where the same page count costs orders of magnitude more; a query that reads the whole table can look free on one machine and take seconds on the other. `tests/stats_page_miss_bench.rs` reports the figure against a real database and `tests/stats_page_miss_test.rs` asserts the properties behind it; `tests/dashboard_page_miss_bench.rs` does the same for the dashboard's first response and its recurring event-stream tick, and `tests/logs_page_miss_bench.rs` for every filter the query log offers, with values drawn from the database under test. All three take `BENCH_DB` and a `BENCH_NOW` that pins the clock, so a copy older than its windows still measures the traffic it holds; `tests/stats_parallel_bench.rs` is the wall-clock companion, and is the one to distrust when the two disagree. diff --git a/CLAUDE.md b/CLAUDE.md index 840372d..6434000 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -135,6 +135,7 @@ Statistics adds the conventions for a page whose readings sit in a **chosen wind - **This page is measured in page misses, not milliseconds.** Development is on an SSD and the appliance runs off an SD card, so a duration measured here says nothing about a Raspberry Pi; the pages a query fetches from the file are the same on both. `cargo nextest run --release --no-capture --run-ignored only stats_page_miss` with `BENCH_DB` pointed at a copy of a real database reports them; `dashboard_page_miss` and `logs_page_miss` do the same for the dashboard (first response and tick) and the query log (every filter), and `BENCH_NOW` pins the clock on a copy older than its windows. A wall-clock reading is the thing to distrust when the two disagree — it is what left the outcome breakdown scanning the whole table through version 11. - **One scan per index, not one per reading.** `stats_scan_since` and `traffic_lists_since` (`src/db.rs`) are the page's two scans — the second answering top domains, the distinct-domain count and top clients from one grouping of `(domain, client_ip, doh_token)` — and every reading on it — the charts included — is folded out of one of them; `compute_range_stats` (`src/admin/stats.rs`) is what the page calls. `stats_scan_since` streams its rows and folds them in Rust rather than grouping in SQL: a grain carrying both the quarter hour and `response_ms` approaches a group per row, which would be a temp b-tree the size of the window. The single-purpose functions `/api/stats/*` uses are statements of their own — adding a seventh reading to the page means folding it out of one of those two scans, not adding a statement. - **A total nobody can count cheaply is maintained, not counted.** `query_logs`' row count lives in `settings` (`query_log_count`), moved by the insert batch, the prune and Clear All inside their own transactions — `SELECT COUNT(*)` walks an index end to end, and both the Database Health card and the query log's pager (whenever no filter is applied, via `count_logs`) ask on every load. A fourth write path to `query_logs` means a fourth `bump_log_count`, not a fourth reader. +- **The same holds for the statistics rollups** (`query_stats_*`, see ARCHITECTURE.md *Rollups*), which must always equal a recount of `query_logs`. Inserts are covered by the `query_logs_maintain_stats` trigger whatever writes them; a new path that *deletes* from `query_logs` has to unwind them in its own transaction, as `prune_logs_before` (`unwind_stats_rollups`) and Clear All do. - Four bar lists in one template share **one askama macro** (`templates/_macros.html`); `{% call … %}` needs a matching `{% endcall %}` in askama 0.16, and `{% include %}` cannot see a loop variable at all. Account adds the conventions for **actions that need a password proof**: diff --git a/src/db.rs b/src/db.rs index 960a001..4b1bd56 100644 --- a/src/db.rs +++ b/src/db.rs @@ -363,6 +363,129 @@ pub struct LatencySummary { /// place a counter can live without a table of its own. const QUERY_LOG_COUNT_KEY: &str = "query_log_count"; +/// Width of one `query_stats_quarter` row, in milliseconds. +const ROLLUP_QUARTER_MS: i64 = QUARTER_SECS * 1000; + +/// Width of one row of the hourly rollups, in milliseconds. +const ROLLUP_HOUR_MS: i64 = 3_600_000; + +/// Pre-aggregated counts of `query_logs`, one table per grain a reader folds. +/// +/// Every statistic the dashboard and the Statistics page show is a count, a +/// sum or a histogram over a time window, and answering one from the table +/// reads an index entry per logged query: with the default retention the +/// window is the whole table, so no index can narrow it. A rollup row stands +/// for every query that shares its key within its unit of time, so the same +/// window reads a few thousand rows instead of a few million. +/// +/// The grains are the coarsest that still answer every reader exactly. The +/// quarter hour is the finest bucket any chart draws and the unit every UTC +/// offset in use is a whole number of; lists and histograms need no finer than +/// the hour. `doh_token` is stored as `''` for plain DNS because a primary key +/// column cannot hold `NULL`, and `has_result` is spelled out rather than read +/// from the generated column so the trigger does not depend on migration 12. +/// +/// Inserts are kept here by a trigger rather than by the logger, so rows that +/// reach the table any other way — the e2e fixtures write theirs with the +/// `sqlite3` CLI — are counted too. Measured by replaying 1.48 M logged +/// queries in the logger's 500-row batches, the trigger writes the same pages +/// as one grouped upsert per batch (886 480 against 885 858). Deletes are not a +/// trigger: a `DELETE` trigger would cost a row-by-row unwind on every prune +/// and switch off `SQLite`'s truncate optimisation for Clear All, so +/// [`unwind_stats_rollups`] does it in the statement's own transaction. +const STATS_ROLLUP_SCHEMA: &str = " + CREATE TABLE IF NOT EXISTS query_stats_quarter ( + quarter INTEGER NOT NULL, + blocked INTEGER NOT NULL, + cached INTEGER NOT NULL, + count INTEGER NOT NULL, + sum_ms INTEGER NOT NULL, + PRIMARY KEY (quarter, blocked, cached) + ) WITHOUT ROWID; + CREATE TABLE IF NOT EXISTS query_stats_domain_hour ( + hour INTEGER NOT NULL, + domain TEXT NOT NULL, + count INTEGER NOT NULL, + PRIMARY KEY (hour, domain) + ) WITHOUT ROWID; + CREATE TABLE IF NOT EXISTS query_stats_client_hour ( + hour INTEGER NOT NULL, + client_ip TEXT NOT NULL, + doh_token TEXT NOT NULL, + count INTEGER NOT NULL, + PRIMARY KEY (hour, client_ip, doh_token) + ) WITHOUT ROWID; + CREATE TABLE IF NOT EXISTS query_stats_upstream_hour ( + hour INTEGER NOT NULL, + upstream TEXT NOT NULL, + count INTEGER NOT NULL, + sum_ms INTEGER NOT NULL, + PRIMARY KEY (hour, upstream) + ) WITHOUT ROWID; + CREATE TABLE IF NOT EXISTS query_stats_metrics_hour ( + hour INTEGER NOT NULL, + blocked INTEGER NOT NULL, + cached INTEGER NOT NULL, + has_result INTEGER NOT NULL, + query_type TEXT NOT NULL, + response_ms INTEGER NOT NULL, + count INTEGER NOT NULL, + PRIMARY KEY (hour, blocked, cached, has_result, query_type, response_ms) + ) WITHOUT ROWID; + CREATE TRIGGER IF NOT EXISTS query_logs_maintain_stats AFTER INSERT ON query_logs BEGIN + INSERT INTO query_stats_quarter (quarter, blocked, cached, count, sum_ms) + VALUES (NEW.timestamp / 900000, NEW.blocked, NEW.cached, 1, NEW.response_ms) + ON CONFLICT DO UPDATE SET count = count + 1, sum_ms = sum_ms + excluded.sum_ms; + INSERT INTO query_stats_domain_hour (hour, domain, count) + VALUES (NEW.timestamp / 3600000, NEW.domain, 1) + ON CONFLICT DO UPDATE SET count = count + 1; + INSERT INTO query_stats_client_hour (hour, client_ip, doh_token, count) + VALUES (NEW.timestamp / 3600000, NEW.client_ip, COALESCE(NEW.doh_token, ''), 1) + ON CONFLICT DO UPDATE SET count = count + 1; + INSERT INTO query_stats_upstream_hour (hour, upstream, count, sum_ms) + SELECT NEW.timestamp / 3600000, NEW.upstream, 1, NEW.response_ms + WHERE NEW.upstream IS NOT NULL + ON CONFLICT DO UPDATE SET count = count + 1, sum_ms = sum_ms + excluded.sum_ms; + INSERT INTO query_stats_metrics_hour + (hour, blocked, cached, has_result, query_type, response_ms, count) + VALUES (NEW.timestamp / 3600000, NEW.blocked, NEW.cached, + NEW.result IS NOT NULL AND NEW.result != '', + NEW.query_type, NEW.response_ms, 1) + ON CONFLICT DO UPDATE SET count = count + 1; + END; +"; + +/// Fills the rollups from what `query_logs` already holds. An upsert that +/// replaces rather than adds, so a migration interrupted after this ran and +/// before `user_version` moved can run it again without doubling anything. +/// +/// `WHERE true` is what lets an upsert follow a `SELECT` — without it the +/// parser reads `ON CONFLICT` as part of the `SELECT`. +const STATS_ROLLUP_BACKFILL: &str = " + INSERT INTO query_stats_quarter (quarter, blocked, cached, count, sum_ms) + SELECT timestamp / 900000, blocked, cached, COUNT(*), SUM(response_ms) + FROM query_logs WHERE true GROUP BY 1, 2, 3 + ON CONFLICT DO UPDATE SET count = excluded.count, sum_ms = excluded.sum_ms; + INSERT INTO query_stats_domain_hour (hour, domain, count) + SELECT timestamp / 3600000, domain, COUNT(*) + FROM query_logs WHERE true GROUP BY 1, 2 + ON CONFLICT DO UPDATE SET count = excluded.count; + INSERT INTO query_stats_client_hour (hour, client_ip, doh_token, count) + SELECT timestamp / 3600000, client_ip, COALESCE(doh_token, ''), COUNT(*) + FROM query_logs WHERE true GROUP BY 1, 2, 3 + ON CONFLICT DO UPDATE SET count = excluded.count; + INSERT INTO query_stats_upstream_hour (hour, upstream, count, sum_ms) + SELECT timestamp / 3600000, upstream, COUNT(*), SUM(response_ms) + FROM query_logs WHERE upstream IS NOT NULL GROUP BY 1, 2 + ON CONFLICT DO UPDATE SET count = excluded.count, sum_ms = excluded.sum_ms; + INSERT INTO query_stats_metrics_hour + (hour, blocked, cached, has_result, query_type, response_ms, count) + SELECT timestamp / 3600000, blocked, cached, result IS NOT NULL AND result != '', + query_type, response_ms, COUNT(*) + FROM query_logs WHERE true GROUP BY 1, 2, 3, 4, 5, 6 + ON CONFLICT DO UPDATE SET count = excluded.count; +"; + /// Default rusqlite cache is 16 statements; the read connection alone has /// ~20 distinct hot SQL strings (settings, stats, filter, token lookup), /// so anything below ~32 starts evicting on every admin poll. @@ -918,7 +1041,16 @@ impl Database { )?; } - const LATEST_VERSION: i64 = 15; + if version < 16 { + // The rollups the dashboard and the Statistics page will fold + // instead of scanning the table; see `STATS_ROLLUP_SCHEMA`. Filled + // from the rows already logged, which on a 1.48 M-row database + // read 101 420 pages and wrote 3 714. + conn.execute_batch(STATS_ROLLUP_SCHEMA)?; + conn.execute_batch(STATS_ROLLUP_BACKFILL)?; + } + + const LATEST_VERSION: i64 = 16; if version < LATEST_VERSION { conn.pragma_update(None, "user_version", LATEST_VERSION)?; } @@ -1127,6 +1259,13 @@ impl Database { .call(|conn| { let tx = conn.transaction()?; tx.execute("DELETE FROM query_logs", [])?; + tx.execute_batch( + "DELETE FROM query_stats_quarter; + DELETE FROM query_stats_domain_hour; + DELETE FROM query_stats_client_hour; + DELETE FROM query_stats_upstream_hour; + DELETE FROM query_stats_metrics_hour;", + )?; set_log_count(&tx, 0)?; tx.commit()?; Ok(()) @@ -1142,6 +1281,7 @@ impl Database { .conn .call(move |conn| { let tx = conn.transaction()?; + unwind_stats_rollups(&tx, timestamp_ms)?; let deleted = tx.execute( "DELETE FROM query_logs WHERE timestamp < ?1", params![timestamp_ms], @@ -2878,6 +3018,74 @@ fn set_log_count(conn: &rusqlite::Connection, count: i64) -> rusqlite::Result<() Ok(()) } +/// Take out of the rollups every row `DELETE FROM query_logs WHERE timestamp < +/// cutoff_ms` is about to remove. Must run before that delete, in its +/// transaction. +/// +/// Whole units before the cutoff are dropped. The unit the cutoff falls inside +/// loses only the rows before it, which are recounted from the table and +/// subtracted — so the prune keeps its exact cutoff and the rollups still agree +/// with the table afterwards, instead of either rounding the retention to the +/// hour or keeping counts for rows that are gone. That recount reads at most one +/// quarter's and one hour's worth of rows. +fn unwind_stats_rollups(conn: &rusqlite::Connection, cutoff_ms: i64) -> rusqlite::Result<()> { + let quarter_start = cutoff_ms.div_euclid(ROLLUP_QUARTER_MS) * ROLLUP_QUARTER_MS; + let hour_start = cutoff_ms.div_euclid(ROLLUP_HOUR_MS) * ROLLUP_HOUR_MS; + + conn.prepare_cached( + "INSERT INTO query_stats_quarter (quarter, blocked, cached, count, sum_ms) + SELECT timestamp / 900000, blocked, cached, -COUNT(*), -SUM(response_ms) + FROM query_logs WHERE timestamp >= ?1 AND timestamp < ?2 GROUP BY 1, 2, 3 + ON CONFLICT DO UPDATE SET count = count + excluded.count, + sum_ms = sum_ms + excluded.sum_ms", + )? + .execute(params![quarter_start, cutoff_ms])?; + conn.prepare_cached( + "DELETE FROM query_stats_quarter WHERE quarter < ?1 OR (quarter = ?1 AND count = 0)", + )? + .execute(params![quarter_start / ROLLUP_QUARTER_MS])?; + + for sql in [ + "INSERT INTO query_stats_domain_hour (hour, domain, count) + SELECT timestamp / 3600000, domain, -COUNT(*) + FROM query_logs WHERE timestamp >= ?1 AND timestamp < ?2 GROUP BY 1, 2 + ON CONFLICT DO UPDATE SET count = count + excluded.count", + "INSERT INTO query_stats_client_hour (hour, client_ip, doh_token, count) + SELECT timestamp / 3600000, client_ip, COALESCE(doh_token, ''), -COUNT(*) + FROM query_logs WHERE timestamp >= ?1 AND timestamp < ?2 GROUP BY 1, 2, 3 + ON CONFLICT DO UPDATE SET count = count + excluded.count", + "INSERT INTO query_stats_upstream_hour (hour, upstream, count, sum_ms) + SELECT timestamp / 3600000, upstream, -COUNT(*), -SUM(response_ms) + FROM query_logs + WHERE timestamp >= ?1 AND timestamp < ?2 AND upstream IS NOT NULL GROUP BY 1, 2 + ON CONFLICT DO UPDATE SET count = count + excluded.count, + sum_ms = sum_ms + excluded.sum_ms", + "INSERT INTO query_stats_metrics_hour + (hour, blocked, cached, has_result, query_type, response_ms, count) + SELECT timestamp / 3600000, blocked, cached, result IS NOT NULL AND result != '', + query_type, response_ms, -COUNT(*) + FROM query_logs WHERE timestamp >= ?1 AND timestamp < ?2 + GROUP BY 1, 2, 3, 4, 5, 6 + ON CONFLICT DO UPDATE SET count = count + excluded.count", + ] { + conn.prepare_cached(sql)? + .execute(params![hour_start, cutoff_ms])?; + } + let hour = hour_start / ROLLUP_HOUR_MS; + for table in [ + "query_stats_domain_hour", + "query_stats_client_hour", + "query_stats_upstream_hour", + "query_stats_metrics_hour", + ] { + conn.prepare_cached(&format!( + "DELETE FROM {table} WHERE hour < ?1 OR (hour = ?1 AND count = 0)" + ))? + .execute(params![hour])?; + } + Ok(()) +} + /// The maintained `query_logs` row count, counted instead only when the /// counter row is missing or unreadable. fn read_log_count(conn: &rusqlite::Connection) -> rusqlite::Result { @@ -3762,4 +3970,209 @@ mod tests { "data should remain queryable after maintenance" ); } + + /// Each rollup, the same grouping recounted from `query_logs`, and the + /// ordering that makes the two comparable row for row. + const ROLLUP_RECOUNTS: &[(&str, &str, &str)] = &[ + ( + "query_stats_quarter", + "SELECT quarter, blocked, cached, count, sum_ms FROM query_stats_quarter", + "SELECT timestamp / 900000, blocked, cached, COUNT(*), SUM(response_ms) \ + FROM query_logs GROUP BY 1, 2, 3", + ), + ( + "query_stats_domain_hour", + "SELECT hour, domain, count FROM query_stats_domain_hour", + "SELECT timestamp / 3600000, domain, COUNT(*) FROM query_logs GROUP BY 1, 2", + ), + ( + "query_stats_client_hour", + "SELECT hour, client_ip, doh_token, count FROM query_stats_client_hour", + "SELECT timestamp / 3600000, client_ip, COALESCE(doh_token, ''), COUNT(*) \ + FROM query_logs GROUP BY 1, 2, 3", + ), + ( + "query_stats_upstream_hour", + "SELECT hour, upstream, count, sum_ms FROM query_stats_upstream_hour", + "SELECT timestamp / 3600000, upstream, COUNT(*), SUM(response_ms) \ + FROM query_logs WHERE upstream IS NOT NULL GROUP BY 1, 2", + ), + ( + "query_stats_metrics_hour", + "SELECT hour, blocked, cached, has_result, query_type, response_ms, count \ + FROM query_stats_metrics_hour", + "SELECT timestamp / 3600000, blocked, cached, (result IS NOT NULL AND result != ''), \ + query_type, response_ms, COUNT(*) FROM query_logs GROUP BY 1, 2, 3, 4, 5, 6", + ), + ]; + + /// Every row of `sql`, rendered and sorted, so two spellings of the same + /// grouping compare equal regardless of the order either returns. + fn rendered_rows(conn: &rusqlite::Connection, sql: &str) -> Vec { + let mut stmt = conn.prepare(sql).unwrap(); + let columns = stmt.column_count(); + let mut rows: Vec = stmt + .query_map([], |row| { + Ok((0..columns) + .map(|i| format!("{:?}", row.get_ref(i).unwrap())) + .collect::>() + .join("|")) + }) + .unwrap() + .collect::>() + .unwrap(); + rows.sort(); + rows + } + + /// Asserts every rollup holds exactly what recounting `query_logs` gives. + fn assert_rollups_match(path: &str, step: &str) { + let conn = rusqlite::Connection::open(path).unwrap(); + for (table, rollup, recount) in ROLLUP_RECOUNTS { + assert_eq!( + rendered_rows(&conn, rollup), + rendered_rows(&conn, recount), + "{table} disagrees with query_logs after {step}" + ); + } + } + + fn rollup_entry(timestamp: i64, i: i64) -> QueryLogEntry { + QueryLogEntry { + timestamp, + domain: format!("host{}.example.com", i % 3), + query_type: if i % 2 == 0 { "A" } else { "AAAA" }.to_string(), + client_ip: format!("10.0.0.{}", i % 2), + blocked: i % 4 == 0, + cached: i % 5 == 0, + upstream: (i % 3 != 0).then(|| format!("udp://9.9.9.{}:53", i % 2)), + doh_token: (i % 2 == 0).then(|| "phone".to_string()), + // All three shapes `has_result` distinguishes: none, empty, an answer. + result: match i % 3 { + 0 => None, + 1 => Some(String::new()), + _ => Some("1.2.3.4".to_string()), + }, + response_ms: 1 + i % 7, + authenticated_data: false, + } + } + + /// The rollups are only as good as their agreement with `query_logs`: a + /// reader that folds them answers for the table, so every write that changes + /// the table has to change them identically — including a prune whose + /// cutoff falls inside a quarter and an hour, which leaves those units half + /// deleted, and rows written straight into the table as the e2e fixtures do. + #[tokio::test] + async fn rollups_follow_every_write_that_changes_query_logs() { + const HOUR: i64 = 3_600_000; + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("rollups.db"); + let path_str = path.to_str().unwrap().to_string(); + let db = Database::open(&path_str).await.unwrap(); + + // Straddles quarter and hour boundaries on both sides. + let offsets = [ + 0, + 899_999, + 900_000, + 1_000_000, + 2_700_001, + HOUR - 1, + HOUR + 5, + 2 * HOUR + 30_000, + ]; + let first: Vec = offsets + .iter() + .enumerate() + .map(|(i, off)| rollup_entry(10 * HOUR + off, i as i64)) + .collect(); + db.insert_query_logs(&first).await.unwrap(); + assert_rollups_match(&path_str, "the first batch"); + + // Same units again, so every rollup row has to accumulate, not replace. + let second: Vec = offsets + .iter() + .enumerate() + .map(|(i, off)| rollup_entry(10 * HOUR + off + 1, i as i64 + 1)) + .collect(); + db.insert_query_logs(&second).await.unwrap(); + assert_rollups_match(&path_str, "a batch into the same units"); + + { + let conn = rusqlite::Connection::open(&path_str).unwrap(); + conn.execute_batch(&format!( + "INSERT INTO query_logs (timestamp, domain, query_type, client_ip, blocked, \ + cached, response_ms, upstream, doh_token, result, authenticated_data) VALUES \ + ({}, 'host0.example.com', 'A', '10.0.0.9', 0, 1, 40, NULL, NULL, '9.9.9.9', 0);", + 10 * HOUR + 1_200_000 + )) + .unwrap(); + } + assert_rollups_match(&path_str, "a row written straight into the table"); + + // Inside the second quarter of hour 10: both units are cut in half. + let cutoff_ms = 10 * HOUR + 1_000_000; + let pruned = db.prune_logs_before(cutoff_ms / 1000).await.unwrap(); + assert!(pruned > 0, "the prune has to remove something to test"); + assert_rollups_match(&path_str, "a prune inside a quarter and an hour"); + + assert_eq!(db.prune_logs_before(cutoff_ms / 1000).await.unwrap(), 0); + assert_rollups_match(&path_str, "a prune that matched nothing"); + + // On an hour boundary: whole units go, none is left partial. + db.prune_logs_before(11 * HOUR / 1000).await.unwrap(); + assert_rollups_match(&path_str, "a prune on an hour boundary"); + + db.delete_all_logs().await.unwrap(); + assert_rollups_match(&path_str, "clearing the log"); + + db.insert_query_logs(&first).await.unwrap(); + assert_rollups_match(&path_str, "a batch after clearing"); + } + + /// A database from before version 16 has to come out of `open` with its + /// rollups already holding what its table holds, and with the trigger that + /// keeps them there — a reader folding an empty rollup would report no + /// traffic for the whole retention window. + #[tokio::test] + async fn migration_v16_backfills_the_rollups() { + const HOUR: i64 = 3_600_000; + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("v15.db"); + let path_str = path.to_str().unwrap().to_string(); + + let entries: Vec = (0..24) + .map(|i| rollup_entry(20 * HOUR + i * 700_000, i)) + .collect(); + { + let db = Database::open(&path_str).await.unwrap(); + db.insert_query_logs(&entries).await.unwrap(); + db.close().await; + } + + // Wind it back to version 15: the rows stay, the rollups do not. + { + let conn = rusqlite::Connection::open(&path_str).unwrap(); + conn.execute_batch( + "DROP TRIGGER query_logs_maintain_stats; + DROP TABLE query_stats_quarter; + DROP TABLE query_stats_domain_hour; + DROP TABLE query_stats_client_hour; + DROP TABLE query_stats_upstream_hour; + DROP TABLE query_stats_metrics_hour; + PRAGMA user_version = 15;", + ) + .unwrap(); + } + + let migrated = Database::open(&path_str).await.unwrap(); + assert_rollups_match(&path_str, "the migration"); + + migrated + .insert_query_logs(&[rollup_entry(21 * HOUR + 5, 99)]) + .await + .unwrap(); + assert_rollups_match(&path_str, "an insert into the migrated database"); + } }