Files
nxdns/src/storage/repositories/queries_repo.zig
T
mokhtar 0f01c2fbd7
Gates / frontend (push) Successful in 2m8s
Gates / test (push) Successful in 2m46s
Gates / test-aarch64 (push) Successful in 8m38s
Gates / package (push) Successful in 4m39s
Gates / container (push) Successful in 15s
CI / gates (push) Successful in 31m58s
storage: version querylog.db and migrate it in place, never reset a healthy file
querylog.db carries a schema version; migrations run at startup as one transaction after a vacuumed 0600 backup, and every failure refuses startup (exit 2, no systemd restart loop) instead of starting empty. corruption is the only automatic recreate left. the cut gate now requires a fixture-proven migration or an explicit versioned break with restore instructions, and locks shipped migration files and fixtures byte-for-byte.
2026-08-28 17:56:19 +02:00

3159 lines
131 KiB
Zig

//! `query_log` and its `domains` dimension table in `querylog.db`.
//!
//! Two shapes live here. The free functions follow the milestone-4 repository
//! idiom — prepare, use, finalize — because retention runs them a handful of
//! times per day, and the API read layer at the bottom of the file runs once
//! per HTTP request. The flush loop is the one hot path in the program, so it
//! gets `BatchWriter`, which owns its three statements for its whole life
//! (`db.zig:360` names this file as the reason `db.zig` carries no statement
//! cache).
//!
//! Every string in a `Row` is borrowed for the duration of the call only:
//! `Stmt.bindText` binds with `SQLITE_TRANSIENT`, so SQLite copies before
//! `writeBatch` returns.
//!
//! The rows are expendable log data. Nothing here retries, and the caller
//! decides what a failed batch means.
const std = @import("std");
const Allocator = std.mem.Allocator;
const db = @import("../db.zig");
const provenance = @import("../provenance.zig");
/// One `query_log` row. The logger applies the privacy transforms of PLAN
/// §11.4 before it builds this, so every domain-bearing field is already
/// whatever the operator agreed to store.
///
/// A `null` text field is a fact the query did not have — no upstream was
/// attempted, no rule matched, no CNAME was uncloaked — and reaches the column
/// as NULL. The three closed enums have no such state: every logged query has a
/// policy verdict, a reason for it and a route, even when the verdict is
/// "not evaluated".
pub const Row = struct {
timestamp: i64,
domain: []const u8,
client_ip: []const u8,
qtype: ?u16,
qclass: u16,
/// Twelve bits: the EDNS extended RCODE the client saw. The column's
/// `CHECK` bounds it to the same range, so a value this type cannot hold
/// is one the schema would have refused anyway.
rcode: u12,
blocked: bool,
response_time_us: ?i64,
cache_hit: ?bool,
upstream: ?[]const u8,
group_id: ?i64,
group_name: ?[]const u8,
policy_action: provenance.PolicyAction,
policy_reason: provenance.PolicyReason,
matched: ?[]const u8,
source_id: ?i64,
source_name: ?[]const u8,
cname_target: ?[]const u8,
safe_search_target: ?[]const u8,
route_kind: provenance.RouteKind,
forward_zone: ?[]const u8,
};
/// The projection grain of the four `bucket_*` tables, in seconds.
///
/// 1800 divides every serving width the Overview offers at or above half an
/// hour (1800, 3600, 21600), which is what lets one grain answer the 24h, 7d
/// and 30d windows exactly. The 1h window is 60-second buckets and is served
/// from the raw rows instead.
pub const grain: i64 = 1800;
/// Floors a timestamp onto the projection grid.
///
/// `@divFloor`, never `@divTrunc` or SQLite's `/`: both truncate toward zero,
/// which for a negative timestamp names the bucket *after* the one the row
/// belongs to. Every producer and every consumer of a `bucket` value in this
/// file goes through this function or an equivalent floor expression.
pub fn bucketOf(timestamp: i64) i64 {
return @divFloor(timestamp, grain) * grain;
}
/// `bucket_types.qtype` is NOT NULL, so the rows that carry no query type need
/// a value of their own. No qtype is a `u16`, so -1 cannot collide with one.
pub const null_qtype: i64 = -1;
/// The answering resolver's identity, as the routes breakdown defines it: the
/// upstream url for `upstream` rows, the zone for `forward_zone` rows, null
/// everywhere else. The Zig twin of `overview_raw_sql`'s CASE, and the only
/// place the writer decides what a route's source is.
fn routeSourceOf(row: Row) ?[]const u8 {
return switch (row.route_kind) {
.upstream => row.upstream,
.forward_zone => row.forward_zone,
else => null,
};
}
const insert_domain_sql = "INSERT OR IGNORE INTO domains (domain) VALUES (?1)";
const select_domain_sql = "SELECT id FROM domains WHERE domain = ?1";
const insert_row_sql =
\\INSERT INTO query_log
\\ (timestamp, domain_id, client_ip, qtype, blocked,
\\ response_time_us, cache_hit, upstream, qclass, rcode,
\\ group_id, group_name, policy_action, policy_reason, matched,
\\ source_id, source_name, cname_target, safe_search_target,
\\ route_kind, forward_zone)
\\VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10,
\\ ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?19, ?20, ?21)
;
/// One `bucket_totals` row's worth of change from a single batch.
const TotalsDelta = struct {
queries: i64 = 0,
blocked: i64 = 0,
cached: i64 = 0,
rt_sum: i64 = 0,
rt_count: i64 = 0,
};
/// The delta-map keys borrow the batch's strings. They live only from the
/// aggregation pass to the end of the `writeBatch` that produced them, which is
/// inside the borrow `Row` already documents.
const ClientKey = struct { bucket: i64, client_ip: []const u8 };
const TypeKey = struct { bucket: i64, qtype: i64 };
const RouteKey = struct { bucket: i64, route: provenance.RouteKind, source: ?[]const u8 };
const ClientKeyContext = struct {
pub fn hash(_: ClientKeyContext, key: ClientKey) u64 {
var hasher: std.hash.Wyhash = .init(@bitCast(key.bucket));
hasher.update(key.client_ip);
return hasher.final();
}
pub fn eql(_: ClientKeyContext, a: ClientKey, b: ClientKey) bool {
return a.bucket == b.bucket and std.mem.eql(u8, a.client_ip, b.client_ip);
}
};
const RouteKeyContext = struct {
pub fn hash(_: RouteKeyContext, key: RouteKey) u64 {
var hasher: std.hash.Wyhash = .init(@bitCast(key.bucket));
hasher.update(@tagName(key.route));
hasher.update(&[_]u8{@intFromBool(key.source != null)});
hasher.update(key.source orelse "");
return hasher.final();
}
pub fn eql(_: RouteKeyContext, a: RouteKey, b: RouteKey) bool {
return a.bucket == b.bucket and a.route == b.route and sameSource(a.source, b.source);
}
};
const load_percentage = std.hash_map.default_max_load_percentage;
const upsert_totals_sql =
\\INSERT INTO bucket_totals (bucket, queries, blocked, cached, rt_sum, rt_count)
\\VALUES (?1, ?2, ?3, ?4, ?5, ?6)
\\ON CONFLICT(bucket) DO UPDATE SET
\\ queries = queries + excluded.queries,
\\ blocked = blocked + excluded.blocked,
\\ cached = cached + excluded.cached,
\\ rt_sum = rt_sum + excluded.rt_sum,
\\ rt_count = rt_count + excluded.rt_count
;
const upsert_clients_sql =
\\INSERT INTO bucket_clients (bucket, client_ip, queries)
\\VALUES (?1, ?2, ?3)
\\ON CONFLICT(bucket, client_ip) DO UPDATE SET queries = queries + excluded.queries
;
const upsert_types_sql =
\\INSERT INTO bucket_types (bucket, qtype, count)
\\VALUES (?1, ?2, ?3)
\\ON CONFLICT(bucket, qtype) DO UPDATE SET count = count + excluded.count
;
const upsert_routes_sql =
\\INSERT INTO bucket_routes (bucket, route_kind, source_present, source_text, count)
\\VALUES (?1, ?2, ?3, ?4, ?5)
\\ON CONFLICT(bucket, route_kind, source_present, source_text)
\\DO UPDATE SET count = count + excluded.count
;
/// Owns the prepared statements of the flush loop. Init once, reuse per batch.
///
/// `database` must outlive the writer and must not move: every `Stmt` holds a
/// `*Db`. Neither `Db` nor `Stmt` is thread-safe, so one writer belongs to one
/// task.
///
/// The writer also maintains the four `bucket_*` projections, in the same
/// transaction as the rows they summarise. `gpa` is owned for the writer's life
/// and is used for nothing but the per-batch delta maps, whose capacity is
/// retained across batches and whose contents are dropped at the end of every
/// `writeBatch`.
pub const BatchWriter = struct {
gpa: Allocator,
database: *db.Db,
insert_domain: db.Stmt,
select_domain: db.Stmt,
insert_row: db.Stmt,
upsert_totals: db.Stmt,
upsert_clients: db.Stmt,
upsert_types: db.Stmt,
upsert_routes: db.Stmt,
totals_deltas: std.AutoHashMapUnmanaged(i64, TotalsDelta) = .empty,
client_deltas: std.HashMapUnmanaged(ClientKey, i64, ClientKeyContext, load_percentage) = .empty,
type_deltas: std.AutoHashMapUnmanaged(TypeKey, i64) = .empty,
route_deltas: std.HashMapUnmanaged(RouteKey, i64, RouteKeyContext, load_percentage) = .empty,
pub fn init(gpa: Allocator, database: *db.Db) db.Error!BatchWriter {
var insert_domain = try database.prepare(insert_domain_sql);
errdefer insert_domain.deinit();
var select_domain = try database.prepare(select_domain_sql);
errdefer select_domain.deinit();
var insert_row = try database.prepare(insert_row_sql);
errdefer insert_row.deinit();
var upsert_totals = try database.prepare(upsert_totals_sql);
errdefer upsert_totals.deinit();
var upsert_clients = try database.prepare(upsert_clients_sql);
errdefer upsert_clients.deinit();
var upsert_types = try database.prepare(upsert_types_sql);
errdefer upsert_types.deinit();
const upsert_routes = try database.prepare(upsert_routes_sql);
return .{
.gpa = gpa,
.database = database,
.insert_domain = insert_domain,
.select_domain = select_domain,
.insert_row = insert_row,
.upsert_totals = upsert_totals,
.upsert_clients = upsert_clients,
.upsert_types = upsert_types,
.upsert_routes = upsert_routes,
};
}
pub fn deinit(self: *BatchWriter) void {
self.route_deltas.deinit(self.gpa);
self.type_deltas.deinit(self.gpa);
self.client_deltas.deinit(self.gpa);
self.totals_deltas.deinit(self.gpa);
self.upsert_routes.deinit();
self.upsert_types.deinit();
self.upsert_clients.deinit();
self.upsert_totals.deinit();
self.insert_row.deinit();
self.select_domain.deinit();
self.insert_domain.deinit();
}
/// One transaction for the whole batch: the raw rows and every projection
/// they touch commit together or not at all. Domains are interned through
/// `INSERT OR IGNORE` followed by `SELECT id`.
///
/// On any failure the transaction rolls back, so a batch is all or
/// nothing, and the writer stays usable for the next batch.
pub fn writeBatch(self: *BatchWriter, rows: []const Row) db.Error!void {
if (rows.len == 0) return;
// Declared before the aggregation so it runs after every `errdefer`
// below, and placed before `BEGIN` so an `error.OutOfMemory` fails the
// batch with no transaction outstanding.
defer self.clearDeltas();
try self.aggregate(rows);
var tx = try db.Tx.begin(self.database);
// `errdefer`s run in reverse: the statements are released before the
// ROLLBACK, so no read cursor is still open when it runs.
errdefer tx.rollback();
errdefer self.resetAll();
for (rows) |row| {
const domain_id = try self.internDomain(row.domain);
try self.write(row, domain_id);
}
try self.applyProjections();
try tx.commit();
}
/// Folds the batch into per-key deltas, so the transaction below runs one
/// UPSERT per touched key rather than per row. Any slice length: the
/// logger's 100-row batching is its own policy, not this API's.
fn aggregate(self: *BatchWriter, rows: []const Row) Allocator.Error!void {
for (rows) |row| {
const bucket = bucketOf(row.timestamp);
const totals = try self.totals_deltas.getOrPut(self.gpa, bucket);
if (!totals.found_existing) totals.value_ptr.* = .{};
totals.value_ptr.queries += 1;
if (row.blocked) totals.value_ptr.blocked += 1;
if (row.cache_hit orelse false) totals.value_ptr.cached += 1;
if (row.response_time_us) |us| {
totals.value_ptr.rt_sum += us;
totals.value_ptr.rt_count += 1;
}
try bump(&self.client_deltas, self.gpa, ClientKey{
.bucket = bucket,
.client_ip = row.client_ip,
});
try bump(&self.type_deltas, self.gpa, TypeKey{
.bucket = bucket,
.qtype = if (row.qtype) |v| @as(i64, v) else null_qtype,
});
try bump(&self.route_deltas, self.gpa, RouteKey{
.bucket = bucket,
.route = row.route_kind,
.source = routeSourceOf(row),
});
}
}
/// One more row on `key`. `map` is any of the three count maps — they
/// differ only in their key type and its hashing.
fn bump(map: anytype, gpa: Allocator, key: anytype) Allocator.Error!void {
const entry = try map.getOrPut(gpa, key);
if (!entry.found_existing) entry.value_ptr.* = 0;
entry.value_ptr.* += 1;
}
fn applyProjections(self: *BatchWriter) db.Error!void {
var totals = self.totals_deltas.iterator();
while (totals.next()) |entry| {
const stmt = &self.upsert_totals;
try stmt.reset();
try stmt.bindInt(1, entry.key_ptr.*);
try stmt.bindInt(2, entry.value_ptr.queries);
try stmt.bindInt(3, entry.value_ptr.blocked);
try stmt.bindInt(4, entry.value_ptr.cached);
try stmt.bindInt(5, entry.value_ptr.rt_sum);
try stmt.bindInt(6, entry.value_ptr.rt_count);
try stmt.exec();
}
var clients = self.client_deltas.iterator();
while (clients.next()) |entry| {
const stmt = &self.upsert_clients;
try stmt.reset();
try stmt.bindInt(1, entry.key_ptr.bucket);
try stmt.bindText(2, entry.key_ptr.client_ip);
try stmt.bindInt(3, entry.value_ptr.*);
try stmt.exec();
}
var types = self.type_deltas.iterator();
while (types.next()) |entry| {
const stmt = &self.upsert_types;
try stmt.reset();
try stmt.bindInt(1, entry.key_ptr.bucket);
try stmt.bindInt(2, entry.key_ptr.qtype);
try stmt.bindInt(3, entry.value_ptr.*);
try stmt.exec();
}
var routes = self.route_deltas.iterator();
while (routes.next()) |entry| {
const stmt = &self.upsert_routes;
try stmt.reset();
try stmt.bindInt(1, entry.key_ptr.bucket);
try stmt.bindText(2, @tagName(entry.key_ptr.route));
try stmt.bindInt(3, @intFromBool(entry.key_ptr.source != null));
try stmt.bindText(4, entry.key_ptr.source orelse "");
try stmt.bindInt(5, entry.value_ptr.*);
try stmt.exec();
}
}
/// Capacity is retained; only the entries go. The keys borrow the batch's
/// strings, so nothing may outlive the `writeBatch` that filled them.
fn clearDeltas(self: *BatchWriter) void {
self.totals_deltas.clearRetainingCapacity();
self.client_deltas.clearRetainingCapacity();
self.type_deltas.clearRetainingCapacity();
self.route_deltas.clearRetainingCapacity();
}
fn internDomain(self: *BatchWriter, domain: []const u8) db.Error!i64 {
try self.insert_domain.reset();
try self.insert_domain.bindText(1, domain);
try self.insert_domain.exec();
try self.select_domain.reset();
try self.select_domain.bindText(1, domain);
// The insert above either created the row or found it already there,
// so a miss means the table changed under this connection.
if (!try self.select_domain.step()) return error.NotFound;
const id = self.select_domain.columnInt(0);
// A statement stopped on a row keeps its cursor open until it is
// reset; the transaction must not carry that to the next row.
try self.select_domain.reset();
return id;
}
fn write(self: *BatchWriter, row: Row, domain_id: i64) db.Error!void {
var stmt = &self.insert_row;
try stmt.reset();
try stmt.bindInt(1, row.timestamp);
try stmt.bindInt(2, domain_id);
try stmt.bindText(3, row.client_ip);
try bindIntOrNull(stmt, 4, if (row.qtype) |v| @as(i64, v) else null);
try stmt.bindBool(5, row.blocked);
try bindIntOrNull(stmt, 6, row.response_time_us);
try bindIntOrNull(stmt, 7, if (row.cache_hit) |v| @as(i64, @intFromBool(v)) else null);
try stmt.bindTextOrNull(8, row.upstream);
try stmt.bindInt(9, row.qclass);
try stmt.bindInt(10, row.rcode);
try bindIntOrNull(stmt, 11, row.group_id);
try stmt.bindTextOrNull(12, row.group_name);
try stmt.bindText(13, @tagName(row.policy_action));
try stmt.bindText(14, @tagName(row.policy_reason));
try stmt.bindTextOrNull(15, row.matched);
try bindIntOrNull(stmt, 16, row.source_id);
try stmt.bindTextOrNull(17, row.source_name);
try stmt.bindTextOrNull(18, row.cname_target);
try stmt.bindTextOrNull(19, row.safe_search_target);
try stmt.bindText(20, @tagName(row.route_kind));
try stmt.bindTextOrNull(21, row.forward_zone);
try stmt.exec();
}
/// Best effort: this runs on the failure path, where the error that
/// matters is the one already on its way to the caller.
fn resetAll(self: *BatchWriter) void {
self.upsert_routes.reset() catch {};
self.upsert_types.reset() catch {};
self.upsert_clients.reset() catch {};
self.upsert_totals.reset() catch {};
self.insert_row.reset() catch {};
self.select_domain.reset() catch {};
self.insert_domain.reset() catch {};
}
};
fn bindIntOrNull(stmt: *db.Stmt, idx: c_int, value: ?i64) db.Error!void {
if (value) |v| return stmt.bindInt(idx, v);
return stmt.bindNull(idx);
}
/// What one prune did, and where coverage now begins.
pub const PruneResult = struct {
deleted: i64,
/// The watermark after the prune, which is what a later `availableSince`
/// will return. Handed back so the caller need not re-read it.
available_since: i64,
};
/// Deletes every `query_log` row strictly older than `cutoff_ts` and advances
/// the coverage watermark to the same cutoff, in one transaction.
///
/// **The two are one operation, not two.** The watermark is the promise that
/// every query since it is still in the file; a delete that commits without the
/// advance breaks that promise, and an advance that commits without the delete
/// hides rows the file still holds. Either failure rolls both back, and the
/// caller retries the whole thing on its next pass.
///
/// The watermark never moves backward: `max` is what makes a prune with a
/// cutoff older than the file's own creation a no-op on it rather than a
/// regression. Orphaned `domains` rows stay — it is a dimension table,
/// re-interning a name costs one indexed insert, and §11.3 asks for no
/// collection.
///
/// The four projections are pruned in the same transaction. Buckets entirely
/// before the cutoff's own bucket are deleted outright; a cutoff that does not
/// land on the grid leaves one straddling bucket, which is recomputed from the
/// rows that survived rather than reduced by an estimate. Any failure in any of
/// that rolls back the raw delete and the watermark advance with it.
pub fn pruneOlderThan(database: *db.Db, cutoff_ts: i64) db.Error!PruneResult {
var tx = try db.Tx.begin(database);
errdefer tx.rollback();
var deleting = try database.prepare("DELETE FROM query_log WHERE timestamp < ?1");
defer deleting.deinit();
try deleting.bindInt(1, cutoff_ts);
try deleting.exec();
const deleted = database.changes();
const grid = bucketOf(cutoff_ts);
for (delete_before_bucket_sql) |sql| {
var stmt = try database.prepare(sql);
defer stmt.deinit();
try stmt.bindInt(1, grid);
try stmt.exec();
}
// On the grid there is nothing partial to fix: the delete above took whole
// buckets and the loop took their projection rows.
if (cutoff_ts != grid) try recomputeBucket(database, grid);
var advancing = try database.prepare(
"UPDATE querylog_meta SET available_since = max(available_since, ?1) WHERE id = 1",
);
defer advancing.deinit();
try advancing.bindInt(1, cutoff_ts);
try advancing.exec();
const watermark = try database.queryInt("SELECT available_since FROM querylog_meta WHERE id = 1");
try tx.commit();
return .{ .deleted = deleted, .available_since = watermark };
}
const delete_before_bucket_sql = [_][]const u8{
"DELETE FROM bucket_totals WHERE bucket < ?1",
"DELETE FROM bucket_clients WHERE bucket < ?1",
"DELETE FROM bucket_types WHERE bucket < ?1",
"DELETE FROM bucket_routes WHERE bucket < ?1",
};
const delete_one_bucket_sql = [_][]const u8{
"DELETE FROM bucket_totals WHERE bucket = ?1",
"DELETE FROM bucket_clients WHERE bucket = ?1",
"DELETE FROM bucket_types WHERE bucket = ?1",
"DELETE FROM bucket_routes WHERE bucket = ?1",
};
/// `?1` is the bucket start and `?2` its exclusive end. Each statement rebuilds
/// one projection table's rows for that bucket from the raw rows still in it.
///
/// `HAVING count(*) > 0` on the totals statement is what keeps an emptied
/// bucket from being written back as a row of zeros: a bare aggregate always
/// produces one row, and a from-scratch recomputation produces none. The other
/// three group by a key, so an empty range already yields nothing.
const recompute_bucket_sql = [_][]const u8{
\\INSERT INTO bucket_totals (bucket, queries, blocked, cached, rt_sum, rt_count)
\\SELECT ?1, count(*), coalesce(sum(blocked <> 0), 0), coalesce(sum(cache_hit = 1), 0),
\\ coalesce(sum(response_time_us), 0), count(response_time_us)
\\ FROM query_log
\\ WHERE timestamp >= ?1 AND timestamp < ?2
\\HAVING count(*) > 0
,
\\INSERT INTO bucket_clients (bucket, client_ip, queries)
\\SELECT ?1, client_ip, count(*)
\\ FROM query_log
\\ WHERE timestamp >= ?1 AND timestamp < ?2
\\ GROUP BY client_ip
,
\\INSERT INTO bucket_types (bucket, qtype, count)
\\SELECT ?1, coalesce(qtype, -1), count(*)
\\ FROM query_log
\\ WHERE timestamp >= ?1 AND timestamp < ?2
\\ GROUP BY coalesce(qtype, -1)
,
\\INSERT INTO bucket_routes (bucket, route_kind, source_present, source_text, count)
\\SELECT ?1, route_kind, source IS NOT NULL, coalesce(source, ''), count(*)
\\ FROM (SELECT route_kind,
\\ CASE route_kind
\\ WHEN 'upstream' THEN upstream
\\ WHEN 'forward_zone' THEN forward_zone
\\ END AS source
\\ FROM query_log
\\ WHERE timestamp >= ?1 AND timestamp < ?2)
\\ GROUP BY route_kind, source
,
};
/// Replaces one bucket's projection rows with a recomputation from the raw rows
/// that are still in it. The caller owns the transaction.
fn recomputeBucket(database: *db.Db, bucket: i64) db.Error!void {
for (delete_one_bucket_sql) |sql| {
var stmt = try database.prepare(sql);
defer stmt.deinit();
try stmt.bindInt(1, bucket);
try stmt.exec();
}
for (recompute_bucket_sql) |sql| {
var stmt = try database.prepare(sql);
defer stmt.deinit();
try stmt.bindInt(1, bucket);
try stmt.bindInt(2, bucket + grain);
try stmt.exec();
}
}
/// The oldest timestamp this file can still answer for. A query window that
/// starts before it is incomplete, and the API says so rather than charting the
/// gap as zero.
pub fn availableSince(database: *db.Db) db.Error!i64 {
return database.queryInt("SELECT available_since FROM querylog_meta WHERE id = 1");
}
/// `PRAGMA wal_checkpoint(TRUNCATE)`: moves the WAL into the database and
/// truncates it to zero bytes, which is what keeps a day of log writes from
/// growing the WAL past the free space the disk monitor watches.
///
/// SQLite reports a checkpoint blocked by a concurrent reader in the row it
/// returns, not as an error code, so a blocked checkpoint is not an error
/// here. Retention checkpoints after every prune, so the next pass retries.
/// On a database that is not in WAL mode the pragma is a no-op.
pub fn checkpointTruncate(database: *db.Db) db.Error!void {
return database.exec("PRAGMA wal_checkpoint(TRUNCATE);");
}
/// Rewrites the whole file. Retention runs this rarely by design — on an SD
/// card a full rewrite is the most expensive thing this program does.
pub fn vacuum(database: *db.Db) db.Error!void {
return database.exec("VACUUM;");
}
pub fn countRows(database: *db.Db) db.Error!i64 {
return database.queryInt("SELECT count(*) FROM query_log");
}
pub fn countDomains(database: *db.Db) db.Error!i64 {
return database.queryInt("SELECT count(*) FROM domains");
}
// ---------------------------------------------------------------------------
// the API read layer (`GET /api/queries`, `GET /api/overview`)
// ---------------------------------------------------------------------------
/// One row of `GET /api/queries`, joined back through the `domains` dimension.
///
/// A summary projection, deliberately narrower than `QueryDetail`: the list is
/// a table the operator scans, and the full provenance of a row is one request
/// away at `GET /api/queries/{id}`.
///
/// The nullable text columns read a NULL as `""` — the same convention
/// `Stmt.columnText` already uses. None of them is ever written as an empty
/// string, so the mapping loses nothing and the API layer can treat `""` as
/// "absent".
pub const QueryRow = struct {
id: i64,
ts: i64,
domain: []const u8,
client_ip: []const u8,
qtype: ?u16,
qclass: u16,
rcode: u12,
blocked: bool,
response_time_us: ?i64,
cache_hit: ?bool,
upstream: []const u8,
policy_action: provenance.PolicyAction,
policy_reason: provenance.PolicyReason,
route_kind: provenance.RouteKind,
};
/// Everything one `query_log` row records about one query, for
/// `GET /api/queries/{id}`.
///
/// Same NULL-reads-as-`""` convention as `QueryRow`, and the same closed enums:
/// a stored value the schema does not define is `error.Mismatch`, never passed
/// through as text.
pub const QueryDetail = struct {
id: i64,
ts: i64,
domain: []const u8,
client_ip: []const u8,
qtype: ?u16,
qclass: u16,
rcode: u12,
blocked: bool,
response_time_us: ?i64,
cache_hit: ?bool,
upstream: []const u8,
group_id: ?i64,
group_name: []const u8,
policy_action: provenance.PolicyAction,
policy_reason: provenance.PolicyReason,
matched: []const u8,
source_id: ?i64,
source_name: []const u8,
cname_target: []const u8,
safe_search_target: []const u8,
route_kind: provenance.RouteKind,
forward_zone: []const u8,
};
/// Every field is an independent narrowing; `null` means "do not filter on it".
///
/// `since` is inclusive and `until` is exclusive, so adjacent windows tile
/// without double-counting a row on the boundary.
pub const QueryFilter = struct {
limit: u32 = 100,
/// Keyset cursor: only rows with a strictly smaller `id`. Rows come back
/// newest-first, so this is the id of the last row of the previous page.
before: ?i64 = null,
/// Matched case-insensitively for ASCII, which is what SQLite's `LIKE`
/// does and what a domain search wants.
domain_substring: ?[]const u8 = null,
client: ?[]const u8 = null,
blocked: ?bool = null,
since: ?i64 = null,
until: ?i64 = null,
};
/// Ruling 11 caps the page at 1000; the repository enforces it too, so a caller
/// that forgets cannot ask this connection for the whole table.
pub const max_limit: u32 = 1000;
const select_head =
\\SELECT q.id, q.timestamp, d.domain, q.client_ip, q.qtype, q.blocked,
\\ q.response_time_us, q.cache_hit, q.upstream, q.qclass, q.rcode,
\\ q.policy_action, q.policy_reason, q.route_kind
\\ FROM query_log q JOIN domains d ON d.id = q.domain_id
;
/// The escape character of `where_domain`. SQLite does not give string literals
/// C escapes, so `'\'` in the SQL text is one backslash.
const like_escape = '\\';
const where_before = " q.id < ?";
const where_domain = " d.domain LIKE ? ESCAPE '\\'";
const where_client = " q.client_ip = ?";
const where_blocked = " q.blocked = ?";
const where_since = " q.timestamp >= ?";
const where_until = " q.timestamp < ?";
const select_tail = " ORDER BY q.id DESC LIMIT ?";
const where_keyword = " WHERE";
const and_keyword = " AND";
/// Assembles the statement from the fixed fragments above and nothing else.
///
/// **No value ever reaches this buffer.** Every filter contributes a `?` and is
/// bound afterwards, in the order the predicates were appended: an unnumbered
/// parameter takes the next free index, so append order and bind order are the
/// same single contract.
const Sql = struct {
/// `where_keyword` is longer than `and_keyword` and is used at most once,
/// so counting six of it bounds every reachable combination.
const capacity = select_head.len + 6 * where_keyword.len + select_tail.len +
where_before.len + where_domain.len + where_client.len +
where_blocked.len + where_since.len + where_until.len;
buf: [capacity]u8 = undefined,
len: usize = 0,
has_where: bool = false,
fn put(self: *Sql, fragment: []const u8) void {
@memcpy(self.buf[self.len..][0..fragment.len], fragment);
self.len += fragment.len;
}
fn predicate(self: *Sql, fragment: []const u8) void {
self.put(if (self.has_where) and_keyword else where_keyword);
self.has_where = true;
self.put(fragment);
}
fn text(self: *const Sql) []const u8 {
return self.buf[0..self.len];
}
};
/// Rows come back newest-first (`id DESC`). Every string is allocated from
/// `arena`, including the list's own storage, so the caller frees the whole
/// result by resetting the arena — there is nothing to unwind on failure.
pub fn selectQueries(database: *db.Db, arena: Allocator, filter: QueryFilter) db.Error!std.ArrayList(QueryRow) {
var sql: Sql = .{};
sql.put(select_head);
if (filter.before != null) sql.predicate(where_before);
if (filter.domain_substring != null) sql.predicate(where_domain);
if (filter.client != null) sql.predicate(where_client);
if (filter.blocked != null) sql.predicate(where_blocked);
if (filter.since != null) sql.predicate(where_since);
if (filter.until != null) sql.predicate(where_until);
sql.put(select_tail);
var stmt = try database.prepare(sql.text());
defer stmt.deinit();
var idx: c_int = 0;
if (filter.before) |v| {
idx += 1;
try stmt.bindInt(idx, v);
}
if (filter.domain_substring) |v| {
idx += 1;
try stmt.bindText(idx, try likePattern(arena, v));
}
if (filter.client) |v| {
idx += 1;
try stmt.bindText(idx, v);
}
if (filter.blocked) |v| {
idx += 1;
try stmt.bindBool(idx, v);
}
if (filter.since) |v| {
idx += 1;
try stmt.bindInt(idx, v);
}
if (filter.until) |v| {
idx += 1;
try stmt.bindInt(idx, v);
}
idx += 1;
try stmt.bindInt(idx, @min(filter.limit, max_limit));
var out: std.ArrayList(QueryRow) = .empty;
while (try stmt.step()) {
try out.append(arena, .{
.id = stmt.columnInt(0),
.ts = stmt.columnInt(1),
.domain = try stmt.columnTextAlloc(arena, 2),
.client_ip = try stmt.columnTextAlloc(arena, 3),
.qtype = if (stmt.isNull(4)) null else std.math.cast(u16, stmt.columnInt(4)) orelse
return error.Mismatch,
.blocked = stmt.columnBool(5),
.response_time_us = if (stmt.isNull(6)) null else stmt.columnInt(6),
.cache_hit = if (stmt.isNull(7)) null else stmt.columnBool(7),
.upstream = try stmt.columnTextAlloc(arena, 8),
.qclass = try columnU16(&stmt, 9),
.rcode = try columnU12(&stmt, 10),
.policy_action = try provenance.parse(provenance.PolicyAction, stmt.columnText(11)),
.policy_reason = try provenance.parse(provenance.PolicyReason, stmt.columnText(12)),
.route_kind = try provenance.parse(provenance.RouteKind, stmt.columnText(13)),
});
}
return out;
}
/// A `NOT NULL` integer column that the schema bounds to 16 bits. A value
/// outside that range means the row came from something other than this schema.
fn columnU16(stmt: *db.Stmt, col: c_int) db.Error!u16 {
return std.math.cast(u16, stmt.columnInt(col)) orelse error.Mismatch;
}
/// The `rcode` column, which the schema's `CHECK` bounds to twelve bits. This
/// build cannot write a wider value — the field is a `u12` all the way from the
/// handler — so a row that carries one was written by something else, and is
/// `error.Mismatch` rather than a value truncated into shape.
fn columnU12(stmt: *db.Stmt, col: c_int) db.Error!u12 {
return std.math.cast(u12, stmt.columnInt(col)) orelse error.Mismatch;
}
const select_detail_sql =
\\SELECT q.id, q.timestamp, d.domain, q.client_ip, q.qtype, q.blocked,
\\ q.response_time_us, q.cache_hit, q.upstream, q.qclass, q.rcode,
\\ q.group_id, q.group_name, q.policy_action, q.policy_reason,
\\ q.matched, q.source_id, q.source_name, q.cname_target,
\\ q.safe_search_target, q.route_kind, q.forward_zone
\\ FROM query_log q JOIN domains d ON d.id = q.domain_id
\\ WHERE q.id = ?1
;
/// One row's full provenance, or `null` when no row has that id — which is what
/// an id the operator kept from before a retention pass looks like, and is a
/// 404 rather than an error.
///
/// Every string is allocated from `arena`, on the same terms as
/// `selectQueries`.
pub fn detailById(database: *db.Db, arena: Allocator, id: i64) db.Error!?QueryDetail {
var stmt = try database.prepare(select_detail_sql);
defer stmt.deinit();
try stmt.bindInt(1, id);
if (!try stmt.step()) return null;
return .{
.id = stmt.columnInt(0),
.ts = stmt.columnInt(1),
.domain = try stmt.columnTextAlloc(arena, 2),
.client_ip = try stmt.columnTextAlloc(arena, 3),
.qtype = if (stmt.isNull(4)) null else try columnU16(&stmt, 4),
.blocked = stmt.columnBool(5),
.response_time_us = if (stmt.isNull(6)) null else stmt.columnInt(6),
.cache_hit = if (stmt.isNull(7)) null else stmt.columnBool(7),
.upstream = try stmt.columnTextAlloc(arena, 8),
.qclass = try columnU16(&stmt, 9),
.rcode = try columnU12(&stmt, 10),
.group_id = if (stmt.isNull(11)) null else stmt.columnInt(11),
.group_name = try stmt.columnTextAlloc(arena, 12),
.policy_action = try provenance.parse(provenance.PolicyAction, stmt.columnText(13)),
.policy_reason = try provenance.parse(provenance.PolicyReason, stmt.columnText(14)),
.matched = try stmt.columnTextAlloc(arena, 15),
.source_id = if (stmt.isNull(16)) null else stmt.columnInt(16),
.source_name = try stmt.columnTextAlloc(arena, 17),
.cname_target = try stmt.columnTextAlloc(arena, 18),
.safe_search_target = try stmt.columnTextAlloc(arena, 19),
.route_kind = try provenance.parse(provenance.RouteKind, stmt.columnText(20)),
.forward_zone = try stmt.columnTextAlloc(arena, 21),
};
}
/// Wraps `needle` in `%` and neutralises the two `LIKE` metacharacters, so a
/// user searching for `a_b` gets domains containing `a_b` and not domains
/// containing `axb`. The escape character escapes itself.
fn likePattern(arena: Allocator, needle: []const u8) Allocator.Error![]const u8 {
var out: std.ArrayList(u8) = try .initCapacity(arena, needle.len * 2 + 2);
out.appendAssumeCapacity('%');
for (needle) |ch| {
if (ch == '%' or ch == '_' or ch == like_escape) out.appendAssumeCapacity(like_escape);
out.appendAssumeCapacity(ch);
}
out.appendAssumeCapacity('%');
return out.items;
}
/// The Overview rollup for one window. `avg_response_time_us` is `null` when no
/// row in the window recorded a response time. The mean is derived from a sum
/// and a count rather than SQL's `avg`, which returns REAL: `Stmt` reads
/// integers, and integer microseconds are exact.
pub const StatsTotals = struct {
queries: u64,
blocked: u64,
distinct_clients: u64,
avg_response_time_us: ?i64,
};
/// `count` and `sum` over non-negative columns cannot go negative; a negative
/// value means the row came from something other than this schema.
fn countOf(value: i64) db.Error!u64 {
if (value < 0) return error.Mismatch;
return @intCast(value);
}
/// One bucket of the Overview timeseries. `ts` is the bucket's inclusive start.
pub const Bucket = struct {
ts: i64,
queries: u64,
blocked: u64,
cached: u64,
};
/// One row of the Overview type breakdown. `qtype` is nullable in the schema, so
/// the rows that carry no type group into a row of their own rather than
/// disappearing from a breakdown that claims to add up.
///
/// No name field: the only qtype-name table lives in the admin, and a second
/// copy here would drift out of agreement with it.
///
/// The list carries no zero rows: a type absent from the window is absent from
/// it. Its order is an order, not an identity — a caller keys on the `qtype`
/// value, never on a row's position, because a rank change between refreshes
/// moves rows and must not move what they mean.
pub const TypeCount = struct {
qtype: ?u16,
count: u64,
};
/// One row of the Overview route breakdown: how a slice of the window was
/// answered.
///
/// `source` is the answering resolver's identity and nothing else — the
/// upstream url for `upstream` rows, the zone for `forward_zone` rows, null
/// everywhere else. It is deliberately not `source_name`, which names the
/// blocklist a block came from and would read as an upstream here.
///
/// A null source on an `upstream` row is its own group, not a dropped row: it
/// is a real state of the log and the caller labels it.
pub const RouteCount = struct {
route: provenance.RouteKind,
source: ?[]const u8,
count: u64,
};
/// How many clients the Overview names before the rest become `other`. Eight is
/// what one legend can carry without becoming a second table.
pub const max_client_series = 8;
/// One named client's series. `buckets` is always the caller's bucket count
/// long, zero-filled, and aligned exactly like `Overview.buckets`.
pub const ClientSeries = struct {
client: []const u8,
buckets: []const u64,
};
/// Named clients are ranked by in-window total, ties broken by address, so the
/// cut at `max_client_series` is the same cut on every read over the same data.
///
/// `other` is always present and always bucket-count sized — including for an
/// empty window and for a window with eight clients or fewer. A caller charting
/// a stack must not have to invent the residual series.
pub const ClientsBreakdown = struct {
clients: []const ClientSeries,
other: []const u64,
};
fn zeroedBuckets(arena: Allocator, bucket_count: u32) Allocator.Error![]u64 {
const buckets = try arena.alloc(u64, bucket_count);
@memset(buckets, 0);
return buckets;
}
// ---------------------------------------------------------------------------
// the Overview read path (milestone 36)
// ---------------------------------------------------------------------------
/// Everything one Overview response reports about one window, from one read of
/// one already-open transaction.
///
/// Storage owns these scalars: `since`, `bucket_seconds` and `bucket_count` are
/// numbers, not a web module's period type, so nothing here has to know the
/// period grammar the handler enforces.
pub const Overview = struct {
totals: StatsTotals,
/// `bucket_count` entries, one per bucket of the window, zero-filled and
/// aligned to `since`.
buckets: []const Bucket,
clients: ClientsBreakdown,
types: []const TypeCount,
routes: []const RouteCount,
};
/// One client's running window total beside its per-bucket series, so the cut
/// at `max_client_series` can be made after the whole window is folded rather
/// than by a second pass over the data.
const ClientAccum = struct {
client: []const u8,
total: u64,
buckets: []u64,
};
/// The shared shape both paths fold into. Every allocation is from `arena`, so
/// there is nothing to unwind on failure.
const Accum = struct {
arena: Allocator,
since: i64,
width: i64,
bucket_count: u32,
buckets: []Bucket,
rt_sum: i64 = 0,
rt_count: i64 = 0,
clients: std.ArrayList(ClientAccum) = .empty,
/// `client_ip` to its index in `clients`. The names are arena copies, so
/// they outlive the statement that produced them.
client_index: std.StringHashMapUnmanaged(u32) = .empty,
types: std.ArrayList(TypeCount) = .empty,
type_index: std.AutoHashMapUnmanaged(i64, u32) = .empty,
/// Household cardinality: a handful of upstreams plus six route kinds. A
/// linear scan is the whole index this needs.
routes: std.ArrayList(RouteCount) = .empty,
fn init(arena: Allocator, since: i64, width: i64, bucket_count: u32) Allocator.Error!Accum {
const buckets = try arena.alloc(Bucket, bucket_count);
for (buckets, 0..) |*bucket, i| bucket.* = .{
.ts = since + width * @as(i64, @intCast(i)),
.queries = 0,
.blocked = 0,
.cached = 0,
};
return .{ .arena = arena, .since = since, .width = width, .bucket_count = bucket_count, .buckets = buckets };
}
/// The serving-bucket index a grid bucket or a raw timestamp falls in.
/// Both paths bound their reads by the window already; the check is what
/// keeps a schema surprise from writing past the slice.
fn indexOf(self: *const Accum, timestamp: i64) db.Error!usize {
const offset = @divFloor(timestamp - self.since, self.width);
const index = std.math.cast(usize, offset) orelse return error.Mismatch;
if (index >= self.buckets.len) return error.Mismatch;
return index;
}
fn addTotals(self: *Accum, index: usize, queries: u64, blocked: u64, cached: u64) void {
self.buckets[index].queries += queries;
self.buckets[index].blocked += blocked;
self.buckets[index].cached += cached;
}
fn addClient(self: *Accum, client: []const u8, index: usize, count: u64) Allocator.Error!void {
const slot = try self.client_index.getOrPut(self.arena, client);
if (!slot.found_existing) {
const owned = try self.arena.dupe(u8, client);
slot.key_ptr.* = owned;
slot.value_ptr.* = @intCast(self.clients.items.len);
try self.clients.append(self.arena, .{
.client = owned,
.total = 0,
.buckets = try zeroedBuckets(self.arena, self.bucket_count),
});
}
const accum = &self.clients.items[slot.value_ptr.*];
accum.total += count;
accum.buckets[index] += count;
}
/// `qtype` is the stored encoding: `null_qtype` for the rows that carry no
/// query type.
fn addType(self: *Accum, qtype: i64, count: u64) db.Error!void {
const slot = try self.type_index.getOrPut(self.arena, qtype);
if (!slot.found_existing) {
slot.value_ptr.* = @intCast(self.types.items.len);
try self.types.append(self.arena, .{
.qtype = if (qtype == null_qtype) null else std.math.cast(u16, qtype) orelse
return error.Mismatch,
.count = 0,
});
}
self.types.items[slot.value_ptr.*].count += count;
}
fn addRoute(self: *Accum, route: provenance.RouteKind, source: ?[]const u8, count: u64) Allocator.Error!void {
for (self.routes.items) |*existing| {
if (existing.route == route and sameSource(existing.source, source)) {
existing.count += count;
return;
}
}
try self.routes.append(self.arena, .{
.route = route,
.source = if (source) |s| try self.arena.dupe(u8, s) else null,
.count = count,
});
}
fn finish(self: *Accum) Allocator.Error!Overview {
var queries: u64 = 0;
var blocked: u64 = 0;
for (self.buckets) |bucket| {
queries += bucket.queries;
blocked += bucket.blocked;
}
// Every comparator below is a total order on a unique key, so an
// unstable sort still produces one list for one window.
std.sort.pdq(TypeCount, self.types.items, {}, typeBefore);
std.sort.pdq(RouteCount, self.routes.items, {}, routeBefore);
std.sort.pdq(ClientAccum, self.clients.items, {}, clientBefore);
const named = @min(self.clients.items.len, max_client_series);
const other = try zeroedBuckets(self.arena, self.bucket_count);
for (self.clients.items[named..]) |folded| {
for (other, folded.buckets) |*slot, count| slot.* += count;
}
const series = try self.arena.alloc(ClientSeries, named);
for (series, self.clients.items[0..named]) |*entry, accum| {
entry.* = .{ .client = accum.client, .buckets = accum.buckets };
}
return .{
.totals = .{
.queries = queries,
.blocked = blocked,
.distinct_clients = self.clients.items.len,
.avg_response_time_us = if (self.rt_count == 0)
null
else
@divTrunc(self.rt_sum, self.rt_count),
},
.buckets = self.buckets,
.clients = .{ .clients = series, .other = other },
.types = self.types.items,
.routes = self.routes.items,
};
}
};
/// Absent and present are different sources, and so are two different present
/// ones: an `upstream` row the log recorded no resolver for is its own group,
/// exactly as `GROUP BY ... source` makes it.
fn sameSource(a: ?[]const u8, b: ?[]const u8) bool {
const left = a orelse return b == null;
const right = b orelse return false;
return std.mem.eql(u8, left, right);
}
/// The oracle's `ORDER BY count(*) DESC, qtype IS NULL, qtype ASC`.
fn typeBefore(_: void, a: TypeCount, b: TypeCount) bool {
if (a.count != b.count) return a.count > b.count;
const a_type = a.qtype orelse return false;
const b_type = b.qtype orelse return true;
return a_type < b_type;
}
/// The oracle's `ORDER BY count(*) DESC, route_kind ASC, source IS NULL,
/// source ASC`. `route_kind` is compared as the stored text, not as the
/// enum's declaration order — the two disagree, and the stored text is what the
/// SQL sorted.
fn routeBefore(_: void, a: RouteCount, b: RouteCount) bool {
if (a.count != b.count) return a.count > b.count;
switch (std.mem.order(u8, @tagName(a.route), @tagName(b.route))) {
.lt => return true,
.gt => return false,
.eq => {},
}
const a_source = a.source orelse return false;
const b_source = b.source orelse return true;
return std.mem.order(u8, a_source, b_source) == .lt;
}
/// The oracle's `ORDER BY count(*) DESC, client_ip ASC`, which is what makes
/// the cut at `max_client_series` the same cut on every read.
fn clientBefore(_: void, a: ClientAccum, b: ClientAccum) bool {
if (a.total != b.total) return a.total > b.total;
return std.mem.order(u8, a.client, b.client) == .lt;
}
const overview_projection_totals_sql =
\\SELECT bucket, queries, blocked, cached, rt_sum, rt_count
\\ FROM bucket_totals
\\ WHERE bucket >= ?1 AND bucket < ?2
;
const overview_projection_clients_sql =
\\SELECT bucket, client_ip, queries
\\ FROM bucket_clients
\\ WHERE bucket >= ?1 AND bucket < ?2
;
const overview_projection_types_sql =
\\SELECT qtype, sum(count)
\\ FROM bucket_types
\\ WHERE bucket >= ?1 AND bucket < ?2
\\ GROUP BY qtype
;
const overview_projection_routes_sql =
\\SELECT route_kind, source_present, source_text, sum(count)
\\ FROM bucket_routes
\\ WHERE bucket >= ?1 AND bucket < ?2
\\ GROUP BY route_kind, source_present, source_text
;
const overview_raw_sql =
\\SELECT timestamp, client_ip, qtype, blocked, cache_hit, response_time_us,
\\ route_kind,
\\ CASE route_kind
\\ WHEN 'upstream' THEN upstream
\\ WHEN 'forward_zone' THEN forward_zone
\\ END AS source
\\ FROM query_log
\\ WHERE timestamp >= ?1 AND timestamp < ?2
;
/// The whole Overview payload for `[since, since + bucket_seconds *
/// bucket_count)`, from one already-open read transaction — the caller owns the
/// transaction and the lock, as it does for every other read here.
///
/// Two implementations behind one entry point. At or above the projection grain
/// the four `bucket_*` tables answer the window, and the cost stops depending
/// on how many raw rows it holds; below it (the 1h window, on 60-second
/// buckets) one pass over the raw rows in the window does the same work in Zig.
/// Both produce the same contracts, which is what the equivalence test asserts.
///
/// Preconditions are caller bugs, not runtime conditions: a zero width or count
/// is `error.Misuse`, and so is a projection-path window that is not on the
/// grid, because the projections cannot express it. The handler's `window()`
/// guarantees all of them.
pub fn overview(
database: *db.Db,
arena: Allocator,
since: i64,
bucket_seconds: u32,
bucket_count: u32,
) db.Error!Overview {
if (bucket_seconds == 0 or bucket_count == 0) return error.Misuse;
const width: i64 = bucket_seconds;
const span = std.math.mul(i64, width, bucket_count) catch return error.Misuse;
const until = std.math.add(i64, since, span) catch return error.Misuse;
const from_projections = width >= grain;
if (from_projections and (@mod(since, grain) != 0 or @rem(width, grain) != 0)) return error.Misuse;
var accum = try Accum.init(arena, since, width, bucket_count);
if (from_projections) {
try readProjections(database, &accum, since, until);
} else {
try readRaw(database, &accum, since, until);
}
return try accum.finish();
}
fn readProjections(database: *db.Db, accum: *Accum, since: i64, until: i64) db.Error!void {
{
var stmt = try database.prepare(overview_projection_totals_sql);
defer stmt.deinit();
try stmt.bindInt(1, since);
try stmt.bindInt(2, until);
while (try stmt.step()) {
const index = try accum.indexOf(stmt.columnInt(0));
accum.addTotals(
index,
try countOf(stmt.columnInt(1)),
try countOf(stmt.columnInt(2)),
try countOf(stmt.columnInt(3)),
);
accum.rt_sum += stmt.columnInt(4);
accum.rt_count += stmt.columnInt(5);
}
}
{
var stmt = try database.prepare(overview_projection_clients_sql);
defer stmt.deinit();
try stmt.bindInt(1, since);
try stmt.bindInt(2, until);
while (try stmt.step()) {
const index = try accum.indexOf(stmt.columnInt(0));
try accum.addClient(stmt.columnText(1), index, try countOf(stmt.columnInt(2)));
}
}
{
var stmt = try database.prepare(overview_projection_types_sql);
defer stmt.deinit();
try stmt.bindInt(1, since);
try stmt.bindInt(2, until);
while (try stmt.step()) {
try accum.addType(stmt.columnInt(0), try countOf(stmt.columnInt(1)));
}
}
{
var stmt = try database.prepare(overview_projection_routes_sql);
defer stmt.deinit();
try stmt.bindInt(1, since);
try stmt.bindInt(2, until);
while (try stmt.step()) {
const route = try provenance.parse(provenance.RouteKind, stmt.columnText(0));
const source: ?[]const u8 = if (stmt.columnInt(1) == 0) null else stmt.columnText(2);
try accum.addRoute(route, source, try countOf(stmt.columnInt(3)));
}
}
}
fn readRaw(database: *db.Db, accum: *Accum, since: i64, until: i64) db.Error!void {
var stmt = try database.prepare(overview_raw_sql);
defer stmt.deinit();
try stmt.bindInt(1, since);
try stmt.bindInt(2, until);
while (try stmt.step()) {
const index = try accum.indexOf(stmt.columnInt(0));
const blocked: u64 = @intFromBool(stmt.columnBool(3));
// `cache_hit = 1`, not "non-zero": the SQL aggregates this replaces
// compare against the literal, and a NULL compares to neither.
const cached: u64 = @intFromBool(!stmt.isNull(4) and stmt.columnInt(4) == 1);
accum.addTotals(index, 1, blocked, cached);
if (!stmt.isNull(5)) {
accum.rt_sum += stmt.columnInt(5);
accum.rt_count += 1;
}
try accum.addClient(stmt.columnText(1), index, 1);
try accum.addType(if (stmt.isNull(2)) null_qtype else stmt.columnInt(2), 1);
try accum.addRoute(
try provenance.parse(provenance.RouteKind, stmt.columnText(6)),
stmt.columnTextOrNull(7),
1,
);
}
}
// ---------------------------------------------------------------------------
// tests
// ---------------------------------------------------------------------------
const logger = @import("../logger.zig");
const querylog_schema = @import("../querylog_schema.zig");
const testing = std.testing;
fn openLog() !db.Db {
var database = try db.Db.open(":memory:", .{ .mode = .memory });
errdefer database.close();
try db.applyPragmas(&database, .{});
try database.exec(querylog_schema.ddl);
return database;
}
fn plainRow(timestamp: i64, domain: []const u8) Row {
return .{
.timestamp = timestamp,
.domain = domain,
.client_ip = "192.0.2.10",
.qtype = 1,
.qclass = 1,
.rcode = 0,
.blocked = false,
.response_time_us = 1200,
.cache_hit = false,
.upstream = "9.9.9.9",
.group_id = 1,
.group_name = "default",
.policy_action = .allow,
.policy_reason = .no_match,
.matched = null,
.source_id = null,
.source_name = null,
.cname_target = null,
.safe_search_target = null,
.route_kind = .upstream,
.forward_zone = null,
};
}
fn domainIdOf(database: *db.Db, domain: []const u8) !i64 {
var stmt = try database.prepare(select_domain_sql);
defer stmt.deinit();
try stmt.bindText(1, domain);
try testing.expect(try stmt.step());
return stmt.columnInt(0);
}
test "a foreign row with an rcode wider than twelve bits is refused, not truncated" {
var database = try db.Db.open(":memory:", .{ .mode = .memory });
defer database.close();
try db.applyPragmas(&database, .{});
// The shipped schema's `CHECK` makes this row impossible in a file this
// build created, so the table is built without it. The read path's job is
// to refuse a `querylog.db` that came from somewhere else rather than to
// narrow a value it cannot represent.
try database.exec(
\\CREATE TABLE domains (id INTEGER PRIMARY KEY, domain TEXT NOT NULL UNIQUE);
\\CREATE TABLE query_log (
\\ id INTEGER PRIMARY KEY, timestamp INTEGER NOT NULL,
\\ domain_id INTEGER NOT NULL, client_ip TEXT NOT NULL,
\\ qtype INTEGER, blocked INTEGER NOT NULL, response_time_us INTEGER,
\\ cache_hit INTEGER, upstream TEXT, qclass INTEGER NOT NULL,
\\ rcode INTEGER NOT NULL, group_id INTEGER, group_name TEXT,
\\ policy_action TEXT NOT NULL, policy_reason TEXT NOT NULL,
\\ matched TEXT, source_id INTEGER, source_name TEXT,
\\ cname_target TEXT, safe_search_target TEXT,
\\ route_kind TEXT NOT NULL, forward_zone TEXT
\\);
\\INSERT INTO domains (id, domain) VALUES (1, 'a.example');
\\INSERT INTO query_log
\\ (id, timestamp, domain_id, client_ip, blocked, qclass, rcode,
\\ policy_action, policy_reason, route_kind)
\\VALUES (1, 10, 1, '192.0.2.10', 0, 1, 4096, 'allow', 'no_match', 'upstream');
);
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
try testing.expectError(error.Mismatch, selectQueries(&database, arena, .{}));
try testing.expectError(error.Mismatch, detailById(&database, arena, 1));
}
test "writeBatch inserts every row and interns each domain once" {
var database = try openLog();
defer database.close();
var writer = try BatchWriter.init(testing.allocator, &database);
defer writer.deinit();
try writer.writeBatch(&.{
plainRow(100, "example.com"),
plainRow(101, "example.com"),
plainRow(102, "ads.example.net"),
});
try testing.expectEqual(@as(i64, 3), try countRows(&database));
try testing.expectEqual(@as(i64, 2), try countDomains(&database));
const first = try domainIdOf(&database, "example.com");
try testing.expectEqual(
@as(i64, 2),
try database.queryInt("SELECT count(*) FROM query_log WHERE domain_id = 1"),
);
try testing.expectEqual(@as(i64, 1), first);
}
test "a second batch reuses the interned domain id" {
var database = try openLog();
defer database.close();
var writer = try BatchWriter.init(testing.allocator, &database);
defer writer.deinit();
try writer.writeBatch(&.{plainRow(100, "example.com")});
const before = try domainIdOf(&database, "example.com");
try writer.writeBatch(&.{ plainRow(200, "example.com"), plainRow(201, "other.example") });
const after = try domainIdOf(&database, "example.com");
try testing.expectEqual(before, after);
try testing.expectEqual(@as(i64, 3), try countRows(&database));
try testing.expectEqual(@as(i64, 2), try countDomains(&database));
try testing.expectEqual(
@as(i64, 2),
try database.queryInt("SELECT count(*) FROM query_log WHERE domain_id = 1"),
);
}
test "every column round-trips a value and a null" {
var database = try openLog();
defer database.close();
var writer = try BatchWriter.init(testing.allocator, &database);
defer writer.deinit();
try writer.writeBatch(&.{
.{
.timestamp = 10,
.domain = "blocked.example",
.client_ip = "2001:db8::1",
.qtype = 28,
.qclass = 1,
.rcode = 3,
.blocked = true,
.response_time_us = 42,
.cache_hit = true,
.upstream = "https://dns.example/dns-query",
.group_id = 7,
.group_name = "kids",
.policy_action = .block,
.policy_reason = .blocklist_wildcard,
.matched = "*.ads.example",
.source_id = 3,
.source_name = "steven black",
.cname_target = "tracker.cdn.example",
.safe_search_target = "forcesafesearch.google.com",
.route_kind = .blocked,
.forward_zone = "home.arpa",
},
// Every nullable column absent at once, which is the shape of a query
// the pipeline answered before any of them applied.
.{
.timestamp = 11,
.domain = "quiet.example",
.client_ip = "hidden",
.qtype = null,
.qclass = 3,
.rcode = 0,
.blocked = false,
.response_time_us = null,
.cache_hit = null,
.upstream = null,
.group_id = null,
.group_name = null,
.policy_action = .not_evaluated,
.policy_reason = .non_in_class,
.matched = null,
.source_id = null,
.source_name = null,
.cname_target = null,
.safe_search_target = null,
.route_kind = .upstream,
.forward_zone = null,
},
});
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
const full = (try detailById(&database, arena, 1)).?;
try testing.expectEqualStrings("blocked.example", full.domain);
try testing.expectEqualStrings("2001:db8::1", full.client_ip);
try testing.expectEqual(@as(?u16, 28), full.qtype);
try testing.expectEqual(@as(u16, 1), full.qclass);
try testing.expectEqual(@as(u12, 3), full.rcode);
try testing.expect(full.blocked);
try testing.expectEqual(@as(?i64, 42), full.response_time_us);
try testing.expectEqual(@as(?bool, true), full.cache_hit);
try testing.expectEqualStrings("https://dns.example/dns-query", full.upstream);
try testing.expectEqual(@as(?i64, 7), full.group_id);
try testing.expectEqualStrings("kids", full.group_name);
try testing.expectEqual(provenance.PolicyAction.block, full.policy_action);
try testing.expectEqual(provenance.PolicyReason.blocklist_wildcard, full.policy_reason);
try testing.expectEqualStrings("*.ads.example", full.matched);
try testing.expectEqual(@as(?i64, 3), full.source_id);
try testing.expectEqualStrings("steven black", full.source_name);
try testing.expectEqualStrings("tracker.cdn.example", full.cname_target);
try testing.expectEqualStrings("forcesafesearch.google.com", full.safe_search_target);
try testing.expectEqual(provenance.RouteKind.blocked, full.route_kind);
try testing.expectEqualStrings("home.arpa", full.forward_zone);
// A NULL text column reads as the empty string, by the documented
// convention; a NULL integer stays null, because 0 is a real id.
const bare = (try detailById(&database, arena, 2)).?;
try testing.expectEqual(@as(?u16, null), bare.qtype);
try testing.expectEqual(@as(u16, 3), bare.qclass);
try testing.expectEqual(@as(?i64, null), bare.response_time_us);
try testing.expectEqual(@as(?bool, null), bare.cache_hit);
try testing.expectEqualStrings("", bare.upstream);
try testing.expectEqual(@as(?i64, null), bare.group_id);
try testing.expectEqualStrings("", bare.group_name);
try testing.expectEqual(provenance.PolicyAction.not_evaluated, bare.policy_action);
try testing.expectEqual(provenance.PolicyReason.non_in_class, bare.policy_reason);
try testing.expectEqualStrings("", bare.matched);
try testing.expectEqual(@as(?i64, null), bare.source_id);
try testing.expectEqualStrings("", bare.source_name);
try testing.expectEqualStrings("", bare.cname_target);
try testing.expectEqualStrings("", bare.safe_search_target);
try testing.expectEqual(provenance.RouteKind.upstream, bare.route_kind);
try testing.expectEqualStrings("", bare.forward_zone);
}
test "detailById returns null for an id no row has" {
var database = try openLog();
defer database.close();
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
try seed(&database, &.{plainRow(10, "a.example")});
// An id from before a retention pass looks exactly like this, and is a 404
// rather than an error.
try testing.expectEqual(@as(?QueryDetail, null), try detailById(&database, arena_state.allocator(), 2));
try testing.expectEqual(@as(?QueryDetail, null), try detailById(&database, arena_state.allocator(), 0));
try testing.expect((try detailById(&database, arena_state.allocator(), 1)) != null);
}
test "a stored enum value the schema does not define is a data error, not a passthrough" {
var database = try openLog();
defer database.close();
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
try seed(&database, &.{plainRow(10, "a.example")});
try database.exec("UPDATE query_log SET policy_reason = 'whatever' WHERE id = 1;");
try testing.expectError(error.Mismatch, detailById(&database, arena, 1));
try testing.expectError(error.Mismatch, selectQueries(&database, arena, .{}));
}
test "an empty batch writes nothing and opens no transaction" {
var database = try openLog();
defer database.close();
var writer = try BatchWriter.init(testing.allocator, &database);
defer writer.deinit();
// A transaction is already open, so a `BEGIN IMMEDIATE` from `writeBatch`
// would fail: this is what proves the empty batch returns before it.
var tx = try db.Tx.begin(&database);
try writer.writeBatch(&.{});
tx.rollback();
try testing.expectEqual(@as(i64, 0), try countRows(&database));
try testing.expectEqual(@as(i64, 0), try countDomains(&database));
}
test "pruneOlderThan deletes strictly older rows and returns the count" {
var database = try openLog();
defer database.close();
var writer = try BatchWriter.init(testing.allocator, &database);
defer writer.deinit();
try writer.writeBatch(&.{
plainRow(100, "old.example"),
plainRow(199, "old.example"),
plainRow(200, "edge.example"),
plainRow(300, "fresh.example"),
});
try testing.expectEqual(@as(i64, 2), (try pruneOlderThan(&database, 200)).deleted);
try testing.expectEqual(@as(i64, 2), try countRows(&database));
// The row exactly at the cutoff stays.
try testing.expectEqual(
@as(i64, 1),
try database.queryInt("SELECT count(*) FROM query_log WHERE timestamp = 200"),
);
// A second pass over the same cutoff finds nothing left to do.
try testing.expectEqual(@as(i64, 0), (try pruneOlderThan(&database, 200)).deleted);
}
test "a prune advances the coverage watermark to its own cutoff" {
var database = try openLog();
defer database.close();
var writer = try BatchWriter.init(testing.allocator, &database);
defer writer.deinit();
// The seeded watermark is `created_at + 1`, which is now-ish; the cutoffs
// below are all in the past, so they start out behind it.
const start = try availableSince(&database);
try writer.writeBatch(&.{ plainRow(start + 100, "a.example"), plainRow(start + 300, "b.example") });
const first = try pruneOlderThan(&database, start + 200);
try testing.expectEqual(@as(i64, 1), first.deleted);
try testing.expectEqual(start + 200, first.available_since);
try testing.expectEqual(start + 200, try availableSince(&database));
// A prune that deletes nothing still advances: the window it swept is
// covered whether or not it held rows.
const second = try pruneOlderThan(&database, start + 250);
try testing.expectEqual(@as(i64, 0), second.deleted);
try testing.expectEqual(start + 250, try availableSince(&database));
}
test "the watermark never moves backward" {
var database = try openLog();
defer database.close();
const start = try availableSince(&database);
const advanced = try pruneOlderThan(&database, start + 1000);
try testing.expectEqual(start + 1000, advanced.available_since);
// A shortened `retention_days`, a clock that stepped back, a pass with a
// stale cutoff: none of them may widen the promise the file makes.
for ([_]i64{ start + 999, start, start - 100_000, 0 }) |older| {
const result = try pruneOlderThan(&database, older);
try testing.expectEqual(start + 1000, result.available_since);
try testing.expectEqual(start + 1000, try availableSince(&database));
}
}
test "a failed delete leaves both the rows and the watermark untouched" {
var database = try openLog();
defer database.close();
var writer = try BatchWriter.init(testing.allocator, &database);
defer writer.deinit();
const start = try availableSince(&database);
try writer.writeBatch(&.{plainRow(start - 100, "old.example")});
try database.exec(
\\CREATE TRIGGER refuse_delete BEFORE DELETE ON query_log
\\BEGIN SELECT RAISE(ABORT, 'refused'); END;
);
try testing.expectError(error.Constraint, pruneOlderThan(&database, start + 1000));
try testing.expectEqual(@as(i64, 1), try countRows(&database));
try testing.expectEqual(start, try availableSince(&database));
}
test "a failed watermark update leaves the rows it had already deleted" {
var database = try openLog();
defer database.close();
var writer = try BatchWriter.init(testing.allocator, &database);
defer writer.deinit();
const start = try availableSince(&database);
try writer.writeBatch(&.{plainRow(start - 100, "old.example")});
// The delete succeeds and the advance does not. Without one transaction
// around the pair, this is the case that loses rows the watermark still
// promises.
try database.exec(
\\CREATE TRIGGER refuse_advance BEFORE UPDATE ON querylog_meta
\\BEGIN SELECT RAISE(ABORT, 'refused'); END;
);
try testing.expectError(error.Constraint, pruneOlderThan(&database, start + 1000));
try testing.expectEqual(@as(i64, 1), try countRows(&database));
try testing.expectEqual(start, try availableSince(&database));
}
test "a failed commit rolls back the delete and the watermark together" {
var database = try openLog();
defer database.close();
var writer = try BatchWriter.init(testing.allocator, &database);
defer writer.deinit();
const start = try availableSince(&database);
try writer.writeBatch(&.{plainRow(start - 100, "old.example")});
// Both statements succeed and COMMIT is what fails: the advance inserts a
// `query_log` row whose `domain_id` references nothing, and
// `defer_foreign_keys` holds that violation back until the commit checks
// it (SQLite's documented semantics for the pragma; the assertions below
// observe the rollback, not the moment the check ran).
try database.exec(
\\CREATE TRIGGER break_at_commit AFTER UPDATE ON querylog_meta
\\BEGIN INSERT INTO query_log
\\ (timestamp, domain_id, client_ip, blocked, qclass, rcode,
\\ policy_action, policy_reason, route_kind)
\\VALUES (1, 999999, 'x', 0, 1, 0, 'allow', 'no_match', 'upstream'); END;
);
try database.exec("PRAGMA defer_foreign_keys = ON;");
try testing.expectError(error.Constraint, pruneOlderThan(&database, start + 1000));
// Nothing survived: not the delete, not the advance, not the row the
// trigger inserted.
try testing.expectEqual(@as(i64, 1), try countRows(&database));
try testing.expectEqual(start, try availableSince(&database));
try testing.expectEqual(
@as(i64, 0),
try database.queryInt("SELECT count(*) FROM query_log WHERE client_ip = 'x'"),
);
}
test "pruneOlderThan leaves the domains dimension table intact" {
var database = try openLog();
defer database.close();
var writer = try BatchWriter.init(testing.allocator, &database);
defer writer.deinit();
try writer.writeBatch(&.{ plainRow(10, "a.example"), plainRow(11, "b.example") });
try testing.expectEqual(@as(i64, 2), (try pruneOlderThan(&database, 1000)).deleted);
try testing.expectEqual(@as(i64, 0), try countRows(&database));
try testing.expectEqual(@as(i64, 2), try countDomains(&database));
}
test "a failing row rolls the whole batch back and the writer survives it" {
var database = try openLog();
defer database.close();
try database.exec(
\\CREATE TRIGGER refuse_boom BEFORE INSERT ON query_log
\\WHEN new.client_ip = 'boom'
\\BEGIN SELECT RAISE(ABORT, 'refused'); END;
);
var writer = try BatchWriter.init(testing.allocator, &database);
defer writer.deinit();
var doomed = plainRow(20, "second.example");
doomed.client_ip = "boom";
try testing.expectError(error.Constraint, writer.writeBatch(&.{
plainRow(10, "first.example"),
doomed,
}));
// The interned domain of the row that did insert is gone with it.
try testing.expectEqual(@as(i64, 0), try countRows(&database));
try testing.expectEqual(@as(i64, 0), try countDomains(&database));
try writer.writeBatch(&.{plainRow(30, "third.example")});
try testing.expectEqual(@as(i64, 1), try countRows(&database));
try testing.expectEqual(@as(i64, 1), try countDomains(&database));
}
test "countRows and countDomains agree with what the batches wrote" {
var database = try openLog();
defer database.close();
var writer = try BatchWriter.init(testing.allocator, &database);
defer writer.deinit();
try testing.expectEqual(@as(i64, 0), try countRows(&database));
try testing.expectEqual(@as(i64, 0), try countDomains(&database));
var rows: [50]Row = undefined;
var names: [50][16]u8 = undefined;
for (&rows, &names, 0..) |*row, *name, i| {
const written = std.fmt.bufPrint(name, "d{d}.example", .{i % 7}) catch unreachable;
row.* = plainRow(@intCast(i), written);
}
try writer.writeBatch(&rows);
try testing.expectEqual(@as(i64, 50), try countRows(&database));
try testing.expectEqual(@as(i64, 7), try countDomains(&database));
}
// `PRAGMA wal_checkpoint` needs a real WAL, which an in-memory database cannot
// have. `std.testing.tmpDir` creates its directory under `.zig-cache/tmp/`
// relative to the process working directory, which is also how SQLite's VFS
// resolves the filename it is handed (`storage_integration_test.zig:44`).
const tmp_prefix = ".zig-cache/tmp/";
const sub_path_len = @typeInfo(@FieldType(testing.TmpDir, "sub_path")).array.len;
test "checkpointTruncate and vacuum run against a WAL file database" {
var tmp = testing.tmpDir(.{});
defer tmp.cleanup();
var path_buf: [tmp_prefix.len + sub_path_len + 32]u8 = undefined;
const path = try std.fmt.bufPrintZ(&path_buf, "{s}{s}/querylog.db", .{ tmp_prefix, &tmp.sub_path });
var database = try db.Db.open(path, .{ .mode = .read_write_create });
defer database.close();
try db.applyPragmas(&database, .{});
{
var stmt = try database.prepare("PRAGMA journal_mode");
defer stmt.deinit();
try testing.expect(try stmt.step());
// `columnText` is borrowed until the next call on the statement, so it
// is compared here rather than carried out of this block.
try testing.expectEqualStrings("wal", stmt.columnText(0));
}
try database.exec(querylog_schema.ddl);
var writer = try BatchWriter.init(testing.allocator, &database);
defer writer.deinit();
try writer.writeBatch(&.{ plainRow(10, "a.example"), plainRow(20, "b.example") });
try checkpointTruncate(&database);
try testing.expectEqual(@as(i64, 1), (try pruneOlderThan(&database, 20)).deleted);
try checkpointTruncate(&database);
try vacuum(&database);
try testing.expectEqual(@as(i64, 1), try countRows(&database));
try testing.expectEqual(@as(i64, 2), try countDomains(&database));
}
// --- the read layer -------------------------------------------------------
/// `BatchWriter` assigns `query_log.id` in the order it is handed the rows, so
/// every test below knows the id of each seeded row: the nth row of the nth
/// batch has id n.
fn seed(database: *db.Db, rows: []const Row) !void {
var writer = try BatchWriter.init(testing.allocator, database);
defer writer.deinit();
try writer.writeBatch(rows);
}
fn ids(rows: []const QueryRow, out: []i64) []const i64 {
for (rows, 0..) |row, i| out[i] = row.id;
return out[0..rows.len];
}
test "selectQueries returns the newest row first and reads every column" {
var database = try openLog();
defer database.close();
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
try seed(&database, &.{
.{
.timestamp = 10,
.domain = "ads.example.net",
.client_ip = "192.0.2.10",
.qtype = 28,
.qclass = 1,
.rcode = 0,
.blocked = true,
.response_time_us = 4200,
.cache_hit = true,
.upstream = "https://dns.example/dns-query",
.group_id = 2,
.group_name = "kids",
.policy_action = .block,
.policy_reason = .blocklist_domain,
.matched = "ads.example.net",
.source_id = 5,
.source_name = "steven black",
.cname_target = null,
.safe_search_target = null,
.route_kind = .blocked,
.forward_zone = null,
},
.{
.timestamp = 20,
.domain = "quiet.example",
.client_ip = "hidden",
.qtype = null,
.qclass = 1,
.rcode = 2,
.blocked = false,
.response_time_us = null,
.cache_hit = null,
.upstream = null,
.group_id = null,
.group_name = null,
.policy_action = .not_evaluated,
.policy_reason = .snapshot_unavailable,
.matched = null,
.source_id = null,
.source_name = null,
.cname_target = null,
.safe_search_target = null,
.route_kind = .rejected,
.forward_zone = null,
},
});
const rows = try selectQueries(&database, arena_state.allocator(), .{});
try testing.expectEqual(@as(usize, 2), rows.items.len);
const newest = rows.items[0];
try testing.expectEqual(@as(i64, 2), newest.id);
try testing.expectEqual(@as(i64, 20), newest.ts);
try testing.expectEqualStrings("quiet.example", newest.domain);
try testing.expectEqualStrings("hidden", newest.client_ip);
try testing.expectEqual(@as(?u16, null), newest.qtype);
try testing.expectEqual(@as(u16, 1), newest.qclass);
try testing.expectEqual(@as(u12, 2), newest.rcode);
try testing.expect(!newest.blocked);
try testing.expectEqual(@as(?i64, null), newest.response_time_us);
try testing.expectEqual(@as(?bool, null), newest.cache_hit);
// A NULL text column reads as the empty string, by documented convention.
try testing.expectEqualStrings("", newest.upstream);
try testing.expectEqual(provenance.PolicyAction.not_evaluated, newest.policy_action);
try testing.expectEqual(provenance.PolicyReason.snapshot_unavailable, newest.policy_reason);
try testing.expectEqual(provenance.RouteKind.rejected, newest.route_kind);
const oldest = rows.items[1];
try testing.expectEqual(@as(i64, 1), oldest.id);
try testing.expectEqual(@as(i64, 10), oldest.ts);
try testing.expectEqualStrings("ads.example.net", oldest.domain);
try testing.expectEqualStrings("192.0.2.10", oldest.client_ip);
try testing.expectEqual(@as(?u16, 28), oldest.qtype);
try testing.expectEqual(@as(u16, 1), oldest.qclass);
try testing.expectEqual(@as(u12, 0), oldest.rcode);
try testing.expect(oldest.blocked);
try testing.expectEqual(@as(?i64, 4200), oldest.response_time_us);
try testing.expectEqual(@as(?bool, true), oldest.cache_hit);
try testing.expectEqualStrings("https://dns.example/dns-query", oldest.upstream);
try testing.expectEqual(provenance.PolicyAction.block, oldest.policy_action);
try testing.expectEqual(provenance.PolicyReason.blocklist_domain, oldest.policy_reason);
try testing.expectEqual(provenance.RouteKind.blocked, oldest.route_kind);
}
test "selectQueries honours the limit and caps it at max_limit" {
var database = try openLog();
defer database.close();
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
var rows: [1005]Row = undefined;
for (&rows, 0..) |*row, i| row.* = plainRow(@intCast(i), "example.com");
try seed(&database, &rows);
const few = try selectQueries(&database, arena, .{ .limit = 3 });
try testing.expectEqual(@as(usize, 3), few.items.len);
// Asked for more than the cap, and for more rows than the cap, so the cap
// is what bounds the answer rather than the table.
const capped = try selectQueries(&database, arena, .{ .limit = 5000 });
try testing.expectEqual(@as(usize, max_limit), capped.items.len);
const none = try selectQueries(&database, arena, .{ .limit = 0 });
try testing.expectEqual(@as(usize, 0), none.items.len);
}
test "keyset paging walks every row exactly once across the page boundaries" {
var database = try openLog();
defer database.close();
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
var seeded: [7]Row = undefined;
for (&seeded, 0..) |*row, i| row.* = plainRow(@intCast(i), "example.com");
try seed(&database, &seeded);
var seen: std.ArrayList(i64) = .empty;
defer seen.deinit(testing.allocator);
var before: ?i64 = null;
var pages: usize = 0;
while (pages < 10) : (pages += 1) {
const page = try selectQueries(&database, arena, .{ .limit = 3, .before = before });
if (page.items.len == 0) break;
for (page.items) |row| try seen.append(testing.allocator, row.id);
before = page.items[page.items.len - 1].id;
}
// Two full pages and one short page; the fourth call returns nothing and
// breaks before the counter, which is how the walk knows it is done.
try testing.expectEqual(@as(usize, 3), pages);
try testing.expectEqualSlices(i64, &.{ 7, 6, 5, 4, 3, 2, 1 }, seen.items);
}
test "each filter narrows the result on its own" {
var database = try openLog();
defer database.close();
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
var blocked_row = plainRow(200, "ads.example.net");
blocked_row.client_ip = "192.0.2.20";
blocked_row.blocked = true;
blocked_row.policy_action = .block;
blocked_row.policy_reason = .blocklist_domain;
blocked_row.route_kind = .blocked;
try seed(&database, &.{
plainRow(100, "one.example.com"),
blocked_row,
plainRow(300, "two.example.com"),
});
var buf: [8]i64 = undefined;
const by_domain = try selectQueries(&database, arena, .{ .domain_substring = "example.com" });
try testing.expectEqualSlices(i64, &.{ 3, 1 }, ids(by_domain.items, &buf));
const by_client = try selectQueries(&database, arena, .{ .client = "192.0.2.20" });
try testing.expectEqualSlices(i64, &.{2}, ids(by_client.items, &buf));
// An exact match, not a prefix: the seeded clients share the first octets.
const no_client = try selectQueries(&database, arena, .{ .client = "192.0.2" });
try testing.expectEqual(@as(usize, 0), no_client.items.len);
const only_blocked = try selectQueries(&database, arena, .{ .blocked = true });
try testing.expectEqualSlices(i64, &.{2}, ids(only_blocked.items, &buf));
const only_allowed = try selectQueries(&database, arena, .{ .blocked = false });
try testing.expectEqualSlices(i64, &.{ 3, 1 }, ids(only_allowed.items, &buf));
// Every filter at once, all satisfied by the one blocked row.
const combined = try selectQueries(&database, arena, .{
.limit = 10,
.before = 3,
.domain_substring = "ads",
.client = "192.0.2.20",
.blocked = true,
.since = 200,
.until = 300,
});
try testing.expectEqualSlices(i64, &.{2}, ids(combined.items, &buf));
}
test "since is inclusive, until is exclusive, and an empty range selects nothing" {
var database = try openLog();
defer database.close();
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
try seed(&database, &.{
plainRow(100, "a.example"),
plainRow(200, "b.example"),
plainRow(300, "c.example"),
});
var buf: [8]i64 = undefined;
const window = try selectQueries(&database, arena, .{ .since = 100, .until = 300 });
try testing.expectEqualSlices(i64, &.{ 2, 1 }, ids(window.items, &buf));
const after = try selectQueries(&database, arena, .{ .since = 300 });
try testing.expectEqualSlices(i64, &.{3}, ids(after.items, &buf));
const empty = try selectQueries(&database, arena, .{ .since = 300, .until = 300 });
try testing.expectEqual(@as(usize, 0), empty.items.len);
const beyond = try selectQueries(&database, arena, .{ .since = 1000 });
try testing.expectEqual(@as(usize, 0), beyond.items.len);
}
test "a domain substring matches % and _ as literal characters" {
var database = try openLog();
defer database.close();
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
try seed(&database, &.{
plainRow(10, "a_b.example"),
plainRow(20, "axb.example"),
plainRow(30, "a%b.example"),
plainRow(40, "azzb.example"),
plainRow(50, "back\\slash.example"),
});
var buf: [8]i64 = undefined;
// Unescaped, `_` is LIKE's single-character wildcard and would also match
// "axb"; escaped, it matches only the underscore.
const underscore = try selectQueries(&database, arena, .{ .domain_substring = "a_b" });
try testing.expectEqualSlices(i64, &.{1}, ids(underscore.items, &buf));
// Unescaped, `%` would match everything from "a" to "b", so "azzb" too.
const percent = try selectQueries(&database, arena, .{ .domain_substring = "a%b" });
try testing.expectEqualSlices(i64, &.{3}, ids(percent.items, &buf));
// The escape character escapes itself, so it is searchable as well.
const backslash = try selectQueries(&database, arena, .{ .domain_substring = "k\\s" });
try testing.expectEqualSlices(i64, &.{5}, ids(backslash.items, &buf));
// An empty needle is `%%`, which matches every row rather than none.
const all = try selectQueries(&database, arena, .{ .domain_substring = "" });
try testing.expectEqual(@as(usize, 5), all.items.len);
}
test "the built SQL never carries a filter value and fits its buffer" {
var sql: Sql = .{};
sql.put(select_head);
sql.predicate(where_before);
sql.predicate(where_domain);
sql.predicate(where_client);
sql.predicate(where_blocked);
sql.predicate(where_since);
sql.predicate(where_until);
sql.put(select_tail);
// Every predicate present is the longest reachable statement.
try testing.expect(sql.len <= Sql.capacity);
try testing.expectEqual(@as(usize, 1), std.mem.count(u8, sql.text(), " WHERE"));
try testing.expectEqual(@as(usize, 5), std.mem.count(u8, sql.text(), " AND"));
// Six filters plus the LIMIT, each a bare parameter.
try testing.expectEqual(@as(usize, 7), std.mem.count(u8, sql.text(), "?"));
}
test "likePattern wraps the needle and neutralises every metacharacter" {
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
try testing.expectEqualStrings("%plain%", try likePattern(arena, "plain"));
try testing.expectEqualStrings("%a\\_b%", try likePattern(arena, "a_b"));
try testing.expectEqualStrings("%a\\%b%", try likePattern(arena, "a%b"));
try testing.expectEqualStrings("%a\\\\b%", try likePattern(arena, "a\\b"));
try testing.expectEqualStrings("%%", try likePattern(arena, ""));
}
// ---------------------------------------------------------------------------
// period aggregations (milestone 30)
// ---------------------------------------------------------------------------
const agg_since: i64 = 1_700_000_000;
const agg_width: u32 = 60;
const agg_buckets: u32 = 10;
/// One row of the aggregation fixtures. Everything the three breakdowns read
/// is a parameter; everything else is the same on every row, so a test that
/// changes an outcome names the reason it changed.
fn aggRow(offset: i64, client: []const u8, qtype: ?u16, kind: provenance.RouteKind, source: ?[]const u8) Row {
return .{
.timestamp = agg_since + offset,
.domain = "example.com",
.client_ip = client,
.qtype = qtype,
.qclass = 1,
.rcode = 0,
.blocked = kind == .blocked,
.response_time_us = 1000,
.cache_hit = kind == .cache,
.upstream = if (kind == .upstream) source else null,
.group_id = 1,
.group_name = "default",
.policy_action = if (kind == .blocked) .block else .allow,
.policy_reason = if (kind == .blocked) .blocklist_domain else .no_match,
.matched = null,
.source_id = null,
// Blocklist provenance, deliberately set on every row: the routes
// breakdown must never group by it.
.source_name = "StevenBlack",
.cname_target = null,
.safe_search_target = null,
.route_kind = kind,
.forward_zone = if (kind == .forward_zone) source else null,
};
}
test "the type breakdown groups by qtype, keeps the null row and orders it last" {
var database = try openLog();
defer database.close();
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
try seed(&database, &.{
aggRow(0, "192.0.2.10", 1, .upstream, "9.9.9.9"),
aggRow(1, "192.0.2.10", 1, .upstream, "9.9.9.9"),
aggRow(2, "192.0.2.10", 1, .upstream, "9.9.9.9"),
aggRow(3, "192.0.2.10", 28, .upstream, "9.9.9.9"),
aggRow(4, "192.0.2.10", 28, .upstream, "9.9.9.9"),
// Ties with qtype 28, so the tie-break puts the lower code first.
aggRow(5, "192.0.2.10", 16, .upstream, "9.9.9.9"),
aggRow(6, "192.0.2.10", 16, .upstream, "9.9.9.9"),
// A row with no type at all: its own group, never a dropped row.
aggRow(7, "192.0.2.10", null, .upstream, "9.9.9.9"),
aggRow(8, "192.0.2.10", null, .upstream, "9.9.9.9"),
// Outside the window.
aggRow(-1, "192.0.2.10", 255, .upstream, "9.9.9.9"),
});
const rows = (try overview(&database, arena, agg_since, agg_width, agg_buckets)).types;
try testing.expectEqual(@as(usize, 4), rows.len);
try testing.expectEqual(@as(?u16, 1), rows[0].qtype);
try testing.expectEqual(@as(u64, 3), rows[0].count);
try testing.expectEqual(@as(?u16, 16), rows[1].qtype);
try testing.expectEqual(@as(?u16, 28), rows[2].qtype);
try testing.expectEqual(@as(?u16, null), rows[3].qtype);
try testing.expectEqual(@as(u64, 2), rows[3].count);
}
test "an empty window has no type rows at all" {
var database = try openLog();
defer database.close();
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const rows = (try overview(&database, arena_state.allocator(), agg_since, agg_width, agg_buckets)).types;
try testing.expectEqual(@as(usize, 0), rows.len);
}
test "the route breakdown keys on the answering resolver, not on blocklist provenance" {
var database = try openLog();
defer database.close();
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
try seed(&database, &.{
aggRow(0, "192.0.2.10", 1, .upstream, "https://a.example/dns-query"),
aggRow(1, "192.0.2.10", 1, .upstream, "https://a.example/dns-query"),
aggRow(2, "192.0.2.10", 1, .upstream, "https://b.example/dns-query"),
// An upstream row whose resolver the log did not record: its own group.
aggRow(3, "192.0.2.10", 1, .upstream, null),
aggRow(4, "192.0.2.10", 1, .forward_zone, "lan"),
aggRow(5, "192.0.2.10", 1, .blocked, null),
aggRow(6, "192.0.2.10", 1, .cache, null),
aggRow(7, "192.0.2.10", 1, .local, null),
aggRow(8, "192.0.2.10", 1, .rejected, null),
});
const rows = (try overview(&database, arena, agg_since, agg_width, agg_buckets)).routes;
// Two upstreams, one null-source upstream, one forward zone and four
// source-less kinds. Every row carries the same `source_name`, so a
// breakdown that grouped by it would collapse to one row.
try testing.expectEqual(@as(usize, 8), rows.len);
try testing.expectEqual(provenance.RouteKind.upstream, rows[0].route);
try testing.expectEqualStrings("https://a.example/dns-query", rows[0].source.?);
try testing.expectEqual(@as(u64, 2), rows[0].count);
// The seven remaining rows all count 1, so the tie-break orders them:
// route ascending, then source ascending with nulls last.
for (rows[1..]) |row| try testing.expectEqual(@as(u64, 1), row.count);
try testing.expectEqual(provenance.RouteKind.blocked, rows[1].route);
try testing.expectEqual(@as(?[]const u8, null), rows[1].source);
try testing.expectEqual(provenance.RouteKind.cache, rows[2].route);
try testing.expectEqual(provenance.RouteKind.forward_zone, rows[3].route);
try testing.expectEqualStrings("lan", rows[3].source.?);
try testing.expectEqual(provenance.RouteKind.local, rows[4].route);
try testing.expectEqual(provenance.RouteKind.rejected, rows[5].route);
try testing.expectEqual(provenance.RouteKind.upstream, rows[6].route);
try testing.expectEqualStrings("https://b.example/dns-query", rows[6].source.?);
try testing.expectEqual(provenance.RouteKind.upstream, rows[7].route);
try testing.expectEqual(@as(?[]const u8, null), rows[7].source);
}
test "an empty window has no route rows at all" {
var database = try openLog();
defer database.close();
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const rows = (try overview(&database, arena_state.allocator(), agg_since, agg_width, agg_buckets)).routes;
try testing.expectEqual(@as(usize, 0), rows.len);
}
test "an empty window still has a zero-filled other series and no named clients" {
var database = try openLog();
defer database.close();
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const result = (try overview(&database, arena_state.allocator(), agg_since, agg_width, agg_buckets)).clients;
try testing.expectEqual(@as(usize, 0), result.clients.len);
try testing.expectEqual(@as(usize, agg_buckets), result.other.len);
for (result.other) |count| try testing.expectEqual(@as(u64, 0), count);
}
test "client series are bucket-aligned, zero-filled and ranked by in-window total" {
var database = try openLog();
defer database.close();
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
try seed(&database, &.{
aggRow(0, "192.0.2.20", 1, .upstream, "9.9.9.9"),
aggRow(1, "192.0.2.20", 1, .upstream, "9.9.9.9"),
aggRow(agg_width * 3, "192.0.2.20", 1, .upstream, "9.9.9.9"),
aggRow(agg_width * 3, "192.0.2.10", 1, .upstream, "9.9.9.9"),
// Outside the window on both sides.
aggRow(-1, "192.0.2.20", 1, .upstream, "9.9.9.9"),
aggRow(agg_width * agg_buckets, "192.0.2.10", 1, .upstream, "9.9.9.9"),
});
const result = (try overview(&database, arena_state.allocator(), agg_since, agg_width, agg_buckets)).clients;
try testing.expectEqual(@as(usize, 2), result.clients.len);
try testing.expectEqualStrings("192.0.2.20", result.clients[0].client);
try testing.expectEqualStrings("192.0.2.10", result.clients[1].client);
for (result.clients) |series| try testing.expectEqual(@as(usize, agg_buckets), series.buckets.len);
try testing.expectEqual(@as(u64, 2), result.clients[0].buckets[0]);
try testing.expectEqual(@as(u64, 1), result.clients[0].buckets[3]);
try testing.expectEqual(@as(u64, 0), result.clients[0].buckets[9]);
try testing.expectEqual(@as(u64, 1), result.clients[1].buckets[3]);
for (result.other) |count| try testing.expectEqual(@as(u64, 0), count);
}
test "the ninth client folds into other and the cut is the same on every read" {
var database = try openLog();
defer database.close();
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
// Nine clients, each with one more query than the next, so the ranking is
// total and the ninth is unambiguously the one that folds.
var address: [16]u8 = undefined;
var index: u32 = 0;
while (index < 9) : (index += 1) {
const client = try std.fmt.bufPrint(&address, "192.0.2.{d}", .{100 + index});
var repeat: u32 = 0;
while (repeat <= index) : (repeat += 1) {
try seed(&database, &.{aggRow(@intCast(repeat), client, 1, .upstream, "9.9.9.9")});
}
}
const result = (try overview(&database, arena, agg_since, agg_width, agg_buckets)).clients;
try testing.expectEqual(@as(usize, max_client_series), result.clients.len);
// The busiest is 192.0.2.108 with nine rows; the lone folded client is
// 192.0.2.100 with one.
try testing.expectEqualStrings("192.0.2.108", result.clients[0].client);
for (result.clients) |series| {
try testing.expect(!std.mem.eql(u8, "192.0.2.100", series.client));
}
var other_total: u64 = 0;
for (result.other) |count| other_total += count;
try testing.expectEqual(@as(u64, 1), other_total);
}
test "the three breakdowns conserve the window's total" {
var database = try openLog();
defer database.close();
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
// A matrix that exercises every branch the aggregations partition on: null
// and non-null qtypes, every route kind, a null resolver identity, and ten
// clients so the top-eight cut has a residual to carry.
const kinds = [_]provenance.RouteKind{ .blocked, .local, .forward_zone, .upstream, .cache, .rejected };
var address: [16]u8 = undefined;
var index: u32 = 0;
while (index < 40) : (index += 1) {
const client = try std.fmt.bufPrint(&address, "192.0.2.{d}", .{100 + index % 10});
const kind = kinds[index % kinds.len];
try seed(&database, &.{aggRow(
@intCast(index % (agg_width * agg_buckets)),
client,
if (index % 7 == 0) null else @intCast(1 + index % 3),
kind,
if (index % 11 == 0) null else "9.9.9.9",
)});
}
const result = try overview(&database, arena, agg_since, agg_width, agg_buckets);
try testing.expect(result.totals.queries > 0);
var typed: u64 = 0;
for (result.types) |row| typed += row.count;
try testing.expectEqual(result.totals.queries, typed);
var routed: u64 = 0;
for (result.routes) |row| routed += row.count;
try testing.expectEqual(result.totals.queries, routed);
// Per bucket, not just in total: a series misaligned by one bucket would
// still sum correctly over the window.
for (result.buckets, 0..) |bucket, at| {
var summed: u64 = result.clients.other[at];
for (result.clients.clients) |series| summed += series.buckets[at];
try testing.expectEqual(bucket.queries, summed);
}
}
test "the aggregations pass a redacted client through as the log stored it" {
var database = try openLog();
defer database.close();
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
// `hide_client_ips` is applied by the logger before the row is written, so
// the read path has nothing to transform — and must not invent anything
// either. The marker is the client's whole identity here.
try seed(&database, &.{
aggRow(0, logger.hidden_marker, 1, .upstream, "9.9.9.9"),
aggRow(1, logger.hidden_marker, 1, .upstream, "9.9.9.9"),
});
const result = (try overview(&database, arena_state.allocator(), agg_since, agg_width, agg_buckets)).clients;
try testing.expectEqual(@as(usize, 1), result.clients.len);
try testing.expectEqualStrings(logger.hidden_marker, result.clients[0].client);
try testing.expectEqual(@as(u64, 2), result.clients[0].buckets[0]);
}
test "a prune committed mid-read is invisible to the reader's transaction" {
var tmp = testing.tmpDir(.{});
defer tmp.cleanup();
var path_buf: [tmp_prefix.len + sub_path_len + 32]u8 = undefined;
const path = try std.fmt.bufPrintZ(&path_buf, "{s}{s}/querylog.db", .{ tmp_prefix, &tmp.sub_path });
var reader = try db.Db.open(path, .{ .mode = .read_write_create });
defer reader.close();
try db.applyPragmas(&reader, .{});
try reader.exec(querylog_schema.ddl);
// The seeded watermark is the file's creation second, which is now; the
// fixtures below are in 2023, so the prune's cutoff would never advance it.
try reader.exec(
\\UPDATE querylog_meta SET created_at = 1700000000, available_since = 1700000000 WHERE id = 1
);
try seed(&reader, &.{
plainRow(agg_since, "a.example"),
plainRow(agg_since + 1, "b.example"),
plainRow(agg_since + 300, "c.example"),
});
// A second connection to the same file, as retention has in production.
var pruner = try db.Db.open(path, .{ .mode = .read_write_create });
defer pruner.close();
try db.applyPragmas(&pruner, .{});
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
var tx = try db.ReadTx.begin(&reader);
const before = (try overview(&reader, arena, agg_since, 1000, 1)).totals;
const watermark_before = try availableSince(&reader);
try testing.expectEqual(@as(u64, 3), before.queries);
// The prune commits while the reader's transaction is open.
const pruned = try pruneOlderThan(&pruner, agg_since + 200);
try testing.expectEqual(@as(i64, 2), pruned.deleted);
try testing.expect(pruned.available_since > watermark_before);
// Neither half of the answer moved: the rows the reader would report and
// the watermark it would tag them with still describe one state.
const during = (try overview(&reader, arena, agg_since, 1000, 1)).totals;
try testing.expectEqual(before.queries, during.queries);
try testing.expectEqual(watermark_before, try availableSince(&reader));
try tx.commit();
// The next response sees the prune — both halves of it.
const after = (try overview(&reader, arena, agg_since, 1000, 1)).totals;
try testing.expectEqual(@as(u64, 1), after.queries);
try testing.expectEqual(pruned.available_since, try availableSince(&reader));
}
// ---------------------------------------------------------------------------
// the projections and the Overview read path (milestone 36)
// ---------------------------------------------------------------------------
/// Floors a `query_log.timestamp` onto the projection grid in SQL. SQLite's `/`
/// truncates toward zero, so a negative timestamp needs the bias — the same
/// reason `bucketOf` is `@divFloor`.
const bucket_expr = "(CASE WHEN timestamp >= 0 THEN timestamp / 1800 * 1800" ++
" ELSE (timestamp - 1799) / 1800 * 1800 END)";
/// Each projection table beside a from-scratch recomputation of it. Both sides
/// have unique keys, so `EXCEPT` in both directions is an exact equality test
/// rather than a containment one.
const recompute_checks = [_]struct { projection: []const u8, recompute: []const u8 }{
.{
.projection = "SELECT bucket, queries, blocked, cached, rt_sum, rt_count FROM bucket_totals",
.recompute = "SELECT " ++ bucket_expr ++ ", count(*), coalesce(sum(blocked <> 0), 0)," ++
" coalesce(sum(cache_hit = 1), 0), coalesce(sum(response_time_us), 0)," ++
" count(response_time_us) FROM query_log GROUP BY 1",
},
.{
.projection = "SELECT bucket, client_ip, queries FROM bucket_clients",
.recompute = "SELECT " ++ bucket_expr ++ ", client_ip, count(*) FROM query_log GROUP BY 1, 2",
},
.{
.projection = "SELECT bucket, qtype, count FROM bucket_types",
.recompute = "SELECT " ++ bucket_expr ++ ", coalesce(qtype, -1), count(*) FROM query_log GROUP BY 1, 2",
},
.{
.projection = "SELECT bucket, route_kind, source_present, source_text, count FROM bucket_routes",
.recompute = "SELECT " ++ bucket_expr ++ ", route_kind, source IS NOT NULL, coalesce(source, ''), count(*)" ++
" FROM (SELECT timestamp, route_kind, CASE route_kind WHEN 'upstream' THEN upstream" ++
" WHEN 'forward_zone' THEN forward_zone END AS source FROM query_log) GROUP BY 1, 2, 3, 4",
},
};
/// Exported for `querylog_migrations.zig`'s fixture and migration tests: the
/// authority on projection coherence is this file, and a second copy of the
/// recompute SQL there would be free to drift from the writer it checks.
pub fn expectProjectionsMatchRecompute(database: *db.Db) !void {
for (recompute_checks) |check| {
var buf: [4096]u8 = undefined;
const sql = try std.fmt.bufPrint(
&buf,
"SELECT (SELECT count(*) FROM ({s} EXCEPT {s})) + (SELECT count(*) FROM ({s} EXCEPT {s}))",
.{ check.projection, check.recompute, check.recompute, check.projection },
);
try testing.expectEqual(@as(i64, 0), try database.queryInt(sql));
}
}
test "bucketOf floors onto the grid on both sides of the epoch" {
try testing.expectEqual(@as(i64, 0), bucketOf(0));
try testing.expectEqual(@as(i64, 0), bucketOf(1799));
try testing.expectEqual(@as(i64, grain), bucketOf(grain));
try testing.expectEqual(@as(i64, grain), bucketOf(grain + 1));
// Truncation toward zero would name bucket 0 for all three of these, which
// is the bucket *after* the one the row belongs to.
try testing.expectEqual(@as(i64, -grain), bucketOf(-1));
try testing.expectEqual(@as(i64, -grain), bucketOf(-grain));
try testing.expectEqual(@as(i64, -2 * grain), bucketOf(-grain - 1));
}
test "a batch maintains all four projections in the transaction that writes the rows" {
var database = try openLog();
defer database.close();
var writer = try BatchWriter.init(testing.allocator, &database);
defer writer.deinit();
var blocked_row = aggRow(0, "192.0.2.20", null, .blocked, null);
blocked_row.response_time_us = null;
try writer.writeBatch(&.{
aggRow(0, "192.0.2.10", 1, .upstream, "9.9.9.9"),
aggRow(5, "192.0.2.10", 1, .cache, null),
blocked_row,
// The next grid bucket, so the floor is what separates them.
aggRow(grain, "192.0.2.10", 28, .upstream, null),
});
// Two grid buckets: three rows floored into the first, one into the second.
try testing.expectEqual(@as(i64, 2), try database.queryInt("SELECT count(*) FROM bucket_totals"));
try testing.expectEqual(bucketOf(agg_since), try database.queryInt("SELECT min(bucket) FROM bucket_totals"));
try testing.expectEqual(@as(i64, 3), try database.queryInt("SELECT queries FROM bucket_totals ORDER BY bucket LIMIT 1"));
try testing.expectEqual(@as(i64, 1), try database.queryInt("SELECT blocked FROM bucket_totals ORDER BY bucket LIMIT 1"));
try testing.expectEqual(@as(i64, 1), try database.queryInt("SELECT cached FROM bucket_totals ORDER BY bucket LIMIT 1"));
// The blocked row carries no response time, so two of the three are timed.
try testing.expectEqual(@as(i64, 2), try database.queryInt("SELECT rt_count FROM bucket_totals ORDER BY bucket LIMIT 1"));
try expectProjectionsMatchRecompute(&database);
// A second batch accumulates onto the same keys rather than replacing them.
try writer.writeBatch(&.{aggRow(7, "192.0.2.10", 1, .upstream, "9.9.9.9")});
try testing.expectEqual(@as(i64, 4), try database.queryInt("SELECT queries FROM bucket_totals ORDER BY bucket LIMIT 1"));
try testing.expectEqual(
@as(i64, 3),
try database.queryInt("SELECT queries FROM bucket_clients ORDER BY bucket, client_ip LIMIT 1"),
);
try expectProjectionsMatchRecompute(&database);
}
test "a NULL qtype and a NULL route source are stored losslessly" {
var database = try openLog();
defer database.close();
var writer = try BatchWriter.init(testing.allocator, &database);
defer writer.deinit();
try writer.writeBatch(&.{
aggRow(0, "192.0.2.10", null, .upstream, null),
aggRow(1, "192.0.2.10", 1, .upstream, ""),
});
try testing.expectEqual(@as(i64, null_qtype), try database.queryInt("SELECT min(qtype) FROM bucket_types"));
// A row whose resolver the log did not record, and a row whose resolver is
// the empty string, are different keys — which is what `source_present`
// buys over "empty means absent".
try testing.expectEqual(@as(i64, 2), try database.queryInt("SELECT count(*) FROM bucket_routes"));
try testing.expectEqual(
@as(i64, 1),
try database.queryInt("SELECT count(*) FROM bucket_routes WHERE source_present = 0 AND source_text = ''"),
);
try testing.expectEqual(
@as(i64, 1),
try database.queryInt("SELECT count(*) FROM bucket_routes WHERE source_present = 1 AND source_text = ''"),
);
try expectProjectionsMatchRecompute(&database);
}
test "a failed projection update rolls the whole batch back and the writer survives it" {
var database = try openLog();
defer database.close();
var writer = try BatchWriter.init(testing.allocator, &database);
defer writer.deinit();
try writer.writeBatch(&.{aggRow(0, "192.0.2.10", 1, .upstream, "9.9.9.9")});
try database.exec(
\\CREATE TRIGGER refuse_projection BEFORE INSERT ON bucket_clients
\\WHEN new.client_ip = 'boom'
\\BEGIN SELECT RAISE(ABORT, 'refused'); END;
);
try testing.expectError(error.Constraint, writer.writeBatch(&.{
aggRow(1, "boom", 1, .upstream, "9.9.9.9"),
}));
// The raw row went in before the projection statement failed; neither
// survived, because they were one transaction.
try testing.expectEqual(@as(i64, 1), try countRows(&database));
try expectProjectionsMatchRecompute(&database);
try database.exec("DROP TRIGGER refuse_projection;");
try writer.writeBatch(&.{aggRow(2, "192.0.2.11", 1, .upstream, "9.9.9.9")});
try testing.expectEqual(@as(i64, 2), try countRows(&database));
try expectProjectionsMatchRecompute(&database);
}
test "a prune on the grid drops whole projection buckets and leaves the rest" {
var database = try openLog();
defer database.close();
var writer = try BatchWriter.init(testing.allocator, &database);
defer writer.deinit();
const base = bucketOf(agg_since);
try seedAt(&database, &.{ base, base + grain, base + 2 * grain });
_ = try pruneOlderThan(&database, base + 2 * grain);
try testing.expectEqual(@as(i64, 1), try countRows(&database));
try testing.expectEqual(@as(i64, 1), try database.queryInt("SELECT count(*) FROM bucket_totals"));
try testing.expectEqual(base + 2 * grain, try database.queryInt("SELECT bucket FROM bucket_totals"));
try expectProjectionsMatchRecompute(&database);
}
test "a prune off the grid recomputes the straddling bucket instead of estimating it" {
var database = try openLog();
defer database.close();
const base = bucketOf(agg_since);
// Four rows in one bucket; the cutoff falls between the second and third.
try seedAt(&database, &.{ base, base + 100, base + 500, base + 900 });
_ = try pruneOlderThan(&database, base + 400);
try testing.expectEqual(@as(i64, 2), try countRows(&database));
try testing.expectEqual(@as(i64, 1), try database.queryInt("SELECT count(*) FROM bucket_totals"));
try testing.expectEqual(@as(i64, 2), try database.queryInt("SELECT queries FROM bucket_totals"));
try expectProjectionsMatchRecompute(&database);
// Emptying the straddling bucket removes its row rather than writing zeros.
_ = try pruneOlderThan(&database, base + grain - 1);
try testing.expectEqual(@as(i64, 0), try countRows(&database));
try testing.expectEqual(@as(i64, 0), try database.queryInt("SELECT count(*) FROM bucket_totals"));
try expectProjectionsMatchRecompute(&database);
}
test "a failed projection prune rolls back the raw delete and the watermark with it" {
var database = try openLog();
defer database.close();
const base = bucketOf(agg_since);
try seedAt(&database, &.{ base, base + grain });
const watermark = try availableSince(&database);
try database.exec(
\\CREATE TRIGGER refuse_bucket_prune BEFORE DELETE ON bucket_clients
\\BEGIN SELECT RAISE(ABORT, 'refused'); END;
);
try testing.expectError(error.Constraint, pruneOlderThan(&database, base + grain));
try testing.expectEqual(@as(i64, 2), try countRows(&database));
try testing.expectEqual(watermark, try availableSince(&database));
try expectProjectionsMatchRecompute(&database);
try database.exec("DROP TRIGGER refuse_bucket_prune;");
_ = try pruneOlderThan(&database, base + grain);
try testing.expectEqual(@as(i64, 1), try countRows(&database));
try expectProjectionsMatchRecompute(&database);
}
test "a failed straddling-bucket replacement rolls back the raw delete and the watermark with it" {
var database = try openLog();
defer database.close();
const base = bucketOf(agg_since);
// One whole bucket to delete outright, and one the cutoff cuts in half.
try seedAt(&database, &.{ base, base + grain, base + grain + 100, base + grain + 900 });
const watermark = try availableSince(&database);
const straddling_totals = "SELECT queries FROM bucket_totals ORDER BY bucket DESC LIMIT 1";
const straddling_queries = try database.queryInt(straddling_totals);
// Only the recompute inserts into a projection table; the bulk prune above
// it deletes. So this trigger fires during the straddling-bucket
// replacement and nowhere else in the pass.
try database.exec(
\\CREATE TRIGGER refuse_recompute BEFORE INSERT ON bucket_totals
\\BEGIN SELECT RAISE(ABORT, 'refused'); END;
);
try testing.expectError(error.Constraint, pruneOlderThan(&database, base + grain + 500));
// The raw delete, the watermark advance, the whole-bucket projection
// deletes and the straddling bucket's own delete all went back together.
try testing.expectEqual(@as(i64, 4), try countRows(&database));
try testing.expectEqual(watermark, try availableSince(&database));
try testing.expectEqual(@as(i64, 2), try database.queryInt("SELECT count(*) FROM bucket_totals"));
try testing.expectEqual(straddling_queries, try database.queryInt(straddling_totals));
try expectProjectionsMatchRecompute(&database);
try database.exec("DROP TRIGGER refuse_recompute;");
_ = try pruneOlderThan(&database, base + grain + 500);
try testing.expectEqual(@as(i64, 1), try countRows(&database));
try testing.expectEqual(@as(i64, 1), try database.queryInt("SELECT queries FROM bucket_totals"));
try expectProjectionsMatchRecompute(&database);
}
/// One row per timestamp, all in the same batch, with the fixture defaults.
fn seedAt(database: *db.Db, timestamps: []const i64) !void {
var writer = try BatchWriter.init(testing.allocator, database);
defer writer.deinit();
var rows: [16]Row = undefined;
for (rows[0..timestamps.len], timestamps) |*row, ts| {
row.* = plainRow(ts, "example.com");
}
try writer.writeBatch(rows[0..timestamps.len]);
}
test "an arbitrary interleaving of batches and prunes leaves every projection exact" {
var database = try openLog();
defer database.close();
var writer = try BatchWriter.init(testing.allocator, &database);
defer writer.deinit();
var prng: std.Random.DefaultPrng = .init(0x36a5_0000_36a5);
const random = prng.random();
const kinds = [_]provenance.RouteKind{ .blocked, .local, .forward_zone, .upstream, .cache, .rejected };
const sources = [_]?[]const u8{ "9.9.9.9", "https://a.example/dns-query", null, "" };
const base = bucketOf(agg_since);
var rows: [24]Row = undefined;
var addresses: [24][20]u8 = undefined;
var step: usize = 0;
while (step < 60) : (step += 1) {
if (step % 5 == 4) {
// Half the cutoffs land off the grid on purpose: the straddling
// bucket is the only case a delete alone cannot get right.
const offset = random.intRangeAtMost(i64, 0, 6 * grain);
const skew: i64 = if (random.boolean()) 0 else random.intRangeLessThan(i64, 1, grain);
_ = try pruneOlderThan(&database, base + offset + skew);
} else {
const count = random.intRangeAtMost(usize, 1, rows.len);
for (rows[0..count], addresses[0..count]) |*row, *address| {
const client = try std.fmt.bufPrint(address, "192.0.2.{d}", .{random.intRangeAtMost(u8, 1, 5)});
const kind = kinds[random.uintLessThan(usize, kinds.len)];
row.* = plainRow(base + random.intRangeAtMost(i64, 0, 8 * grain), "example.com");
row.client_ip = client;
row.qtype = if (random.boolean()) null else random.intRangeAtMost(u16, 1, 3);
row.blocked = kind == .blocked;
row.cache_hit = if (random.boolean()) null else kind == .cache;
row.response_time_us = if (random.boolean()) null else random.intRangeAtMost(i64, 0, 5000);
row.route_kind = kind;
const source = sources[random.uintLessThan(usize, sources.len)];
row.upstream = if (kind == .upstream) source else null;
row.forward_zone = if (kind == .forward_zone) source else null;
row.policy_action = if (kind == .blocked) .block else .allow;
row.policy_reason = if (kind == .blocked) .blocklist_domain else .no_match;
}
try writer.writeBatch(rows[0..count]);
}
try expectProjectionsMatchRecompute(&database);
}
// The walk has to have exercised both halves of it.
try testing.expect((try countRows(&database)) > 0);
}
// --- the Overview equivalence oracle --------------------------------------
/// A test-only copy of the five SQL aggregates `overview` replaces, kept here
/// so it survives their deletion. It is the independent statement of the
/// contract: if `overview` and this disagree on any window, one of them is
/// wrong, and this one is the one the goldens were written against.
const oracle = struct {
const totals_sql =
\\SELECT count(*),
\\ coalesce(sum(blocked <> 0), 0),
\\ count(DISTINCT client_ip),
\\ coalesce(sum(response_time_us), 0),
\\ count(response_time_us)
\\ FROM query_log
\\ WHERE timestamp >= ?1 AND timestamp < ?2
;
const buckets_sql =
\\SELECT (timestamp - ?1) / ?2,
\\ count(*),
\\ coalesce(sum(blocked <> 0), 0),
\\ coalesce(sum(cache_hit = 1), 0)
\\ FROM query_log
\\ WHERE timestamp >= ?1 AND timestamp < ?3
\\ GROUP BY 1
;
const types_sql =
\\SELECT qtype, count(*)
\\ FROM query_log
\\ WHERE timestamp >= ?1 AND timestamp < ?2
\\ GROUP BY qtype
\\ ORDER BY count(*) DESC, qtype IS NULL, qtype ASC
;
const routes_sql =
\\SELECT route_kind,
\\ CASE route_kind
\\ WHEN 'upstream' THEN upstream
\\ WHEN 'forward_zone' THEN forward_zone
\\ END AS source,
\\ count(*)
\\ FROM query_log
\\ WHERE timestamp >= ?1 AND timestamp < ?2
\\ GROUP BY route_kind, source
\\ ORDER BY count(*) DESC, route_kind ASC, source IS NULL, source ASC
;
const clients_rank_sql =
\\SELECT client_ip
\\ FROM query_log
\\ WHERE timestamp >= ?1 AND timestamp < ?2
\\ GROUP BY client_ip
\\ ORDER BY count(*) DESC, client_ip ASC
\\ LIMIT ?3
;
const clients_buckets_sql =
\\SELECT client_ip, (timestamp - ?1) / ?2, count(*)
\\ FROM query_log
\\ WHERE timestamp >= ?1 AND timestamp < ?3
\\ GROUP BY 1, 2
;
fn build(
database: *db.Db,
arena: Allocator,
since: i64,
bucket_seconds: u32,
bucket_count: u32,
) !Overview {
const width: i64 = bucket_seconds;
const until = since + width * bucket_count;
var totals: StatsTotals = undefined;
{
var stmt = try database.prepare(totals_sql);
defer stmt.deinit();
try stmt.bindInt(1, since);
try stmt.bindInt(2, until);
try testing.expect(try stmt.step());
const timed = stmt.columnInt(4);
totals = .{
.queries = try countOf(stmt.columnInt(0)),
.blocked = try countOf(stmt.columnInt(1)),
.distinct_clients = try countOf(stmt.columnInt(2)),
.avg_response_time_us = if (timed == 0) null else @divTrunc(stmt.columnInt(3), timed),
};
}
const buckets = try arena.alloc(Bucket, bucket_count);
for (buckets, 0..) |*bucket, i| bucket.* = .{
.ts = since + width * @as(i64, @intCast(i)),
.queries = 0,
.blocked = 0,
.cached = 0,
};
{
var stmt = try database.prepare(buckets_sql);
defer stmt.deinit();
try stmt.bindInt(1, since);
try stmt.bindInt(2, width);
try stmt.bindInt(3, until);
while (try stmt.step()) {
const index: usize = @intCast(stmt.columnInt(0));
buckets[index].queries = try countOf(stmt.columnInt(1));
buckets[index].blocked = try countOf(stmt.columnInt(2));
buckets[index].cached = try countOf(stmt.columnInt(3));
}
}
var types: std.ArrayList(TypeCount) = .empty;
{
var stmt = try database.prepare(types_sql);
defer stmt.deinit();
try stmt.bindInt(1, since);
try stmt.bindInt(2, until);
while (try stmt.step()) {
try types.append(arena, .{
.qtype = if (stmt.isNull(0)) null else try columnU16(&stmt, 0),
.count = try countOf(stmt.columnInt(1)),
});
}
}
var routes: std.ArrayList(RouteCount) = .empty;
{
var stmt = try database.prepare(routes_sql);
defer stmt.deinit();
try stmt.bindInt(1, since);
try stmt.bindInt(2, until);
while (try stmt.step()) {
try routes.append(arena, .{
.route = try provenance.parse(provenance.RouteKind, stmt.columnText(0)),
.source = try stmt.columnTextAllocOrNull(arena, 1),
.count = try countOf(stmt.columnInt(2)),
});
}
}
var names: std.ArrayList([]const u8) = .empty;
var series: std.ArrayList([]u64) = .empty;
{
var stmt = try database.prepare(clients_rank_sql);
defer stmt.deinit();
try stmt.bindInt(1, since);
try stmt.bindInt(2, until);
try stmt.bindInt(3, max_client_series);
while (try stmt.step()) {
try names.append(arena, try stmt.columnTextAlloc(arena, 0));
try series.append(arena, try zeroedBuckets(arena, bucket_count));
}
}
const other = try zeroedBuckets(arena, bucket_count);
{
var stmt = try database.prepare(clients_buckets_sql);
defer stmt.deinit();
try stmt.bindInt(1, since);
try stmt.bindInt(2, width);
try stmt.bindInt(3, until);
while (try stmt.step()) {
const index: usize = @intCast(stmt.columnInt(1));
const count = try countOf(stmt.columnInt(2));
const client = stmt.columnText(0);
const target = for (names.items, series.items) |name, target_buckets| {
if (std.mem.eql(u8, name, client)) break target_buckets;
} else other;
target[index] += count;
}
}
const clients = try arena.alloc(ClientSeries, names.items.len);
for (clients, names.items, series.items) |*entry, name, entry_buckets| {
entry.* = .{ .client = name, .buckets = entry_buckets };
}
return .{
.totals = totals,
.buckets = buckets,
.clients = .{ .clients = clients, .other = other },
.types = types.items,
.routes = routes.items,
};
}
};
fn expectOverviewEqual(expected: Overview, actual: Overview) !void {
try testing.expectEqual(expected.totals.queries, actual.totals.queries);
try testing.expectEqual(expected.totals.blocked, actual.totals.blocked);
try testing.expectEqual(expected.totals.distinct_clients, actual.totals.distinct_clients);
try testing.expectEqual(expected.totals.avg_response_time_us, actual.totals.avg_response_time_us);
try testing.expectEqualSlices(Bucket, expected.buckets, actual.buckets);
try testing.expectEqual(expected.types.len, actual.types.len);
for (expected.types, actual.types) |want, got| {
try testing.expectEqual(want.qtype, got.qtype);
try testing.expectEqual(want.count, got.count);
}
try testing.expectEqual(expected.routes.len, actual.routes.len);
for (expected.routes, actual.routes) |want, got| {
try testing.expectEqual(want.route, got.route);
try testing.expectEqual(want.count, got.count);
if (want.source) |source| {
try testing.expectEqualStrings(source, got.source orelse return error.TestExpectedEqual);
} else {
try testing.expectEqual(@as(?[]const u8, null), got.source);
}
}
try testing.expectEqual(expected.clients.clients.len, actual.clients.clients.len);
for (expected.clients.clients, actual.clients.clients) |want, got| {
try testing.expectEqualStrings(want.client, got.client);
try testing.expectEqualSlices(u64, want.buckets, got.buckets);
}
try testing.expectEqualSlices(u64, expected.clients.other, actual.clients.other);
}
/// Every window the equivalence test walks. The first is the raw path (60 s
/// buckets, the 1h period); the rest are the projection path at each serving
/// width the period grammar offers.
const overview_windows = [_]struct { bucket_seconds: u32, bucket_count: u32 }{
.{ .bucket_seconds = 60, .bucket_count = 60 },
.{ .bucket_seconds = 1800, .bucket_count = 48 },
.{ .bucket_seconds = 3600, .bucket_count = 168 },
.{ .bucket_seconds = 21600, .bucket_count = 120 },
};
fn expectOverviewMatchesOracle(database: *db.Db, since: i64) !void {
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
for (overview_windows) |window| {
const want = try oracle.build(database, arena, since, window.bucket_seconds, window.bucket_count);
const got = try overview(database, arena, since, window.bucket_seconds, window.bucket_count);
expectOverviewEqual(want, got) catch |err| {
std.debug.print("overview mismatch at bucket_seconds={d}\n", .{window.bucket_seconds});
return err;
};
}
}
test "overview equals the oracle on an empty database" {
var database = try openLog();
defer database.close();
try expectOverviewMatchesOracle(&database, bucketOf(agg_since));
}
test "overview equals the oracle over a window with no rows in it" {
var database = try openLog();
defer database.close();
try seedAt(&database, &.{ agg_since, agg_since + 10 });
// Two full 30-day spans later: every panel is empty and the axis is still
// whole.
try expectOverviewMatchesOracle(&database, bucketOf(agg_since) + 240 * grain);
}
test "overview equals the oracle across nulls, ties and a bucket still in progress" {
var database = try openLog();
defer database.close();
var writer = try BatchWriter.init(testing.allocator, &database);
defer writer.deinit();
const since = bucketOf(agg_since);
const kinds = [_]provenance.RouteKind{ .blocked, .local, .forward_zone, .upstream, .cache, .rejected };
// Ten clients so the top-eight cut has a residual, and the counts are
// deliberately equal in pairs so the `client_ip` tie-break decides.
var rows: [120]Row = undefined;
var addresses: [120][20]u8 = undefined;
for (&rows, &addresses, 0..) |*row, *address, i| {
const client = try std.fmt.bufPrint(address, "192.0.2.{d}", .{100 + (i / 2) % 10});
const kind = kinds[i % kinds.len];
row.* = plainRow(since + @as(i64, @intCast(i)) * 137, "example.com");
row.client_ip = client;
// Every seventh row carries no query type, and the types collide in
// count so the null-last tie-break is exercised.
row.qtype = if (i % 7 == 0) null else @intCast(1 + i % 3);
row.blocked = kind == .blocked;
row.cache_hit = if (i % 5 == 0) null else kind == .cache;
row.response_time_us = if (i % 4 == 0) null else @intCast(100 + i);
row.route_kind = kind;
// Every eleventh upstream row records no resolver: the NULL-source
// group the routes breakdown must keep rather than drop.
const source: ?[]const u8 = if (i % 11 == 0) null else if (i % 2 == 0) "9.9.9.9" else "https://a.example/dns-query";
row.upstream = if (kind == .upstream) source else null;
row.forward_zone = if (kind == .forward_zone) source else null;
row.policy_action = if (kind == .blocked) .block else .allow;
row.policy_reason = if (kind == .blocked) .blocklist_domain else .no_match;
}
try writer.writeBatch(&rows);
try expectOverviewMatchesOracle(&database, since);
// A window whose last bucket is only partly filled: `since` moved so the
// newest rows land inside the final bucket rather than closing it.
try expectOverviewMatchesOracle(&database, since - 40 * grain);
// And after a prune off the grid, so the projections the read path uses are
// the recomputed ones.
_ = try pruneOlderThan(&database, since + 900);
try expectProjectionsMatchRecompute(&database);
try expectOverviewMatchesOracle(&database, since);
}
test "overview equals the oracle after an interleaving of batches and prunes" {
var database = try openLog();
defer database.close();
var writer = try BatchWriter.init(testing.allocator, &database);
defer writer.deinit();
const since = bucketOf(agg_since);
var prng: std.Random.DefaultPrng = .init(0x0e17_3a99);
const random = prng.random();
var rows: [16]Row = undefined;
var addresses: [16][20]u8 = undefined;
var step: usize = 0;
while (step < 12) : (step += 1) {
const count = random.intRangeAtMost(usize, 1, rows.len);
for (rows[0..count], addresses[0..count]) |*row, *address| {
const client = try std.fmt.bufPrint(address, "10.0.0.{d}", .{random.intRangeAtMost(u8, 1, 12)});
row.* = plainRow(since + random.intRangeAtMost(i64, 0, 20 * grain), "example.com");
row.client_ip = client;
row.qtype = if (random.boolean()) null else random.intRangeAtMost(u16, 1, 4);
row.response_time_us = if (random.boolean()) null else random.intRangeAtMost(i64, 0, 9000);
row.cache_hit = if (random.boolean()) null else random.boolean();
}
try writer.writeBatch(rows[0..count]);
if (step % 4 == 3) {
_ = try pruneOlderThan(&database, since + random.intRangeAtMost(i64, 0, 10 * grain));
}
try expectOverviewMatchesOracle(&database, since);
}
}
test "overview refuses a window the projections cannot express" {
var database = try openLog();
defer database.close();
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
const since = bucketOf(agg_since);
try testing.expectError(error.Misuse, overview(&database, arena, since, 0, 48));
try testing.expectError(error.Misuse, overview(&database, arena, since, 1800, 0));
try testing.expectError(error.Misuse, overview(&database, arena, std.math.maxInt(i64) - 1, 3600, 48));
// On the projection path the grid is part of the contract: an unaligned
// start or a width that is not a whole number of grains cannot be answered
// from 30-minute rows, and guessing would be worse than refusing.
try testing.expectError(error.Misuse, overview(&database, arena, since + 1, 1800, 48));
try testing.expectError(error.Misuse, overview(&database, arena, since, 2700, 48));
// Below the grain none of that applies: the raw rows carry every second.
_ = try overview(&database, arena, since + 1, 60, 60);
}