diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index ffd9453..4adc4d0 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -239,15 +239,17 @@ Everything is in a single SQLite file (`noadd.sqlite3` by default; a legacy `noa | `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: +`query_logs` carries five indexes. Since version 17 every one of them serves the query log or the rollup readers' table arms; no statistic scans an index any more: | Index | Serves | | --- | --- | -| `timestamp` | the time-window filter every stats query starts with | +| `timestamp` | the rollup readers' table arms, and the query log's unfiltered and domain-contains pages | | `(domain, timestamp)` | the query log's domain search | -| `(timestamp, domain, client_ip, doh_token)` | nothing since the top domain and client lists moved onto the rollups (*Rollups* below) | -| `(timestamp, blocked, cached, response_ms, query_type, has_result)` | the Statistics page's scan, the `/api/stats/*` timeline and breakdowns, the query log's action and type filters | -| `(timestamp, upstream, response_ms) WHERE upstream IS NOT NULL` | nothing since top upstreams moved onto the rollups | +| `(doh_token, timestamp)` | the query log's token filter | +| `(blocked, timestamp)` | the query log's blocked/allowed filter | +| `(query_type, blocked, timestamp)` | the query log's type filter, alone or with a verdict | + +The rest of this section is how the indexes before version 17 were chosen; *Query log filters* below is how the current ones were. `(domain, timestamp)` puts the grouped column first and `timestamp` last, which made it covering for a `GROUP BY … WHERE timestamp >= ?` shape — the aggregation reads the index alone instead of scanning the window and building a temp b-tree over it. The client index that sat beside it, `(client_ip, doh_token, timestamp)`, did the same for top clients and took them from 143 ms to 20 ms on a 447 k-row database. The cost of that order is that the window cannot restrict it: a 24-hour question skip-scans the whole index, one seek per distinct group. @@ -261,6 +263,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. +### Query log filters + +Once every statistic folded the rollups, three indexes had no statistic left to serve — `(timestamp, domain, client_ip, doh_token)`, the metrics index and the upstream index — and version 17 drops them. What was left to index was the query log's filters, which a timestamp-first index can only serve by walking the window: on a 1.48 M-row database the newest page of a quiet `DoH` token read 1 200 pages and its count 23 130, and a quiet record type 9 812 for its first page. Each filter now leads its own index with what it matches, and those readings are 28, 15 and 19 pages. + +`(query_type, blocked, timestamp)` carries `blocked` in the middle so the type and verdict filters together are one seek, and that is what stops a type on its own from coming back in page order — asked as one range, the planner reads every row of the type and sorts them. `query_logs` asks it as two runs instead, one per verdict, each already newest first and cut at `offset + limit`, and merges them. The runs carry `id` and `timestamp` only, which the index holds, and the table is read for the page's rows alone: carrying every column looked up each row the runs passed over, 290 pages for page 20 of a busy type against 22. Page 1 of a busy type is the one reading left above where it was, 14 pages against 12, because it now reads two runs where the metrics index read one. `a_query_type_filter_pages_exactly_like_the_table` (`tests/db_test.rs`) holds the merge to the plain statement for every page size and offset `/api/logs` accepts. + +`(blocked, timestamp)` exists because the metrics index was also what answered the blocked filter without a table lookup. Dropping it alone left page 20 of the blocked and allowed filters at 102 and 57 pages against 33 and 18; with the index they are 18 and 12. + +Across the 42 readings `logs_page_miss_bench` makes, the query log went from 368 963 page misses to 159 649 on the 1.48 M-row database and from 100 029 to 44 968 on a 370 k-row one; the dashboard and the Statistics page read exactly what they did. Replaying the 1.48 M queries in the logger's batches, the five indexes write 957 160 pages against 885 858 for the ones they replace, and the file holds 92 224 pages against 111 291. Migrating that database takes 3 s on an SSD and leaves a freelist of 25%, past `VACUUM_FREELIST_RATIO`, so the first hourly maintenance rewrites the file once — to 86 876 pages from 119 603. Measure it on the appliance before relying on either figure there. + +`the_query_log_filters_seek_their_indexes` (`tests/stats_page_miss_test.rs`) holds each filter to a small fraction of the file, and `every_database_opens_to_the_same_query_log_indexes` (`src/db.rs`) opens databases from versions 9, 10, 11 and 16 and a fresh one and requires the same five indexes of each, then runs every reader that once named a dropped index. + ### 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. @@ -299,7 +313,7 @@ The Database Health card's row count is the one reading that is not a scan of an The dashboard pays for its readings every 10 seconds rather than once a visit, which makes a scan it repeats the most expensive kind. Its summary asked two statements for totals and blocks, then cache hits and latency, over the same 30 days of the metrics index; `summary_multi_since` moves the allowed-only filter from the `WHERE` into each `CASE` and answers both from one scan, 2 152 pages a tick instead of 4 304. With the upstream index, a tick on that database dropped from 8 304 page misses to 4 755. Those readings have since moved onto the rollups (see *Rollups*), which no longer read either index. -`INDEXED BY` appears on every statement that reads this index, in both directions. The ones that needed `blocked`, `cached` or `has_result` named `idx_query_logs_ts_metrics` because the planner otherwise takes the smaller `idx_query_logs_timestamp` and pays a rowid lookup per row — none is left, every such reading having moved onto the rollups; the rollup readers' table arms name `idx_query_logs_timestamp`, where that lookup is the point — they read at most one unit of rows; the heatmap, which reads `timestamp` and nothing else, names `idx_query_logs_timestamp` for the opposite reason — left alone the planner took the metrics index and read 2 153 pages where 1 386 answer it. +`INDEXED BY` appears on every statement that reads this index, in both directions. The ones that needed `blocked`, `cached` or `has_result` named `idx_query_logs_ts_metrics` because the planner otherwise takes the smaller `idx_query_logs_timestamp` and pays a rowid lookup per row — none is left, every such reading having moved onto the rollups, and version 17 dropped the index; the query log's type runs name `idx_query_logs_type_blocked_ts`, so each run is the seek the merge depends on; the rollup readers' table arms name `idx_query_logs_timestamp`, where that lookup is the point — they read at most one unit of rows; the heatmap, which reads `timestamp` and nothing else, names `idx_query_logs_timestamp` for the opposite reason — left alone the planner took the metrics index and read 2 153 pages where 1 386 answer it. Every index migration runs `ANALYZE`. A new index alone is not always enough — the planner keeps its old plan until `sqlite_stat1` is refreshed — and the hourly `PRAGMA optimize` lets those statistics drift a long way in the meantime. diff --git a/src/db.rs b/src/db.rs index 6b5edd9..dcf5056 100644 --- a/src/db.rs +++ b/src/db.rs @@ -252,16 +252,16 @@ pub struct MetricsBucket { pub struct WindowMetricsRow { pub blocked: bool, pub cached: bool, - /// Whether `result` held an answer. Carried by `idx_query_logs_ts_metrics` - /// as a generated column so classifying an outcome never reads the table. + /// Whether `result` held an answer. Carried by `query_stats_metrics_hour` + /// so classifying an outcome never reads the table. pub has_result: bool, pub query_type: String, pub response_ms: i64, pub count: i64, } -/// The three Statistics readings that come off `idx_query_logs_ts_metrics` in -/// one scan, answered together by [`Database::window_metrics_since`]. +/// The three Statistics readings folded out of `query_stats_metrics_hour` in +/// one statement, answered together by [`Database::window_metrics_since`]. #[derive(Debug, Clone)] pub struct WindowMetrics { pub outcomes: Vec<(String, i64)>, @@ -309,8 +309,8 @@ pub struct QuarterSeries { pub heatmap: Vec, } -/// Everything the Statistics page reads off `idx_query_logs_ts_metrics`, from -/// the one scan [`Database::stats_scan_since`] makes. +/// Everything the Statistics page folds out of the quarter and metrics +/// rollups, from the one statement [`Database::stats_scan_since`] makes. #[derive(Debug, Clone)] pub struct StatsScan { pub metrics: WindowMetrics, @@ -739,13 +739,9 @@ impl Database { ); CREATE INDEX IF NOT EXISTS idx_query_logs_timestamp ON query_logs(timestamp); CREATE INDEX IF NOT EXISTS idx_query_logs_domain_ts ON query_logs(domain, timestamp); - CREATE INDEX IF NOT EXISTS idx_query_logs_ts_domain_client ON query_logs(timestamp, domain, client_ip, doh_token); - -- Without has_result: a database predating version 12 has - -- no such column until that migration adds it, and this - -- batch runs first. Migration 12 rebuilds the index with it - -- for fresh and legacy databases alike. - CREATE INDEX IF NOT EXISTS idx_query_logs_ts_metrics ON query_logs(timestamp, blocked, cached, response_ms, query_type); - CREATE INDEX IF NOT EXISTS idx_query_logs_ts_upstream ON query_logs(timestamp, upstream, response_ms) WHERE upstream IS NOT NULL; + CREATE INDEX IF NOT EXISTS idx_query_logs_token_ts ON query_logs(doh_token, timestamp); + CREATE INDEX IF NOT EXISTS idx_query_logs_blocked_ts ON query_logs(blocked, timestamp); + CREATE INDEX IF NOT EXISTS idx_query_logs_type_blocked_ts ON query_logs(query_type, blocked, timestamp); CREATE TABLE IF NOT EXISTS filter_lists ( id INTEGER PRIMARY KEY AUTOINCREMENT, @@ -1061,7 +1057,45 @@ impl Database { conn.execute_batch(STATS_ROLLUP_BACKFILL)?; } - const LATEST_VERSION: i64 = 16; + if version < 17 { + // Every statistic now folds the rollups, which left three indexes + // with no statistic to serve: the metrics, upstream and + // domain/client indexes were timestamp-first scans for readings + // that no longer scan. The query log's filters are what is left to + // index, and a timestamp-first index can only filter them by + // walking the window: a quiet token or record type walked the whole + // table for its first page and again for its count. Each filter + // gets an index that leads with what it matches instead — see + // `query_logs` for why the type index carries `blocked`. + // + // The blocked filter used to be answered off the metrics index + // without a table lookup, so dropping that index alone made its + // deep pages three times the cost; `(blocked, timestamp)` puts them + // below where they were. Replaying 1.48 M logged queries in the + // logger's batches, the five indexes this leaves write 957 160 + // pages against 885 858 before, in a file of 92 224 pages against + // 111 291. + // + // A database from before version 17 also carries the indexes the + // earlier steps created on the way here, so they are dropped rather + // than never made. On a 1.48 M-row database the freed pages put + // the freelist past `VACUUM_FREELIST_RATIO`, so the first hourly + // maintenance after upgrading rewrites the file once. + conn.execute_batch( + "DROP INDEX IF EXISTS idx_query_logs_ts_metrics; + DROP INDEX IF EXISTS idx_query_logs_ts_upstream; + DROP INDEX IF EXISTS idx_query_logs_ts_domain_client; + CREATE INDEX IF NOT EXISTS idx_query_logs_token_ts \ + ON query_logs(doh_token, timestamp); + CREATE INDEX IF NOT EXISTS idx_query_logs_blocked_ts \ + ON query_logs(blocked, timestamp); + CREATE INDEX IF NOT EXISTS idx_query_logs_type_blocked_ts \ + ON query_logs(query_type, blocked, timestamp); + ANALYZE;", + )?; + } + + const LATEST_VERSION: i64 = 17; if version < LATEST_VERSION { conn.pragma_update(None, "user_version", LATEST_VERSION)?; } @@ -1164,7 +1198,8 @@ impl Database { let rows = self .reader() .call(move |conn| { - let mut sql = "SELECT timestamp, domain, query_type, client_ip, blocked, cached, response_ms, upstream, doh_token, result, authenticated_data FROM query_logs WHERE 1=1".to_string(); + const COLUMNS: &str = "timestamp, domain, query_type, client_ip, blocked, cached, response_ms, upstream, doh_token, result, authenticated_data"; + let mut sql = format!("SELECT {COLUMNS} FROM query_logs WHERE 1=1"); let mut param_values = append_log_filters( &mut sql, search.as_deref(), @@ -1172,9 +1207,54 @@ impl Database { token.as_deref(), query_type.as_deref(), ); - sql.push_str(" ORDER BY timestamp DESC LIMIT ? OFFSET ?"); - param_values.push(Box::new(limit)); - param_values.push(Box::new(offset)); + if let (Some(qt), [_]) = (&query_type, param_values.as_slice()) { + // A query type on its own reads `(query_type, blocked, + // timestamp)` as two runs already in timestamp order, one + // per verdict, and merges the newest of each. `blocked` + // sits in the middle of that index for the blocked filter's + // sake, and it is what stops a single seek from returning + // rows in page order: asked as one range, the planner reads + // every row of the type and sorts them. Each run only has + // to reach the end of the page, so neither reads more than + // `offset + limit` index entries. + // + // The runs carry only `id` and `timestamp`, which the index + // holds, and the table is read for the page's rows alone. + // Carrying every column instead looked up each row the runs + // passed over: 290 pages for page 20 of a busy type on a + // 1.48 M-row database, against 26. + sql = format!( + "SELECT {COLUMNS} FROM query_logs WHERE id IN ( \ + SELECT id FROM ( \ + SELECT id, timestamp FROM (SELECT id, timestamp FROM query_logs \ + INDEXED BY idx_query_logs_type_blocked_ts \ + WHERE query_type = ?1 AND blocked = 0 ORDER BY timestamp DESC LIMIT ?2) \ + UNION ALL \ + SELECT id, timestamp FROM (SELECT id, timestamp FROM query_logs \ + INDEXED BY idx_query_logs_type_blocked_ts \ + WHERE query_type = ?1 AND blocked = 1 ORDER BY timestamp DESC LIMIT ?2) \ + ORDER BY timestamp DESC LIMIT ?3 OFFSET ?4)) \ + ORDER BY timestamp DESC" + ); + // `SQLite` reads a negative `LIMIT` as none and a negative + // `OFFSET` as zero, and `/api/logs` passes both through, so + // the runs follow the same rules the outer page does. + let run = if limit < 0 { + -1 + } else { + offset.max(0).saturating_add(limit) + }; + param_values = vec![ + Box::new(qt.clone()), + Box::new(run), + Box::new(limit), + Box::new(offset), + ]; + } else { + sql.push_str(" ORDER BY timestamp DESC LIMIT ? OFFSET ?"); + param_values.push(Box::new(limit)); + param_values.push(Box::new(offset)); + } let params_refs: Vec<&dyn rusqlite::types::ToSql> = param_values.iter().map(std::convert::AsRef::as_ref).collect(); @@ -2555,13 +2635,12 @@ impl Database { Ok(timeline_from_buckets(&buckets)) } - /// Every `idx_query_logs_ts_metrics` reading the Statistics page renders, - /// in one scan of that index. - /// /// The outcome breakdown, the query-type breakdown and the latency - /// histogram are three foldings of the same window of the same index. - /// Asked one statement each, `SQLite` reads that index end to end three - /// times — and the read pool round-robins them onto four connections with + /// histogram, in one read of `query_stats_metrics_hour`. + /// + /// The three are foldings of the same window of the same grain. When they + /// were read off `idx_query_logs_ts_metrics`, asking one statement each + /// read that index end to end three times — and the read pool round-robins them onto four connections with /// 2 MiB of page cache each, so nothing is warm for the next one. The /// grain below is fine enough to derive all three and costs one scan: /// 2 157 pages instead of 6 471 on a 370 k-row database. @@ -3663,196 +3742,144 @@ mod tests { ); } + /// The `query_logs` indexes a database ends up with, whichever version it + /// opened at. Versions 10 to 15 each added or reshaped an index that + /// version 17 drops, so a legacy database carries indexes a fresh one never + /// keeps, and every one of them has to be gone once it opens — and the + /// statements that used to name them with `INDEXED BY` still have to + /// prepare, which is only proved by running them. #[tokio::test] - async fn fresh_schema_has_the_domain_client_index_and_not_the_client_one() { - let dir = tempfile::tempdir().unwrap(); - let path = dir.path().join("noadd.sqlite3"); - let db = Database::open(path.to_str().unwrap()).await.unwrap(); - - let indexes = query_log_index_names(&db).await; - assert!( - indexes - .iter() - .any(|n| n == "idx_query_logs_ts_domain_client"), - "(timestamp, domain, client_ip, doh_token) index should exist: {indexes:?}" - ); - assert!( - !indexes.iter().any(|n| n == "idx_query_logs_client_ts"), - "the client index it replaced should not: {indexes:?}" - ); - } - - #[tokio::test] - async fn fresh_schema_has_metrics_index() { - let dir = tempfile::tempdir().unwrap(); - let path = dir.path().join("noadd.sqlite3"); - let db = Database::open(path.to_str().unwrap()).await.unwrap(); - - let indexes = query_log_index_names(&db).await; - assert!( - indexes.iter().any(|n| n == "idx_query_logs_ts_metrics"), - "(timestamp, blocked, cached, response_ms, query_type) index should exist: {indexes:?}" - ); - } - - /// The metrics index only earns its disk if the planner treats it as - /// *covering* — a plain `SEARCH … USING INDEX` would still pay the row - /// lookup this index exists to avoid. - #[tokio::test] - async fn timeline_query_uses_the_metrics_index_as_covering() { - let dir = tempfile::tempdir().unwrap(); - let path = dir.path().join("plan.db"); - let db = Database::open(path.to_str().unwrap()).await.unwrap(); - - let logs: Vec = (0..500) - .map(|i| QueryLogEntry { - timestamp: 1_704_067_200_000 + i * 60_000, - domain: format!("d{}.example", i % 50), - query_type: if i % 3 == 0 { "AAAA" } else { "A" }.into(), - client_ip: format!("10.0.0.{}", i % 25), - blocked: i % 5 == 0, - cached: i % 4 == 0, - response_ms: i % 7, - upstream: None, - doh_token: None, - result: None, - authenticated_data: false, - }) - .collect(); - db.insert_query_logs(&logs).await.unwrap(); - - let plan = db - .conn - .call(|conn| { - conn.execute_batch("ANALYZE;")?; - let mut stmt = conn.prepare( - "EXPLAIN QUERY PLAN \ - SELECT (timestamp / 3600000) * 3600000 AS b, COUNT(*), \ - COALESCE(SUM(blocked), 0), COALESCE(SUM(cached), 0) \ - FROM query_logs WHERE timestamp >= ?1 GROUP BY b", - )?; - let rows = stmt - .query_map(params![0_i64], |row| row.get::<_, String>(3))? - .collect::, _>>()?; - Ok::<_, tokio_rusqlite::Error>(rows.join(" | ")) - }) - .await - .unwrap(); - - assert!( - plan.contains("COVERING INDEX idx_query_logs_ts_metrics"), - "timeline query should be covered by the metrics index, got: {plan}" + async fn every_database_opens_to_the_same_query_log_indexes() { + const LEGACY_TABLE: &str = "CREATE TABLE query_logs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + timestamp INTEGER NOT NULL, + domain TEXT NOT NULL, + query_type TEXT NOT NULL, + client_ip TEXT NOT NULL, + blocked INTEGER NOT NULL DEFAULT 0, + cached INTEGER NOT NULL DEFAULT 0, + response_ms INTEGER NOT NULL DEFAULT 0, + upstream TEXT, + doh_token TEXT, + result TEXT, + authenticated_data INTEGER NOT NULL DEFAULT 0 ); - } + CREATE INDEX idx_query_logs_timestamp ON query_logs(timestamp); + CREATE INDEX idx_query_logs_domain_ts ON query_logs(domain, timestamp);"; + let legacy = [ + ("v9", String::new(), 9), + ( + "v10", + "CREATE INDEX idx_query_logs_client_ts ON query_logs(client_ip, doh_token, timestamp);" + .to_string(), + 10, + ), + ( + "v11", + "CREATE INDEX idx_query_logs_client_ts ON query_logs(client_ip, doh_token, timestamp); + CREATE INDEX idx_query_logs_ts_metrics + ON query_logs(timestamp, blocked, cached, response_ms, query_type);" + .to_string(), + 11, + ), + ]; - #[tokio::test] - async fn migration_v11_adds_metrics_index() { let dir = tempfile::tempdir().unwrap(); - let path = dir.path().join("v10.db"); - let path_str = path.to_str().unwrap().to_string(); - - // A v10 database: has the client index, lacks the metrics one. - { - let conn = rusqlite::Connection::open(&path_str).unwrap(); - conn.execute_batch( - "CREATE TABLE query_logs ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - timestamp INTEGER NOT NULL, - domain TEXT NOT NULL, - query_type TEXT NOT NULL, - client_ip TEXT NOT NULL, - blocked INTEGER NOT NULL DEFAULT 0, - cached INTEGER NOT NULL DEFAULT 0, - response_ms INTEGER NOT NULL DEFAULT 0, - upstream TEXT, - doh_token TEXT, - result TEXT, - authenticated_data INTEGER NOT NULL DEFAULT 0 - ); - CREATE INDEX idx_query_logs_timestamp ON query_logs(timestamp); - CREATE INDEX idx_query_logs_domain_ts ON query_logs(domain, timestamp); - CREATE INDEX idx_query_logs_client_ts - ON query_logs(client_ip, doh_token, timestamp); - PRAGMA user_version = 10;", - ) + let mut databases = Vec::new(); + for (label, indexes, version) in legacy { + let path = dir.path().join(format!("{label}.db")); + let conn = rusqlite::Connection::open(&path).unwrap(); + conn.execute_batch(&format!( + "{LEGACY_TABLE}\n{indexes}\nPRAGMA user_version = {version};" + )) .unwrap(); + drop(conn); + databases.push((label, path)); } - - let db = Database::open(&path_str).await.unwrap(); - - let indexes = query_log_index_names(&db).await; - assert!( - indexes.iter().any(|n| n == "idx_query_logs_ts_metrics"), - "metrics index should be created by migration: {indexes:?}" - ); - } - - /// The metrics index has to end up carrying `has_result` whichever way the - /// database arrived at version 12 — created fresh, or migrated from a - /// version whose `query_logs` had no such column. - #[tokio::test] - async fn migration_v12_puts_the_outcome_flag_in_the_metrics_index() { - async fn metrics_index_columns(db: &Database) -> Vec { - db.reader() - .call(|conn| { - let mut stmt = conn.prepare( - "SELECT name FROM pragma_index_info('idx_query_logs_ts_metrics')", - )?; - let names = stmt - .query_map([], |row| row.get::<_, String>(0))? - .collect::, _>>()?; - Ok::<_, tokio_rusqlite::Error>(names) - }) - .await - .unwrap() - } - - let dir = tempfile::tempdir().unwrap(); - - // A v11 database: the metrics index exists, without the outcome flag. - let legacy = dir.path().join("v11.db"); - let legacy_str = legacy.to_str().unwrap().to_string(); + // Version 16 as it shipped: every index the steps up to it created. + let v16 = dir.path().join("v16.db"); { - let conn = rusqlite::Connection::open(&legacy_str).unwrap(); - conn.execute_batch( - "CREATE TABLE query_logs ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - timestamp INTEGER NOT NULL, - domain TEXT NOT NULL, - query_type TEXT NOT NULL, - client_ip TEXT NOT NULL, - blocked INTEGER NOT NULL DEFAULT 0, - cached INTEGER NOT NULL DEFAULT 0, - response_ms INTEGER NOT NULL DEFAULT 0, - upstream TEXT, - doh_token TEXT, - result TEXT, - authenticated_data INTEGER NOT NULL DEFAULT 0 - ); - CREATE INDEX idx_query_logs_timestamp ON query_logs(timestamp); - CREATE INDEX idx_query_logs_domain_ts ON query_logs(domain, timestamp); - CREATE INDEX idx_query_logs_client_ts - ON query_logs(client_ip, doh_token, timestamp); - CREATE INDEX idx_query_logs_ts_metrics - ON query_logs(timestamp, blocked, cached, response_ms, query_type); - PRAGMA user_version = 11;", + let db = Database::open(v16.to_str().unwrap()).await.unwrap(); + db.close().await; + } + rusqlite::Connection::open(&v16) + .unwrap() + .execute_batch( + "DROP INDEX idx_query_logs_token_ts; + DROP INDEX idx_query_logs_blocked_ts; + DROP INDEX idx_query_logs_type_blocked_ts; + CREATE INDEX idx_query_logs_ts_domain_client + ON query_logs(timestamp, domain, client_ip, doh_token); + CREATE INDEX idx_query_logs_ts_metrics + ON query_logs(timestamp, blocked, cached, response_ms, query_type, has_result); + CREATE INDEX idx_query_logs_ts_upstream + ON query_logs(timestamp, upstream, response_ms) WHERE upstream IS NOT NULL; + PRAGMA user_version = 16;", ) .unwrap(); - } - let migrated = Database::open(&legacy_str).await.unwrap(); + databases.push(("v16", v16)); + databases.push(("fresh", dir.path().join("fresh.db"))); + + let expected = [ + "idx_query_logs_blocked_ts", + "idx_query_logs_domain_ts", + "idx_query_logs_timestamp", + "idx_query_logs_token_ts", + "idx_query_logs_type_blocked_ts", + ]; + for (label, path) in databases { + let db = Database::open(path.to_str().unwrap()).await.unwrap(); + let mut entries: Vec = (0..6) + .map(|i| sample_entry(1_000_000 + i, "example.com")) + .collect(); + entries[1].blocked = true; + entries[2].upstream = Some("tls://1.1.1.1:853".to_string()); + entries[3].doh_token = Some("phone".to_string()); + entries[4].result = Some("1.2.3.4".to_string()); + db.insert_query_logs(&entries).await.unwrap(); - let fresh_path = dir.path().join("fresh.db"); - let fresh = Database::open(fresh_path.to_str().unwrap()).await.unwrap(); + let mut indexes = query_log_index_names(&db).await; + indexes.retain(|n| !n.starts_with("sqlite_autoindex")); + indexes.sort(); + assert_eq!(indexes, expected, "{label} database's indexes"); - for (label, db) in [("migrated", &migrated), ("fresh", &fresh)] { - let columns = metrics_index_columns(db).await; - assert!( - columns.iter().any(|c| c == "has_result"), - "{label} database's metrics index lacks has_result: {columns:?}" + assert_eq!(db.summary_multi_since(0, 0, 0).await.unwrap()[2].total, 6); + assert_eq!( + db.traffic_lists_since(0, 10).await.unwrap().clients.len(), + 2 + ); + assert_eq!(db.top_upstreams_since(0, 10).await.unwrap().len(), 1); + assert_eq!( + db.stats_scan_since(0, 0).await.unwrap().series.total, + vec![6] + ); + assert_eq!(db.timeline_multi_since(0, 3_600, 0).await.unwrap().len(), 1); + assert_eq!(db.hourly_heatmap_since(0, 0).await.unwrap().len(), 1); + assert_eq!( + db.outcome_breakdown_since(0).await.unwrap().len(), + 3, + "{label}: blocked, resolved and empty" + ); + assert_eq!( + db.query_logs(10, 0, None, None, Some("phone"), None) + .await + .unwrap() + .len(), + 1 + ); + assert_eq!( + db.query_logs(10, 0, None, None, None, Some("A")) + .await + .unwrap() + .len(), + 6 + ); + assert_eq!( + db.count_logs(None, Some(true), None, Some("A")) + .await + .unwrap(), + 1 ); - // Reachable through the query that depends on it — the index could - // carry the column and still be the wrong one for `INDEXED BY`. - assert!(db.outcome_breakdown_since(0).await.is_ok()); } } @@ -3911,106 +3938,6 @@ mod tests { assert_eq!(migrated.total_log_count().await.unwrap(), 3); } - /// A database from before version 14 has to come out of `open` holding the - /// upstream index, because `top_upstreams_since` names it with `INDEXED BY` - /// and a statement naming a missing index does not prepare at all — the - /// dashboard would lose its upstream list rather than merely run slower. - #[tokio::test] - async fn migration_v14_adds_the_upstream_index() { - let dir = tempfile::tempdir().unwrap(); - let path = dir.path().join("v13.db"); - let path_str = path.to_str().unwrap().to_string(); - - let entries: Vec = (0..4) - .map(|i| QueryLogEntry { - timestamp: 1_000_000 + i, - domain: "example.com".to_string(), - query_type: "A".to_string(), - client_ip: "10.0.0.1".to_string(), - blocked: false, - cached: false, - upstream: (i != 0).then(|| "tls://1.1.1.1:853".to_string()), - doh_token: None, - result: None, - response_ms: 10 * i, - authenticated_data: false, - }) - .collect(); - { - let db = Database::open(&path_str).await.unwrap(); - db.insert_query_logs(&entries).await.unwrap(); - db.close().await; - } - { - let conn = rusqlite::Connection::open(&path_str).unwrap(); - conn.execute_batch( - "DROP INDEX idx_query_logs_ts_upstream; - PRAGMA user_version = 13;", - ) - .unwrap(); - } - - let migrated = Database::open(&path_str).await.unwrap(); - let indexes = query_log_index_names(&migrated).await; - assert!( - indexes.iter().any(|n| n == "idx_query_logs_ts_upstream"), - "upstream index should exist after migrating: {indexes:?}" - ); - let top = migrated.top_upstreams_since(0, 10).await.unwrap(); - assert_eq!(top.len(), 1); - assert_eq!(top[0].count, 3, "the unforwarded row is not an upstream's"); - assert!((top[0].avg_ms - 20.0).abs() < 1e-9); - } - - /// Version 10 added the client index and version 15 replaced it, so a - /// database from before either has to come out holding the replacement and - /// nothing of the index in between. - #[tokio::test] - async fn a_v9_database_migrates_to_the_domain_client_index() { - let dir = tempfile::tempdir().unwrap(); - let path = dir.path().join("v9.db"); - let path_str = path.to_str().unwrap().to_string(); - - // A v9 database: everything current except the client index. - { - let conn = rusqlite::Connection::open(&path_str).unwrap(); - conn.execute_batch( - "CREATE TABLE query_logs ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - timestamp INTEGER NOT NULL, - domain TEXT NOT NULL, - query_type TEXT NOT NULL, - client_ip TEXT NOT NULL, - blocked INTEGER NOT NULL DEFAULT 0, - cached INTEGER NOT NULL DEFAULT 0, - response_ms INTEGER NOT NULL DEFAULT 0, - upstream TEXT, - doh_token TEXT, - result TEXT, - authenticated_data INTEGER NOT NULL DEFAULT 0 - ); - CREATE INDEX idx_query_logs_timestamp ON query_logs(timestamp); - CREATE INDEX idx_query_logs_domain_ts ON query_logs(domain, timestamp); - PRAGMA user_version = 9;", - ) - .unwrap(); - } - - let db = Database::open(&path_str).await.unwrap(); - - let indexes = query_log_index_names(&db).await; - assert!( - indexes - .iter() - .any(|n| n == "idx_query_logs_ts_domain_client"), - "domain/client index should be created by migration: {indexes:?}" - ); - assert!( - !indexes.iter().any(|n| n == "idx_query_logs_client_ts"), - "the client index should be gone after migration: {indexes:?}" - ); - } - #[tokio::test] async fn migration_v6_drops_credential_and_adds_tables() { let dir = tempfile::tempdir().unwrap(); diff --git a/tests/db_test.rs b/tests/db_test.rs index 0040c6a..e54466e 100644 --- a/tests/db_test.rs +++ b/tests/db_test.rs @@ -1041,3 +1041,72 @@ async fn the_log_count_follows_every_write_that_changes_it() { db.delete_all_logs().await.unwrap(); assert_holds(&db, 0).await; } + +/// A query type on its own is answered as two runs, one per verdict, merged +/// back into one page. The page has to be the one the plain statement returns +/// for every page size and offset `/api/logs` accepts, including the negative +/// ones `SQLite` reads as "no limit" and "from the start". +#[tokio::test] +async fn a_query_type_filter_pages_exactly_like_the_table() { + let dir = tempdir().unwrap(); + let path = dir.keep().join("paging.db"); + let path_str = path.to_str().unwrap().to_string(); + let db = Database::open(&path_str).await.unwrap(); + + // Verdicts come in uneven runs, so a page often draws from only one of the + // two runs and the merge point moves from page to page. + let entries: Vec = (0..300_i64) + .map(|i| QueryLogEntry { + timestamp: 1_000_000 + i * 1000, + domain: format!("d{i}.example"), + query_type: if i % 3 == 0 { "AAAA" } else { "A" }.to_string(), + client_ip: "10.0.0.1".to_string(), + blocked: (i / 7) % 3 == 0 || i % 11 == 0, + cached: false, + upstream: None, + doh_token: None, + result: None, + response_ms: 1, + authenticated_data: false, + }) + .collect(); + db.insert_query_logs(&entries).await.unwrap(); + let raw = rusqlite::Connection::open(&path_str).unwrap(); + + for query_type in ["A", "AAAA", "TXT"] { + for (limit, offset) in [ + (10, 0), + (10, 40), + (50, 90), + (7, 193), + (50, 180), + (10, 1_000), + (0, 0), + (-1, 0), + (-1, 30), + (10, -5), + ] { + let expected: Vec = raw + .prepare( + "SELECT domain FROM query_logs WHERE query_type = ?1 \ + ORDER BY timestamp DESC LIMIT ?2 OFFSET ?3", + ) + .unwrap() + .query_map(rusqlite::params![query_type, limit, offset], |r| r.get(0)) + .unwrap() + .collect::>() + .unwrap(); + let got: Vec = db + .query_logs(limit, offset, None, None, None, Some(query_type)) + .await + .unwrap() + .into_iter() + .map(|e| e.domain) + .collect(); + assert_eq!( + got, expected, + "type {query_type}, limit {limit}, offset {offset}" + ); + } + } +} diff --git a/tests/stats_page_miss_test.rs b/tests/stats_page_miss_test.rs index 19b8aeb..efdb454 100644 --- a/tests/stats_page_miss_test.rs +++ b/tests/stats_page_miss_test.rs @@ -333,3 +333,87 @@ async fn the_unfiltered_query_log_count_is_read_rather_than_counted() { ); } } + +/// Twenty thousand queries where one in four hundred came from a quiet `DoH` +/// token and one in four hundred asked for a quiet record type — the filters +/// an operator reaches for to find the few rows a busy log buries. +async fn quiet_filters_db() -> Database { + let dir = tempdir().unwrap(); + let path = dir.keep().join("filters.db"); + let db = Database::open(path.to_str().unwrap()).await.unwrap(); + + let entries: Vec = (0..ROWS) + .map(|i| QueryLogEntry { + timestamp: i * 1000, + domain: format!("host{}.example.com", i % 500), + query_type: if i % 400 == 7 { "TXT" } else { "A" }.to_string(), + client_ip: format!("10.0.0.{}", i % 20), + blocked: i % 7 == 0, + cached: i % 5 == 0, + upstream: None, + doh_token: (i % 400 == 3).then(|| "quiet".to_string()), + result: Some("x".repeat(RESULT_PADDING)), + response_ms: i % 50, + authenticated_data: false, + }) + .collect(); + for chunk in entries.chunks(2_000) { + db.insert_query_logs(chunk).await.unwrap(); + } + db +} + +/// Filtering the query log by a token, a record type or a verdict seeks an +/// index. With nothing to seek, the newest page of a quiet token walks the +/// table back until it has fifty rows — here, all of it — counting the matches +/// walks all of it every time, and a deep page of blocked queries looks up +/// every row it passes over to learn its verdict. +#[tokio::test] +async fn the_query_log_filters_seek_their_indexes() { + let db = quiet_filters_db().await; + let db_pages = db.db_storage_stats().await.unwrap().main_bytes / 4096; + let readings = [ + ( + "token page", + page_misses(&db, || { + db.query_logs(50, 0, None, None, Some("quiet"), None) + }) + .await, + ), + ( + "token count", + page_misses(&db, || db.count_logs(None, None, Some("quiet"), None)).await, + ), + ( + "type page", + page_misses(&db, || db.query_logs(50, 0, None, None, None, Some("TXT"))).await, + ), + ( + "type count", + page_misses(&db, || db.count_logs(None, None, None, Some("TXT"))).await, + ), + ( + "blocked page 20", + page_misses(&db, || db.query_logs(50, 950, None, Some(true), None, None)).await, + ), + ( + "type + blocked page", + page_misses(&db, || { + db.query_logs(50, 0, None, Some(false), None, Some("TXT")) + }) + .await, + ), + ]; + + for (label, read) in readings { + assert!( + read > 0, + "{label} read no pages at all — the measurement is not working" + ); + assert!( + read * 10 < db_pages, + "{label} read {read} of the database's {db_pages} pages — \ + is the filter back to walking the table?" + ); + } +}