milestone 6: dns cache, rate limiting, query logging, disk monitoring and retention
This commit is contained in:
@@ -0,0 +1,420 @@
|
||||
//! Free-space monitor (PLAN §11.6). Samples the filesystem holding the data
|
||||
//! directory every 60 seconds, classifies the result against the configured
|
||||
//! thresholds, and publishes the state plus three size gauges through atomics.
|
||||
//!
|
||||
//! The state is the gate other components read before a non-essential write:
|
||||
//! the query logger holds its batches while `writesAllowed` is false, and
|
||||
//! Phase 7 gates blocklist updates the same way. Nothing here edits a
|
||||
//! milestone-5 file; the gate is pulled, not pushed.
|
||||
|
||||
const std = @import("std");
|
||||
const model = @import("../config/model.zig");
|
||||
const statfs = @import("../platform/statfs.zig");
|
||||
|
||||
const log = std.log.scoped(.disk_monitor);
|
||||
|
||||
pub const sample_interval_s = 60;
|
||||
|
||||
pub const State = enum(u8) { ok, warn, critical };
|
||||
|
||||
pub const Gauges = struct {
|
||||
free_bytes: u64,
|
||||
db_bytes: u64,
|
||||
log_bytes: u64,
|
||||
};
|
||||
|
||||
/// `min_free_mb` is checked first, so a configuration whose warn threshold sits
|
||||
/// below its critical threshold still reports the more severe of the two.
|
||||
pub fn classify(free_bytes: u64, cfg: model.Disk) State {
|
||||
if (free_bytes < model.minFreeBytes(cfg)) return .critical;
|
||||
if (free_bytes < model.warnFreeBytes(cfg)) return .warn;
|
||||
return .ok;
|
||||
}
|
||||
|
||||
pub const Monitor = struct {
|
||||
cfg: model.Disk,
|
||||
data_dir: std.Io.Dir,
|
||||
data_path: [:0]const u8,
|
||||
log_dir_path: ?[:0]const u8,
|
||||
|
||||
state_raw: std.atomic.Value(u8),
|
||||
free_bytes: std.atomic.Value(u64),
|
||||
db_bytes: std.atomic.Value(u64),
|
||||
log_bytes: std.atomic.Value(u64),
|
||||
sample_failures: std.atomic.Value(u64),
|
||||
|
||||
/// `data_dir` must be open with `.iterate = true`; sizing the databases
|
||||
/// scans it. `data_path` names the filesystem to measure and `data_dir` the
|
||||
/// directory to size — normally the same place, but `statvfs` takes a path
|
||||
/// and the scan takes a handle. `log_dir_path` resolves against the process
|
||||
/// working directory and is null when logs do not go to a file.
|
||||
pub fn init(
|
||||
cfg: model.Disk,
|
||||
data_dir: std.Io.Dir,
|
||||
data_path: [:0]const u8,
|
||||
log_dir_path: ?[:0]const u8,
|
||||
) Monitor {
|
||||
return .{
|
||||
.cfg = cfg,
|
||||
.data_dir = data_dir,
|
||||
.data_path = data_path,
|
||||
.log_dir_path = log_dir_path,
|
||||
.state_raw = .init(@intFromEnum(State.ok)),
|
||||
.free_bytes = .init(0),
|
||||
.db_bytes = .init(0),
|
||||
.log_bytes = .init(0),
|
||||
.sample_failures = .init(0),
|
||||
};
|
||||
}
|
||||
|
||||
pub fn state(self: *const Monitor) State {
|
||||
return @enumFromInt(self.state_raw.load(.monotonic));
|
||||
}
|
||||
|
||||
pub fn writesAllowed(self: *const Monitor) bool {
|
||||
return self.state() != .critical;
|
||||
}
|
||||
|
||||
pub fn gauges(self: *const Monitor) Gauges {
|
||||
return .{
|
||||
.free_bytes = self.free_bytes.load(.monotonic),
|
||||
.db_bytes = self.db_bytes.load(.monotonic),
|
||||
.log_bytes = self.log_bytes.load(.monotonic),
|
||||
};
|
||||
}
|
||||
|
||||
/// One pass: free space from `statvfs`, then the two size gauges. A failed
|
||||
/// `statvfs` leaves the state untouched — an unreadable filesystem is not
|
||||
/// evidence that the disk filled — and a failed size scan leaves that one
|
||||
/// gauge at its previous reading. Every failure increments
|
||||
/// `sample_failures` and logs one line at `warn`.
|
||||
pub fn sample(self: *Monitor, io: std.Io) void {
|
||||
const free = statfs.freeBytes(self.data_path) catch {
|
||||
self.countFailure();
|
||||
log.warn("statvfs on {s} failed", .{self.data_path});
|
||||
return;
|
||||
};
|
||||
self.free_bytes.store(free, .monotonic);
|
||||
|
||||
if (sumDir(io, self.data_dir, isDatabaseFile)) |bytes| {
|
||||
self.db_bytes.store(bytes, .monotonic);
|
||||
} else |err| {
|
||||
self.countFailure();
|
||||
log.warn("sizing the data directory failed: {s}", .{@errorName(err)});
|
||||
}
|
||||
|
||||
if (self.log_dir_path) |path| {
|
||||
if (self.sumLogDir(io, path)) |bytes| {
|
||||
self.log_bytes.store(bytes, .monotonic);
|
||||
} else |err| {
|
||||
self.countFailure();
|
||||
log.warn("sizing {s} failed: {s}", .{ path, @errorName(err) });
|
||||
}
|
||||
}
|
||||
|
||||
self.publish(classify(free, self.cfg), free);
|
||||
}
|
||||
|
||||
/// Sample first, then sleep: a process that starts on a full disk must not
|
||||
/// serve a whole interval believing the state is `.ok`. `.boot` so a
|
||||
/// suspended box still sees the interval elapse.
|
||||
pub fn run(self: *Monitor, io: std.Io) std.Io.Cancelable!void {
|
||||
const interval: std.Io.Clock.Duration = .{
|
||||
.raw = .fromSeconds(sample_interval_s),
|
||||
.clock = .boot,
|
||||
};
|
||||
while (true) {
|
||||
self.sample(io);
|
||||
try interval.sleep(io);
|
||||
}
|
||||
}
|
||||
|
||||
fn countFailure(self: *Monitor) void {
|
||||
_ = self.sample_failures.fetchAdd(1, .monotonic);
|
||||
}
|
||||
|
||||
/// Logs on transitions only. A disk that sits at `.warn` for a week
|
||||
/// produces one line, not ten thousand.
|
||||
fn publish(self: *Monitor, next: State, free: u64) void {
|
||||
const previous: State = @enumFromInt(self.state_raw.swap(@intFromEnum(next), .monotonic));
|
||||
if (previous == next) return;
|
||||
log.warn("disk state {t} -> {t}: {d} bytes free on {s}", .{
|
||||
previous,
|
||||
next,
|
||||
free,
|
||||
self.data_path,
|
||||
});
|
||||
}
|
||||
|
||||
fn sumLogDir(self: *Monitor, io: std.Io, path: [:0]const u8) !u64 {
|
||||
_ = self;
|
||||
var dir = try std.Io.Dir.cwd().openDir(io, path, .{ .iterate = true });
|
||||
defer dir.close(io);
|
||||
return sumDir(io, dir, everyFile);
|
||||
}
|
||||
};
|
||||
|
||||
fn everyFile(_: []const u8) bool {
|
||||
return true;
|
||||
}
|
||||
|
||||
/// The sqlite trio: `x.db`, its write-ahead log and its shared-memory index.
|
||||
/// All three live on the watched filesystem and all three grow.
|
||||
fn isDatabaseFile(name: []const u8) bool {
|
||||
return std.mem.endsWith(u8, name, ".db") or
|
||||
std.mem.endsWith(u8, name, ".db-wal") or
|
||||
std.mem.endsWith(u8, name, ".db-shm");
|
||||
}
|
||||
|
||||
fn sumDir(io: std.Io, dir: std.Io.Dir, accept: *const fn ([]const u8) bool) !u64 {
|
||||
var total: u64 = 0;
|
||||
var it = dir.iterate();
|
||||
while (try it.next(io)) |entry| {
|
||||
if (entry.kind != .file) continue;
|
||||
if (!accept(entry.name)) continue;
|
||||
// A file that vanishes between `iterate` and `statFile` is normal:
|
||||
// rotation and database recreate both delete under a running scan. Any
|
||||
// other stat failure makes the whole scan fail, because a partial total
|
||||
// published as a gauge reads as a shrinking database.
|
||||
const st = dir.statFile(io, entry.name, .{}) catch |err| switch (err) {
|
||||
error.FileNotFound => continue,
|
||||
else => return err,
|
||||
};
|
||||
total += st.size;
|
||||
}
|
||||
return total;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// tests
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
const testing = std.testing;
|
||||
|
||||
const mb = 1024 * 1024;
|
||||
|
||||
/// Set by the vanished-file test only: `acceptGoneDeleted` needs a handle and an
|
||||
/// io, and the `accept` signature carries neither.
|
||||
var vanish_dir: ?std.Io.Dir = null;
|
||||
var vanish_io: ?std.Io = null;
|
||||
|
||||
/// Deletes `gone.db` between `iterate` and `statFile`, which is the race the
|
||||
/// scan must tolerate.
|
||||
fn acceptGoneDeleted(name: []const u8) bool {
|
||||
if (!isDatabaseFile(name)) return false;
|
||||
if (std.mem.eql(u8, name, "gone.db")) {
|
||||
vanish_dir.?.deleteFile(vanish_io.?, name) catch {};
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
test "classify below the critical threshold" {
|
||||
const cfg: model.Disk = .{ .min_free_mb = 200, .warn_free_mb = 500 };
|
||||
try testing.expectEqual(State.critical, classify(0, cfg));
|
||||
try testing.expectEqual(State.critical, classify(199 * mb, cfg));
|
||||
try testing.expectEqual(State.critical, classify(200 * mb - 1, cfg));
|
||||
}
|
||||
|
||||
test "classify at and above the critical threshold" {
|
||||
const cfg: model.Disk = .{ .min_free_mb = 200, .warn_free_mb = 500 };
|
||||
try testing.expectEqual(State.warn, classify(200 * mb, cfg));
|
||||
try testing.expectEqual(State.warn, classify(350 * mb, cfg));
|
||||
try testing.expectEqual(State.warn, classify(500 * mb - 1, cfg));
|
||||
}
|
||||
|
||||
test "classify at and above the warn threshold" {
|
||||
const cfg: model.Disk = .{ .min_free_mb = 200, .warn_free_mb = 500 };
|
||||
try testing.expectEqual(State.ok, classify(500 * mb, cfg));
|
||||
try testing.expectEqual(State.ok, classify(64 * 1024 * mb, cfg));
|
||||
}
|
||||
|
||||
test "classify with both thresholds at zero never leaves ok" {
|
||||
const cfg: model.Disk = .{ .min_free_mb = 0, .warn_free_mb = 0 };
|
||||
try testing.expectEqual(State.ok, classify(0, cfg));
|
||||
try testing.expectEqual(State.ok, classify(1, cfg));
|
||||
}
|
||||
|
||||
test "classify reports the more severe state when warn sits below min" {
|
||||
const cfg: model.Disk = .{ .min_free_mb = 500, .warn_free_mb = 200 };
|
||||
try testing.expectEqual(State.critical, classify(300 * mb, cfg));
|
||||
try testing.expectEqual(State.ok, classify(500 * mb, cfg));
|
||||
}
|
||||
|
||||
test "the database file filter accepts the sqlite trio only" {
|
||||
try testing.expect(isDatabaseFile("querylog.db"));
|
||||
try testing.expect(isDatabaseFile("querylog.db-wal"));
|
||||
try testing.expect(isDatabaseFile("querylog.db-shm"));
|
||||
try testing.expect(!isDatabaseFile("querylog.db.bak"));
|
||||
try testing.expect(!isDatabaseFile("nxdns.log"));
|
||||
try testing.expect(!isDatabaseFile(""));
|
||||
}
|
||||
|
||||
test "init reports ok with zero gauges and allows writes" {
|
||||
var monitor: Monitor = .init(.{}, std.Io.Dir.cwd(), ".", null);
|
||||
try testing.expectEqual(State.ok, monitor.state());
|
||||
try testing.expect(monitor.writesAllowed());
|
||||
try testing.expectEqual(Gauges{ .free_bytes = 0, .db_bytes = 0, .log_bytes = 0 }, monitor.gauges());
|
||||
try testing.expectEqual(@as(u64, 0), monitor.sample_failures.load(.monotonic));
|
||||
}
|
||||
|
||||
test "writesAllowed is false only at critical" {
|
||||
var monitor: Monitor = .init(.{}, std.Io.Dir.cwd(), ".", null);
|
||||
monitor.state_raw.store(@intFromEnum(State.warn), .monotonic);
|
||||
try testing.expect(monitor.writesAllowed());
|
||||
monitor.state_raw.store(@intFromEnum(State.critical), .monotonic);
|
||||
try testing.expect(!monitor.writesAllowed());
|
||||
}
|
||||
|
||||
test "a sample sizes the databases and ignores every other file" {
|
||||
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
|
||||
var tmp = testing.tmpDir(.{ .iterate = true });
|
||||
defer tmp.cleanup();
|
||||
|
||||
try tmp.dir.writeFile(io, .{ .sub_path = "querylog.db", .data = &[_]u8{'a'} ** 100 });
|
||||
try tmp.dir.writeFile(io, .{ .sub_path = "querylog.db-wal", .data = &[_]u8{'b'} ** 50 });
|
||||
try tmp.dir.writeFile(io, .{ .sub_path = "querylog.db-shm", .data = &[_]u8{'c'} ** 10 });
|
||||
try tmp.dir.writeFile(io, .{ .sub_path = "notes.txt", .data = &[_]u8{'d'} ** 4096 });
|
||||
|
||||
var monitor: Monitor = .init(.{ .min_free_mb = 0, .warn_free_mb = 0 }, tmp.dir, ".", null);
|
||||
monitor.sample(io);
|
||||
|
||||
const g = monitor.gauges();
|
||||
try testing.expectEqual(@as(u64, 160), g.db_bytes);
|
||||
try testing.expectEqual(@as(u64, 0), g.log_bytes);
|
||||
try testing.expect(g.free_bytes > 0);
|
||||
try testing.expectEqual(@as(u64, 0), monitor.sample_failures.load(.monotonic));
|
||||
try testing.expectEqual(State.ok, monitor.state());
|
||||
}
|
||||
|
||||
test "a sample sizes every file in the log directory" {
|
||||
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
|
||||
var tmp = testing.tmpDir(.{ .iterate = true });
|
||||
defer tmp.cleanup();
|
||||
try tmp.dir.createDirPath(io, "logs");
|
||||
|
||||
var logs = try tmp.dir.openDir(io, "logs", .{});
|
||||
defer logs.close(io);
|
||||
try logs.writeFile(io, .{ .sub_path = "nxdns.log", .data = &[_]u8{'a'} ** 300 });
|
||||
try logs.writeFile(io, .{ .sub_path = "nxdns.log.1", .data = &[_]u8{'b'} ** 700 });
|
||||
|
||||
var path_buf: [256]u8 = undefined;
|
||||
const log_path = try std.fmt.bufPrintZ(&path_buf, ".zig-cache/tmp/{s}/logs", .{tmp.sub_path});
|
||||
|
||||
var monitor: Monitor = .init(.{ .min_free_mb = 0, .warn_free_mb = 0 }, tmp.dir, ".", log_path);
|
||||
monitor.sample(io);
|
||||
|
||||
try testing.expectEqual(@as(u64, 1000), monitor.gauges().log_bytes);
|
||||
try testing.expectEqual(@as(u64, 0), monitor.sample_failures.load(.monotonic));
|
||||
}
|
||||
|
||||
test "a failed statvfs counts and keeps the previous state" {
|
||||
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
|
||||
var monitor: Monitor = .init(.{}, std.Io.Dir.cwd(), "./nxdns-no-such-path-7c21", null);
|
||||
monitor.state_raw.store(@intFromEnum(State.warn), .monotonic);
|
||||
monitor.sample(io);
|
||||
|
||||
try testing.expectEqual(State.warn, monitor.state());
|
||||
try testing.expectEqual(@as(u64, 1), monitor.sample_failures.load(.monotonic));
|
||||
try testing.expectEqual(@as(u64, 0), monitor.gauges().free_bytes);
|
||||
}
|
||||
|
||||
test "an unreadable log directory counts a failure but still publishes a state" {
|
||||
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
|
||||
var tmp = testing.tmpDir(.{ .iterate = true });
|
||||
defer tmp.cleanup();
|
||||
|
||||
var monitor: Monitor = .init(
|
||||
.{ .min_free_mb = 0, .warn_free_mb = 0 },
|
||||
tmp.dir,
|
||||
".",
|
||||
"./nxdns-no-such-dir-4f8a",
|
||||
);
|
||||
monitor.sample(io);
|
||||
|
||||
try testing.expectEqual(@as(u64, 1), monitor.sample_failures.load(.monotonic));
|
||||
try testing.expectEqual(State.ok, monitor.state());
|
||||
try testing.expect(monitor.gauges().free_bytes > 0);
|
||||
}
|
||||
|
||||
test "a file deleted during the scan is skipped and the rest still counts" {
|
||||
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
|
||||
var tmp = testing.tmpDir(.{ .iterate = true });
|
||||
defer tmp.cleanup();
|
||||
|
||||
try tmp.dir.writeFile(io, .{ .sub_path = "gone.db", .data = &[_]u8{'a'} ** 100 });
|
||||
try tmp.dir.writeFile(io, .{ .sub_path = "stays.db", .data = &[_]u8{'b'} ** 40 });
|
||||
|
||||
vanish_dir = tmp.dir;
|
||||
vanish_io = io;
|
||||
defer vanish_dir = null;
|
||||
|
||||
try testing.expectEqual(@as(u64, 40), try sumDir(io, tmp.dir, acceptGoneDeleted));
|
||||
}
|
||||
|
||||
test "an unreadable data directory fails the scan and keeps the previous gauge" {
|
||||
// Mode bits do not apply to root, so the denial the test needs cannot happen.
|
||||
if (std.c.geteuid() == 0) return error.SkipZigTest;
|
||||
|
||||
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
|
||||
var tmp = testing.tmpDir(.{ .iterate = true });
|
||||
defer tmp.cleanup();
|
||||
|
||||
try tmp.dir.writeFile(io, .{ .sub_path = "querylog.db", .data = &[_]u8{'a'} ** 100 });
|
||||
|
||||
var monitor: Monitor = .init(.{ .min_free_mb = 0, .warn_free_mb = 0 }, tmp.dir, ".", null);
|
||||
monitor.db_bytes.store(4096, .monotonic);
|
||||
|
||||
// The handle keeps its read permission from open time, so `iterate` still
|
||||
// lists the file, but path resolution under the directory now fails.
|
||||
try tmp.dir.setPermissions(io, .fromMode(0o600));
|
||||
monitor.sample(io);
|
||||
try tmp.dir.setPermissions(io, .fromMode(0o700));
|
||||
|
||||
try testing.expectEqual(@as(u64, 4096), monitor.gauges().db_bytes);
|
||||
try testing.expectEqual(@as(u64, 1), monitor.sample_failures.load(.monotonic));
|
||||
try testing.expectEqual(State.ok, monitor.state());
|
||||
}
|
||||
|
||||
test "a threshold above the real free space drives the state to critical" {
|
||||
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
|
||||
var tmp = testing.tmpDir(.{ .iterate = true });
|
||||
defer tmp.cleanup();
|
||||
|
||||
const unreachable_mb = std.math.maxInt(u32);
|
||||
var monitor: Monitor = .init(
|
||||
.{ .min_free_mb = unreachable_mb, .warn_free_mb = unreachable_mb },
|
||||
tmp.dir,
|
||||
".",
|
||||
null,
|
||||
);
|
||||
monitor.sample(io);
|
||||
try testing.expectEqual(State.critical, monitor.state());
|
||||
try testing.expect(!monitor.writesAllowed());
|
||||
|
||||
monitor.cfg = .{ .min_free_mb = 0, .warn_free_mb = 0 };
|
||||
monitor.sample(io);
|
||||
try testing.expectEqual(State.ok, monitor.state());
|
||||
try testing.expect(monitor.writesAllowed());
|
||||
try testing.expectEqual(@as(u64, 0), monitor.sample_failures.load(.monotonic));
|
||||
}
|
||||
@@ -0,0 +1,833 @@
|
||||
//! Async query logger (PLAN §11.4). The query path hands an `Entry` to `log`
|
||||
//! and never touches the database: one writer task owns the `db.Db` handle, and
|
||||
//! everything between the two is an `std.Io.Queue`.
|
||||
//!
|
||||
//! `Io.Queue` copies elements as raw bytes (`Io.zig:2189`), so an `Entry` owns
|
||||
//! every byte it carries — a slice into the caller's packet buffer would dangle
|
||||
//! the moment the query finishes. That is the whole reason this file has fixed
|
||||
//! buffers instead of slices.
|
||||
//!
|
||||
//! The privacy transforms of §11.4 run inside `log`, before the entry is
|
||||
//! enqueued, so nothing downstream — the database now, Phase 8's event stream
|
||||
//! later — can observe a value the operator asked to hide.
|
||||
//!
|
||||
//! Log rows are expendable. A full queue drops the oldest unflushed entry, a
|
||||
//! failed batch is dropped whole, and a disk that crossed the critical
|
||||
//! threshold holds batches back indefinitely. Each of the three has a counter.
|
||||
//! A writer that cannot prepare its statements closes the queue and marks
|
||||
//! `writer_failed`, so the loss is visible rather than silent.
|
||||
|
||||
const std = @import("std");
|
||||
|
||||
const db = @import("db.zig");
|
||||
const disk_monitor = @import("disk_monitor.zig");
|
||||
const model = @import("../config/model.zig");
|
||||
const queries_repo = @import("repositories/queries_repo.zig");
|
||||
|
||||
/// Named `scope` rather than `log`: `Logger.log` is the enqueue entry point,
|
||||
/// and the two names collide inside the struct.
|
||||
const scope = std.log.scoped(.query_logger);
|
||||
|
||||
/// Flush tuning is comptime: §12.1 defines no configuration keys for it and a
|
||||
/// household deployment has no reason to tune it.
|
||||
pub const flush_batch = 100;
|
||||
pub const flush_interval_ms = 100;
|
||||
|
||||
/// What `hide_domains` and `hide_client_ips` store instead of the real value.
|
||||
pub const hidden_marker = "hidden";
|
||||
|
||||
/// How long a batch waits before it re-reads the disk monitor.
|
||||
pub const gate_retry_s = 1;
|
||||
|
||||
const max_domain_len = 253;
|
||||
/// RFC 5952 text of any IPv6 address, zone identifier included.
|
||||
const max_client_len = 45;
|
||||
const max_reason_len = 32;
|
||||
const max_upstream_len = 64;
|
||||
|
||||
/// One row on its way to `query_log`, carrying its own bytes.
|
||||
pub const Entry = struct {
|
||||
timestamp: i64,
|
||||
domain_buf: [max_domain_len]u8,
|
||||
domain_len: u8,
|
||||
client_buf: [max_client_len]u8,
|
||||
client_len: u8,
|
||||
qtype: ?u16,
|
||||
blocked: bool,
|
||||
reason_buf: [max_reason_len]u8,
|
||||
reason_len: u8,
|
||||
response_time_us: ?i64,
|
||||
cache_hit: ?bool,
|
||||
upstream_buf: [max_upstream_len]u8,
|
||||
upstream_len: u8,
|
||||
|
||||
/// The borrowed shape of an entry. `init` copies out of it, so a caller can
|
||||
/// build one from slices that die with the query.
|
||||
pub const Fields = struct {
|
||||
timestamp: i64,
|
||||
domain: []const u8,
|
||||
client_ip: []const u8,
|
||||
qtype: ?u16 = null,
|
||||
blocked: bool = false,
|
||||
/// Empty means "no reason", which reaches the database as NULL.
|
||||
block_reason: []const u8 = "",
|
||||
response_time_us: ?i64 = null,
|
||||
cache_hit: ?bool = null,
|
||||
/// Empty means "no upstream", which reaches the database as NULL.
|
||||
upstream: []const u8 = "",
|
||||
};
|
||||
|
||||
/// Copies each string in, truncated to what its buffer holds. A name longer
|
||||
/// than 253 bytes is not a valid domain name, so truncation here means the
|
||||
/// caller skipped the parser, not that a real name was lost.
|
||||
pub fn init(f: Fields) Entry {
|
||||
var entry: Entry = .{
|
||||
.timestamp = f.timestamp,
|
||||
.domain_buf = undefined,
|
||||
.domain_len = 0,
|
||||
.client_buf = undefined,
|
||||
.client_len = 0,
|
||||
.qtype = f.qtype,
|
||||
.blocked = f.blocked,
|
||||
.reason_buf = undefined,
|
||||
.reason_len = 0,
|
||||
.response_time_us = f.response_time_us,
|
||||
.cache_hit = f.cache_hit,
|
||||
.upstream_buf = undefined,
|
||||
.upstream_len = 0,
|
||||
};
|
||||
entry.setDomain(f.domain);
|
||||
entry.setClientIp(f.client_ip);
|
||||
entry.reason_len = copyInto(&entry.reason_buf, f.block_reason);
|
||||
entry.upstream_len = copyInto(&entry.upstream_buf, f.upstream);
|
||||
return entry;
|
||||
}
|
||||
|
||||
pub fn setDomain(self: *Entry, value: []const u8) void {
|
||||
self.domain_len = copyInto(&self.domain_buf, value);
|
||||
}
|
||||
|
||||
pub fn setClientIp(self: *Entry, value: []const u8) void {
|
||||
self.client_len = copyInto(&self.client_buf, value);
|
||||
}
|
||||
|
||||
pub fn domain(self: *const Entry) []const u8 {
|
||||
return self.domain_buf[0..self.domain_len];
|
||||
}
|
||||
|
||||
pub fn clientIp(self: *const Entry) []const u8 {
|
||||
return self.client_buf[0..self.client_len];
|
||||
}
|
||||
|
||||
pub fn blockReason(self: *const Entry) []const u8 {
|
||||
return self.reason_buf[0..self.reason_len];
|
||||
}
|
||||
|
||||
pub fn upstream(self: *const Entry) []const u8 {
|
||||
return self.upstream_buf[0..self.upstream_len];
|
||||
}
|
||||
};
|
||||
|
||||
fn copyInto(buf: []u8, value: []const u8) u8 {
|
||||
const n = @min(buf.len, value.len);
|
||||
@memcpy(buf[0..n], value[0..n]);
|
||||
return @intCast(n);
|
||||
}
|
||||
|
||||
/// The row borrows from `entry`, which must outlive the `writeBatch` call.
|
||||
fn toRow(entry: *const Entry) queries_repo.Row {
|
||||
return .{
|
||||
.timestamp = entry.timestamp,
|
||||
.domain = entry.domain(),
|
||||
.client_ip = entry.clientIp(),
|
||||
.qtype = entry.qtype,
|
||||
.blocked = entry.blocked,
|
||||
.block_reason = emptyAsNull(entry.blockReason()),
|
||||
.response_time_us = entry.response_time_us,
|
||||
.cache_hit = entry.cache_hit,
|
||||
.upstream = emptyAsNull(entry.upstream()),
|
||||
};
|
||||
}
|
||||
|
||||
fn emptyAsNull(value: []const u8) ?[]const u8 {
|
||||
return if (value.len == 0) null else value;
|
||||
}
|
||||
|
||||
const EntryQueue = std.Io.Queue(Entry);
|
||||
|
||||
/// What the flush interval race can produce. `Select` demands that each field
|
||||
/// type match its task's return type exactly.
|
||||
const Outcome = union(enum) {
|
||||
entry: std.Io.Cancelable!?Entry,
|
||||
expiry: std.Io.Cancelable!void,
|
||||
};
|
||||
|
||||
pub const Logger = struct {
|
||||
cfg: model.Logging,
|
||||
queue: EntryQueue,
|
||||
queries_dropped: std.atomic.Value(u64),
|
||||
rows_written: std.atomic.Value(u64),
|
||||
batches_gated: std.atomic.Value(u64),
|
||||
/// Set when `runWriter` gives up before it consumed anything. The queue is
|
||||
/// closed and every entry counts as dropped from that point, so a caller
|
||||
/// that sees this must not expect rows.
|
||||
writer_failed: std.atomic.Value(bool),
|
||||
|
||||
/// `queue_buf.len` is the backpressure cap — Phase 7 passes
|
||||
/// `cfg.query_log_buffer_max` entries. The queue holds waiting tasks in
|
||||
/// intrusive lists, so a `Logger` must not be moved once anything has
|
||||
/// touched it.
|
||||
pub fn init(cfg: model.Logging, queue_buf: []Entry) Logger {
|
||||
return .{
|
||||
.cfg = cfg,
|
||||
.queue = .init(queue_buf),
|
||||
.queries_dropped = .init(0),
|
||||
.rows_written = .init(0),
|
||||
.batches_gated = .init(0),
|
||||
.writer_failed = .init(false),
|
||||
};
|
||||
}
|
||||
|
||||
/// Applies the privacy transforms and enqueues without ever blocking the
|
||||
/// query path. A full queue loses its oldest unflushed entry (§11.4).
|
||||
pub fn log(self: *Logger, io: std.Io, entry: Entry) void {
|
||||
var transformed = entry;
|
||||
if (self.cfg.hide_domains) transformed.setDomain(hidden_marker);
|
||||
if (self.cfg.hide_client_ips) transformed.setClientIp(hidden_marker);
|
||||
self.enqueue(io, transformed);
|
||||
}
|
||||
|
||||
/// Retries until the put succeeds, and each failed attempt drops exactly
|
||||
/// one oldest entry. A fixed attempt cap would break the policy under
|
||||
/// contention: a producer that steals the slot this call freed would make
|
||||
/// this call pay for two entries, the dropped one and its own.
|
||||
fn enqueue(self: *Logger, io: std.Io, entry: Entry) void {
|
||||
while (true) {
|
||||
// A closed queue or a canceled task means shutdown is underway;
|
||||
// both leave this entry unwritten, which is what the counter says.
|
||||
const put = self.queue.put(io, &.{entry}, 0) catch break;
|
||||
if (put == 1) return;
|
||||
|
||||
// A zero-capacity queue holds nothing to drop: the put above was
|
||||
// this entry's one chance at a waiting getter.
|
||||
if (self.queue.capacity() == 0) break;
|
||||
|
||||
var oldest: [1]Entry = undefined;
|
||||
const got = self.queue.get(io, &oldest, 0) catch break;
|
||||
if (got == 1) self.countDropped(1);
|
||||
}
|
||||
self.countDropped(1);
|
||||
}
|
||||
|
||||
/// The writer task: owns `database` and its prepared statements for its
|
||||
/// whole life. Returns when `shutdown` closes the queue and the last batch
|
||||
/// is flushed, or when the task is canceled.
|
||||
///
|
||||
/// `monitor` is the §11.6 gate. Null disables gating.
|
||||
pub fn runWriter(
|
||||
self: *Logger,
|
||||
io: std.Io,
|
||||
database: *db.Db,
|
||||
monitor: ?*disk_monitor.Monitor,
|
||||
) std.Io.Cancelable!void {
|
||||
var writer = queries_repo.BatchWriter.init(database) catch |err| {
|
||||
scope.warn("query logger: preparing the batch statements failed: {s}", .{@errorName(err)});
|
||||
// Without a writer there is no consumer, so leaving the queue open
|
||||
// would silently swallow every later entry.
|
||||
self.writer_failed.store(true, .release);
|
||||
self.queue.close(io);
|
||||
self.dropRemaining(io);
|
||||
return;
|
||||
};
|
||||
defer writer.deinit();
|
||||
|
||||
var batch: [flush_batch]Entry = undefined;
|
||||
while (true) {
|
||||
// A closed queue hands over its buffered elements before it reports
|
||||
// `Closed` (`Io.zig:2118`), so this drains before it returns.
|
||||
batch[0] = self.queue.getOne(io) catch |err| switch (err) {
|
||||
error.Closed => return,
|
||||
error.Canceled => |e| return e,
|
||||
};
|
||||
const deadline: std.Io.Clock.Timestamp = .fromNow(io, .{
|
||||
.raw = .fromMilliseconds(flush_interval_ms),
|
||||
.clock = .awake,
|
||||
});
|
||||
// `n` is live across both calls: entries already taken off the
|
||||
// queue are lost if either one is canceled, so they must count.
|
||||
var n: usize = 1;
|
||||
self.fill(io, &batch, deadline, &n) catch |err| {
|
||||
self.countDropped(n);
|
||||
return err;
|
||||
};
|
||||
self.flush(io, &writer, batch[0..n], monitor) catch |err| {
|
||||
self.countDropped(n);
|
||||
return err;
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
/// Counts every entry left in a closed queue as dropped. The drain is
|
||||
/// uncancelable: a cancellation racing the writer's own failure would
|
||||
/// otherwise abandon the buffered entries without counting them.
|
||||
fn dropRemaining(self: *Logger, io: std.Io) void {
|
||||
var leftover: [flush_batch]Entry = undefined;
|
||||
while (true) {
|
||||
const n = self.queue.getUncancelable(io, &leftover, 0) catch |err| switch (err) {
|
||||
error.Closed => break,
|
||||
};
|
||||
if (n == 0) break;
|
||||
self.countDropped(n);
|
||||
}
|
||||
}
|
||||
|
||||
/// Closes the queue. `log` drops from here on and `runWriter` returns once
|
||||
/// it has flushed what was left.
|
||||
///
|
||||
/// A writer held by the disk gate keeps holding: it flushes when the disk
|
||||
/// recovers, and Phase 7 cancels the task if it will not wait. A canceled
|
||||
/// writer counts the batch it holds under `queries_dropped`.
|
||||
pub fn shutdown(self: *Logger, io: std.Io) void {
|
||||
self.queue.close(io);
|
||||
}
|
||||
|
||||
/// Fills `batch` behind the entry already in slot 0, until it is full or
|
||||
/// `deadline` passes. `n` counts the slots that hold an entry, and stays
|
||||
/// accurate on the cancellation path so the caller can count what is lost.
|
||||
fn fill(
|
||||
self: *Logger,
|
||||
io: std.Io,
|
||||
batch: *[flush_batch]Entry,
|
||||
deadline: std.Io.Clock.Timestamp,
|
||||
n: *usize,
|
||||
) std.Io.Cancelable!void {
|
||||
n.* += self.drainAvailable(io, batch[n.*..]);
|
||||
|
||||
while (n.* < batch.len) {
|
||||
const remaining = deadline.durationFromNow(io);
|
||||
if (remaining.raw.nanoseconds <= 0) break;
|
||||
const entry = try self.getWithin(io, remaining) orelse break;
|
||||
batch[n.*] = entry;
|
||||
n.* += 1;
|
||||
n.* += self.drainAvailable(io, batch[n.*..]);
|
||||
}
|
||||
}
|
||||
|
||||
/// Whatever is already queued, without blocking.
|
||||
fn drainAvailable(self: *Logger, io: std.Io, room: []Entry) usize {
|
||||
if (room.len == 0) return 0;
|
||||
return self.queue.get(io, room, 0) catch 0;
|
||||
}
|
||||
|
||||
/// Races one blocking `getOne` against the rest of the flush interval —
|
||||
/// `std.Io.Condition` has no timed wait, so the timer is a task.
|
||||
///
|
||||
/// The loser is drained rather than discarded: a `getOne` that finishes
|
||||
/// just after the timer has already taken an entry off the queue, and
|
||||
/// `Select.cancelDiscard` would throw that entry away.
|
||||
fn getWithin(
|
||||
self: *Logger,
|
||||
io: std.Io,
|
||||
budget: std.Io.Clock.Duration,
|
||||
) std.Io.Cancelable!?Entry {
|
||||
var outcomes: [2]Outcome = undefined;
|
||||
var race: std.Io.Select(Outcome) = .init(io, &outcomes);
|
||||
|
||||
race.concurrent(.entry, takeOne, .{ &self.queue, io }) catch |err| switch (err) {
|
||||
// No second unit of concurrency: the caller flushes what it holds
|
||||
// rather than block past the interval.
|
||||
error.ConcurrencyUnavailable => return null,
|
||||
};
|
||||
race.concurrent(.expiry, expire, .{ io, budget }) catch |err| switch (err) {
|
||||
error.ConcurrencyUnavailable => return drainRace(&race),
|
||||
};
|
||||
|
||||
const first = race.await() catch |err| {
|
||||
// Teardown: the entry the getter already took has nowhere to go.
|
||||
if (drainRace(&race)) |_| self.countDropped(1);
|
||||
return err;
|
||||
};
|
||||
const late = drainRace(&race);
|
||||
return outcomeEntry(first) orelse late;
|
||||
}
|
||||
|
||||
/// One batch, one transaction. A batch is dropped whole on a database
|
||||
/// failure: these are log rows, and blocking on them would fill the queue
|
||||
/// and cost live queries instead.
|
||||
fn flush(
|
||||
self: *Logger,
|
||||
io: std.Io,
|
||||
writer: *queries_repo.BatchWriter,
|
||||
entries: []const Entry,
|
||||
monitor: ?*disk_monitor.Monitor,
|
||||
) std.Io.Cancelable!void {
|
||||
if (entries.len == 0) return;
|
||||
|
||||
if (monitor) |m| {
|
||||
const pause: std.Io.Clock.Duration = .{
|
||||
.raw = .fromSeconds(gate_retry_s),
|
||||
.clock = .awake,
|
||||
};
|
||||
while (!m.writesAllowed()) {
|
||||
_ = self.batches_gated.fetchAdd(1, .monotonic);
|
||||
try pause.sleep(io);
|
||||
}
|
||||
}
|
||||
|
||||
var rows: [flush_batch]queries_repo.Row = undefined;
|
||||
for (entries, rows[0..entries.len]) |*entry, *row| row.* = toRow(entry);
|
||||
|
||||
writer.writeBatch(rows[0..entries.len]) catch |err| {
|
||||
scope.warn("query log batch of {d} rows dropped: {s}", .{ entries.len, @errorName(err) });
|
||||
self.countDropped(entries.len);
|
||||
return;
|
||||
};
|
||||
_ = self.rows_written.fetchAdd(entries.len, .monotonic);
|
||||
}
|
||||
|
||||
fn countDropped(self: *Logger, n: usize) void {
|
||||
_ = self.queries_dropped.fetchAdd(n, .monotonic);
|
||||
}
|
||||
};
|
||||
|
||||
fn takeOne(queue: *EntryQueue, io: std.Io) std.Io.Cancelable!?Entry {
|
||||
const entry = queue.getOne(io) catch |err| switch (err) {
|
||||
error.Closed => return null,
|
||||
error.Canceled => |e| return e,
|
||||
};
|
||||
return entry;
|
||||
}
|
||||
|
||||
fn drainRace(race: *std.Io.Select(Outcome)) ?Entry {
|
||||
var found: ?Entry = null;
|
||||
while (race.cancel()) |outcome| {
|
||||
if (outcomeEntry(outcome)) |entry| found = entry;
|
||||
}
|
||||
return found;
|
||||
}
|
||||
|
||||
fn expire(io: std.Io, budget: std.Io.Clock.Duration) std.Io.Cancelable!void {
|
||||
return budget.sleep(io);
|
||||
}
|
||||
|
||||
fn outcomeEntry(outcome: Outcome) ?Entry {
|
||||
return switch (outcome) {
|
||||
.entry => |result| result catch null,
|
||||
.expiry => null,
|
||||
};
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// tests
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
const querylog_schema = @import("querylog_schema.zig");
|
||||
|
||||
const testing = std.testing;
|
||||
|
||||
fn sampleEntry(timestamp: i64, domain: []const u8) Entry {
|
||||
return .init(.{
|
||||
.timestamp = timestamp,
|
||||
.domain = domain,
|
||||
.client_ip = "192.0.2.10",
|
||||
.qtype = 1,
|
||||
.blocked = false,
|
||||
.response_time_us = 900,
|
||||
.cache_hit = false,
|
||||
.upstream = "9.9.9.9",
|
||||
});
|
||||
}
|
||||
|
||||
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;
|
||||
}
|
||||
|
||||
test "an entry carries its own bytes and reads them back" {
|
||||
const entry: Entry = .init(.{
|
||||
.timestamp = 1700000000,
|
||||
.domain = "ads.example.com",
|
||||
.client_ip = "2001:db8::1",
|
||||
.qtype = 28,
|
||||
.blocked = true,
|
||||
.block_reason = "blocklist",
|
||||
.response_time_us = 42,
|
||||
.cache_hit = true,
|
||||
.upstream = "dns.example",
|
||||
});
|
||||
|
||||
try testing.expectEqualStrings("ads.example.com", entry.domain());
|
||||
try testing.expectEqualStrings("2001:db8::1", entry.clientIp());
|
||||
try testing.expectEqualStrings("blocklist", entry.blockReason());
|
||||
try testing.expectEqualStrings("dns.example", entry.upstream());
|
||||
try testing.expectEqual(@as(?u16, 28), entry.qtype);
|
||||
try testing.expect(entry.blocked);
|
||||
try testing.expectEqual(@as(?i64, 42), entry.response_time_us);
|
||||
try testing.expectEqual(@as(?bool, true), entry.cache_hit);
|
||||
}
|
||||
|
||||
test "an oversize string is truncated to what its buffer holds" {
|
||||
const long_domain = "a" ** 400;
|
||||
const entry: Entry = .init(.{
|
||||
.timestamp = 1,
|
||||
.domain = long_domain,
|
||||
.client_ip = "192.0.2.1",
|
||||
.block_reason = "r" ** 64,
|
||||
.upstream = "u" ** 128,
|
||||
});
|
||||
|
||||
try testing.expectEqual(@as(usize, max_domain_len), entry.domain().len);
|
||||
try testing.expectEqual(@as(usize, max_reason_len), entry.blockReason().len);
|
||||
try testing.expectEqual(@as(usize, max_upstream_len), entry.upstream().len);
|
||||
try testing.expectEqualStrings("a" ** max_domain_len, entry.domain());
|
||||
}
|
||||
|
||||
test "toRow maps the empty strings to null and passes the rest through" {
|
||||
const bare: Entry = .init(.{
|
||||
.timestamp = 7,
|
||||
.domain = "example.com",
|
||||
.client_ip = "192.0.2.5",
|
||||
});
|
||||
const bare_row = toRow(&bare);
|
||||
try testing.expectEqual(@as(i64, 7), bare_row.timestamp);
|
||||
try testing.expectEqualStrings("example.com", bare_row.domain);
|
||||
try testing.expectEqualStrings("192.0.2.5", bare_row.client_ip);
|
||||
try testing.expectEqual(@as(?[]const u8, null), bare_row.block_reason);
|
||||
try testing.expectEqual(@as(?[]const u8, null), bare_row.upstream);
|
||||
try testing.expectEqual(@as(?u16, null), bare_row.qtype);
|
||||
try testing.expectEqual(@as(?bool, null), bare_row.cache_hit);
|
||||
|
||||
const full: Entry = .init(.{
|
||||
.timestamp = 8,
|
||||
.domain = "blocked.example",
|
||||
.client_ip = "192.0.2.6",
|
||||
.blocked = true,
|
||||
.block_reason = "blocklist",
|
||||
.upstream = "9.9.9.9",
|
||||
});
|
||||
const full_row = toRow(&full);
|
||||
try testing.expect(full_row.blocked);
|
||||
try testing.expectEqualStrings("blocklist", full_row.block_reason.?);
|
||||
try testing.expectEqualStrings("9.9.9.9", full_row.upstream.?);
|
||||
}
|
||||
|
||||
test "log applies both privacy transforms before the entry reaches the queue" {
|
||||
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
|
||||
var buf: [4]Entry = undefined;
|
||||
var logger: Logger = .init(.{ .hide_domains = true, .hide_client_ips = true }, &buf);
|
||||
|
||||
logger.log(io, sampleEntry(100, "tracker.example"));
|
||||
|
||||
const queued = try logger.queue.getOne(io);
|
||||
try testing.expectEqualStrings(hidden_marker, queued.domain());
|
||||
try testing.expectEqualStrings(hidden_marker, queued.clientIp());
|
||||
try testing.expectEqual(@as(i64, 100), queued.timestamp);
|
||||
try testing.expectEqual(@as(u64, 0), logger.queries_dropped.load(.monotonic));
|
||||
}
|
||||
|
||||
test "log hides only the field its switch names" {
|
||||
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
|
||||
var buf: [4]Entry = undefined;
|
||||
var domains_only: Logger = .init(.{ .hide_domains = true }, &buf);
|
||||
domains_only.log(io, sampleEntry(1, "tracker.example"));
|
||||
const hidden_domain = try domains_only.queue.getOne(io);
|
||||
try testing.expectEqualStrings(hidden_marker, hidden_domain.domain());
|
||||
try testing.expectEqualStrings("192.0.2.10", hidden_domain.clientIp());
|
||||
|
||||
var clients_only: Logger = .init(.{ .hide_client_ips = true }, &buf);
|
||||
clients_only.log(io, sampleEntry(2, "tracker.example"));
|
||||
const hidden_client = try clients_only.queue.getOne(io);
|
||||
try testing.expectEqualStrings("tracker.example", hidden_client.domain());
|
||||
try testing.expectEqualStrings(hidden_marker, hidden_client.clientIp());
|
||||
|
||||
var neither: Logger = .init(.{}, &buf);
|
||||
neither.log(io, sampleEntry(3, "tracker.example"));
|
||||
const untouched = try neither.queue.getOne(io);
|
||||
try testing.expectEqualStrings("tracker.example", untouched.domain());
|
||||
try testing.expectEqualStrings("192.0.2.10", untouched.clientIp());
|
||||
}
|
||||
|
||||
test "a full queue drops the oldest entry and counts it" {
|
||||
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
|
||||
var buf: [2]Entry = undefined;
|
||||
var logger: Logger = .init(.{}, &buf);
|
||||
|
||||
logger.log(io, sampleEntry(1, "first.example"));
|
||||
logger.log(io, sampleEntry(2, "second.example"));
|
||||
logger.log(io, sampleEntry(3, "third.example"));
|
||||
|
||||
try testing.expectEqual(@as(u64, 1), logger.queries_dropped.load(.monotonic));
|
||||
|
||||
const older = try logger.queue.getOne(io);
|
||||
const newer = try logger.queue.getOne(io);
|
||||
try testing.expectEqualStrings("second.example", older.domain());
|
||||
try testing.expectEqualStrings("third.example", newer.domain());
|
||||
}
|
||||
|
||||
test "a zero-capacity queue drops every entry exactly once" {
|
||||
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
|
||||
var buf: [0]Entry = undefined;
|
||||
var logger: Logger = .init(.{}, &buf);
|
||||
|
||||
for (0..5) |i| logger.log(io, sampleEntry(@intCast(i), "example.com"));
|
||||
try testing.expectEqual(@as(u64, 5), logger.queries_dropped.load(.monotonic));
|
||||
}
|
||||
|
||||
test "log after shutdown drops instead of blocking" {
|
||||
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
|
||||
var buf: [4]Entry = undefined;
|
||||
var logger: Logger = .init(.{}, &buf);
|
||||
logger.shutdown(io);
|
||||
|
||||
logger.log(io, sampleEntry(1, "example.com"));
|
||||
try testing.expectEqual(@as(u64, 1), logger.queries_dropped.load(.monotonic));
|
||||
}
|
||||
|
||||
test "the writer drains every entry and shutdown ends it" {
|
||||
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
|
||||
var database = try openLog();
|
||||
defer database.close();
|
||||
|
||||
var buf: [512]Entry = undefined;
|
||||
var logger: Logger = .init(.{}, &buf);
|
||||
|
||||
var future = try io.concurrent(Logger.runWriter, .{
|
||||
&logger,
|
||||
io,
|
||||
&database,
|
||||
@as(?*disk_monitor.Monitor, null),
|
||||
});
|
||||
|
||||
var names: [250][32]u8 = undefined;
|
||||
for (&names, 0..) |*name, i| {
|
||||
const written = try std.fmt.bufPrint(name, "d{d}.example", .{i % 10});
|
||||
logger.log(io, sampleEntry(@intCast(i), written));
|
||||
}
|
||||
logger.shutdown(io);
|
||||
try future.await(io);
|
||||
|
||||
try testing.expectEqual(@as(u64, 0), logger.queries_dropped.load(.monotonic));
|
||||
try testing.expectEqual(@as(u64, 250), logger.rows_written.load(.monotonic));
|
||||
try testing.expectEqual(@as(i64, 250), try queries_repo.countRows(&database));
|
||||
try testing.expectEqual(@as(i64, 10), try queries_repo.countDomains(&database));
|
||||
}
|
||||
|
||||
test "the writer flushes an entry once the interval passes" {
|
||||
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
|
||||
var database = try openLog();
|
||||
defer database.close();
|
||||
|
||||
var buf: [8]Entry = undefined;
|
||||
var logger: Logger = .init(.{}, &buf);
|
||||
|
||||
var future = try io.concurrent(Logger.runWriter, .{
|
||||
&logger,
|
||||
io,
|
||||
&database,
|
||||
@as(?*disk_monitor.Monitor, null),
|
||||
});
|
||||
|
||||
logger.log(io, sampleEntry(1, "only.example"));
|
||||
|
||||
const poll: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(5), .clock = .awake };
|
||||
var waited: usize = 0;
|
||||
while (logger.rows_written.load(.monotonic) == 0) : (waited += 1) {
|
||||
// Ten times the interval; a flush that has not happened by then is a
|
||||
// failure, not slowness.
|
||||
try testing.expect(waited < 200);
|
||||
try poll.sleep(io);
|
||||
}
|
||||
|
||||
logger.shutdown(io);
|
||||
try future.await(io);
|
||||
try testing.expectEqual(@as(i64, 1), try queries_repo.countRows(&database));
|
||||
}
|
||||
|
||||
test "a gated flush holds the batch until the disk recovers" {
|
||||
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
|
||||
var database = try openLog();
|
||||
defer database.close();
|
||||
var writer = try queries_repo.BatchWriter.init(&database);
|
||||
defer writer.deinit();
|
||||
|
||||
var buf: [4]Entry = undefined;
|
||||
var logger: Logger = .init(.{}, &buf);
|
||||
|
||||
var monitor: disk_monitor.Monitor = .init(.{}, std.Io.Dir.cwd(), ".", null);
|
||||
monitor.state_raw.store(@intFromEnum(disk_monitor.State.critical), .monotonic);
|
||||
try testing.expect(!monitor.writesAllowed());
|
||||
|
||||
const entries = [_]Entry{ sampleEntry(1, "held.example"), sampleEntry(2, "held.example") };
|
||||
var future = try io.concurrent(Logger.flush, .{
|
||||
&logger,
|
||||
io,
|
||||
&writer,
|
||||
@as([]const Entry, &entries),
|
||||
@as(?*disk_monitor.Monitor, &monitor),
|
||||
});
|
||||
|
||||
const poll: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(5), .clock = .awake };
|
||||
var waited: usize = 0;
|
||||
while (logger.batches_gated.load(.monotonic) == 0) : (waited += 1) {
|
||||
try testing.expect(waited < 200);
|
||||
try poll.sleep(io);
|
||||
}
|
||||
try testing.expectEqual(@as(i64, 0), try queries_repo.countRows(&database));
|
||||
|
||||
monitor.state_raw.store(@intFromEnum(disk_monitor.State.ok), .monotonic);
|
||||
try future.await(io);
|
||||
|
||||
try testing.expect(logger.batches_gated.load(.monotonic) >= 1);
|
||||
try testing.expectEqual(@as(u64, 2), logger.rows_written.load(.monotonic));
|
||||
try testing.expectEqual(@as(i64, 2), try queries_repo.countRows(&database));
|
||||
}
|
||||
|
||||
test "a failing batch is dropped whole and the writer stays usable" {
|
||||
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
|
||||
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 queries_repo.BatchWriter.init(&database);
|
||||
defer writer.deinit();
|
||||
|
||||
var buf: [4]Entry = undefined;
|
||||
var logger: Logger = .init(.{}, &buf);
|
||||
|
||||
var doomed = sampleEntry(10, "poison.example");
|
||||
doomed.setClientIp("boom");
|
||||
const bad = [_]Entry{ sampleEntry(9, "good.example"), doomed };
|
||||
try logger.flush(io, &writer, &bad, null);
|
||||
|
||||
try testing.expectEqual(@as(i64, 0), try queries_repo.countRows(&database));
|
||||
try testing.expectEqual(@as(u64, 0), logger.rows_written.load(.monotonic));
|
||||
try testing.expectEqual(@as(u64, 2), logger.queries_dropped.load(.monotonic));
|
||||
|
||||
const good = [_]Entry{sampleEntry(11, "next.example")};
|
||||
try logger.flush(io, &writer, &good, null);
|
||||
try testing.expectEqual(@as(i64, 1), try queries_repo.countRows(&database));
|
||||
try testing.expectEqual(@as(u64, 1), logger.rows_written.load(.monotonic));
|
||||
}
|
||||
|
||||
test "a writer that cannot prepare closes the queue and counts every entry" {
|
||||
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
|
||||
// No schema: `BatchWriter.init` cannot prepare against a missing table.
|
||||
var database = try db.Db.open(":memory:", .{ .mode = .memory });
|
||||
defer database.close();
|
||||
|
||||
var buf: [8]Entry = undefined;
|
||||
var logger: Logger = .init(.{}, &buf);
|
||||
|
||||
for (0..3) |i| logger.log(io, sampleEntry(@intCast(i), "early.example"));
|
||||
|
||||
try logger.runWriter(io, &database, null);
|
||||
|
||||
try testing.expect(logger.writer_failed.load(.acquire));
|
||||
try testing.expectEqual(@as(u64, 3), logger.queries_dropped.load(.monotonic));
|
||||
try testing.expectEqual(@as(u64, 0), logger.rows_written.load(.monotonic));
|
||||
|
||||
// The queue is closed, so later entries drop and count instead of piling up.
|
||||
logger.log(io, sampleEntry(99, "late.example"));
|
||||
try testing.expectEqual(@as(u64, 4), logger.queries_dropped.load(.monotonic));
|
||||
}
|
||||
|
||||
test "a canceled writer counts the batch it was holding" {
|
||||
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
|
||||
var database = try openLog();
|
||||
defer database.close();
|
||||
|
||||
var buf: [8]Entry = undefined;
|
||||
var logger: Logger = .init(.{}, &buf);
|
||||
|
||||
var monitor: disk_monitor.Monitor = .init(.{}, std.Io.Dir.cwd(), ".", null);
|
||||
monitor.state_raw.store(@intFromEnum(disk_monitor.State.critical), .monotonic);
|
||||
|
||||
// Both entries are queued before the writer starts, so the batch it takes
|
||||
// into the gate holds exactly two.
|
||||
logger.log(io, sampleEntry(1, "held.example"));
|
||||
logger.log(io, sampleEntry(2, "held.example"));
|
||||
|
||||
var future = try io.concurrent(Logger.runWriter, .{
|
||||
&logger,
|
||||
io,
|
||||
&database,
|
||||
@as(?*disk_monitor.Monitor, &monitor),
|
||||
});
|
||||
|
||||
const poll: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(5), .clock = .awake };
|
||||
var waited: usize = 0;
|
||||
while (logger.batches_gated.load(.monotonic) == 0) : (waited += 1) {
|
||||
try testing.expect(waited < 400);
|
||||
try poll.sleep(io);
|
||||
}
|
||||
|
||||
try testing.expectError(error.Canceled, future.cancel(io));
|
||||
|
||||
try testing.expectEqual(@as(u64, 2), logger.queries_dropped.load(.monotonic));
|
||||
try testing.expectEqual(@as(u64, 0), logger.rows_written.load(.monotonic));
|
||||
try testing.expectEqual(@as(i64, 0), try queries_repo.countRows(&database));
|
||||
}
|
||||
|
||||
test "an empty batch touches neither the database nor the counters" {
|
||||
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
|
||||
var database = try openLog();
|
||||
defer database.close();
|
||||
var writer = try queries_repo.BatchWriter.init(&database);
|
||||
defer writer.deinit();
|
||||
|
||||
var buf: [4]Entry = undefined;
|
||||
var logger: Logger = .init(.{}, &buf);
|
||||
|
||||
var monitor: disk_monitor.Monitor = .init(.{}, std.Io.Dir.cwd(), ".", null);
|
||||
monitor.state_raw.store(@intFromEnum(disk_monitor.State.critical), .monotonic);
|
||||
|
||||
// Gated or not, an empty batch returns before it reads the monitor.
|
||||
try logger.flush(io, &writer, &.{}, &monitor);
|
||||
|
||||
try testing.expectEqual(@as(u64, 0), logger.batches_gated.load(.monotonic));
|
||||
try testing.expectEqual(@as(u64, 0), logger.rows_written.load(.monotonic));
|
||||
try testing.expectEqual(@as(i64, 0), try queries_repo.countRows(&database));
|
||||
}
|
||||
@@ -0,0 +1,609 @@
|
||||
//! Milestone-6 integration tests (spec S8): the phase-6 components against real
|
||||
//! files, a real `querylog.db`, a real filesystem sample and a real log sink.
|
||||
//!
|
||||
//! This lives in its own file because it needs `@import("build_options")`, which
|
||||
//! only exists when the compilation is driven by `build.zig`. The body compiles
|
||||
//! on every `zig build test` run, so it cannot rot, and every case skips at run
|
||||
//! time unless `-Dintegration` is passed.
|
||||
//!
|
||||
//! Hermetic: every case works inside one `std.testing.tmpDir` and none of them
|
||||
//! opens a socket or resolves a name.
|
||||
//!
|
||||
//! Two mechanisms resolve the same paths here. `std.Io.Dir` calls go through the
|
||||
//! temporary directory handle, while SQLite and the log sink resolve their
|
||||
//! filenames through the process working directory. Every path handed to those
|
||||
//! two is therefore built from `Fixture.root`.
|
||||
|
||||
const std = @import("std");
|
||||
const build_options = @import("build_options");
|
||||
|
||||
const dns_cache = @import("../cache/dns_cache.zig");
|
||||
const address = @import("../platform/address.zig");
|
||||
const logging = @import("../platform/logging.zig");
|
||||
const packet = @import("../dns/packet.zig");
|
||||
const rate_limiter = @import("../server/rate_limiter.zig");
|
||||
const db = @import("db.zig");
|
||||
const disk_monitor = @import("disk_monitor.zig");
|
||||
const logger = @import("logger.zig");
|
||||
const queries_repo = @import("repositories/queries_repo.zig");
|
||||
const querylog_schema = @import("querylog_schema.zig");
|
||||
const retention = @import("retention.zig");
|
||||
|
||||
const testing = std.testing;
|
||||
|
||||
/// `std.testing.tmpDir` creates its directory against `std.testing.io`, so every
|
||||
/// call into the code under test uses the same `Io` instance. That instance is
|
||||
/// an `Io.Threaded` (`lib/std/testing.zig:34`), which is what makes
|
||||
/// `io.concurrent` available to the writer-task cases.
|
||||
const io = testing.io;
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// fixture
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// Where `std.testing.tmpDir` puts its directories (`lib/std/testing.zig:634`).
|
||||
const tmp_prefix = ".zig-cache/tmp/";
|
||||
|
||||
const sub_path_len = @typeInfo(@FieldType(testing.TmpDir, "sub_path")).array.len;
|
||||
|
||||
const path_buf_len = 256;
|
||||
|
||||
const Fixture = struct {
|
||||
tmp: testing.TmpDir,
|
||||
root_buf: [tmp_prefix.len + sub_path_len]u8,
|
||||
|
||||
fn init() Fixture {
|
||||
var self: Fixture = .{
|
||||
.tmp = testing.tmpDir(.{ .iterate = true }),
|
||||
.root_buf = undefined,
|
||||
};
|
||||
@memcpy(self.root_buf[0..tmp_prefix.len], tmp_prefix);
|
||||
@memcpy(self.root_buf[tmp_prefix.len..], &self.tmp.sub_path);
|
||||
return self;
|
||||
}
|
||||
|
||||
fn deinit(self: *Fixture) void {
|
||||
self.tmp.cleanup();
|
||||
}
|
||||
|
||||
/// The temporary directory as a path relative to the process working
|
||||
/// directory.
|
||||
fn root(self: *const Fixture) []const u8 {
|
||||
return &self.root_buf;
|
||||
}
|
||||
|
||||
fn path(self: *const Fixture, buf: []u8, name: []const u8) ![]const u8 {
|
||||
return std.fmt.bufPrint(buf, "{s}/{s}", .{ self.root(), name });
|
||||
}
|
||||
|
||||
fn pathZ(self: *const Fixture, buf: []u8, name: []const u8) ![:0]const u8 {
|
||||
return std.fmt.bufPrintZ(buf, "{s}/{s}", .{ self.root(), name });
|
||||
}
|
||||
|
||||
fn rootZ(self: *const Fixture, buf: []u8) ![:0]const u8 {
|
||||
return std.fmt.bufPrintZ(buf, "{s}", .{self.root()});
|
||||
}
|
||||
|
||||
fn exists(self: *const Fixture, name: []const u8) !bool {
|
||||
self.tmp.dir.access(io, name, .{}) catch |e| switch (e) {
|
||||
error.FileNotFound => return false,
|
||||
else => |other| return other,
|
||||
};
|
||||
return true;
|
||||
}
|
||||
|
||||
fn sizeOf(self: *const Fixture, name: []const u8) !u64 {
|
||||
const stat = try self.tmp.dir.statFile(io, name, .{});
|
||||
return stat.size;
|
||||
}
|
||||
};
|
||||
|
||||
/// A fresh `querylog.db` inside the fixture, opened the way the daemon opens it.
|
||||
const QueryLog = struct {
|
||||
opened: querylog_schema.OpenResult,
|
||||
|
||||
fn create(f: *const Fixture) !QueryLog {
|
||||
var buf: [path_buf_len]u8 = undefined;
|
||||
const path = try f.pathZ(&buf, "querylog.db");
|
||||
return .{ .opened = try querylog_schema.open(io, std.Io.Dir.cwd(), path) };
|
||||
}
|
||||
|
||||
fn deinit(self: *QueryLog) void {
|
||||
self.opened.database.close();
|
||||
}
|
||||
|
||||
fn database(self: *QueryLog) *db.Db {
|
||||
return &self.opened.database;
|
||||
}
|
||||
};
|
||||
|
||||
fn entryAt(timestamp: i64, domain: []const u8) logger.Entry {
|
||||
return .init(.{
|
||||
.timestamp = timestamp,
|
||||
.domain = domain,
|
||||
.client_ip = "192.0.2.10",
|
||||
.qtype = 1,
|
||||
.blocked = false,
|
||||
.response_time_us = 1200,
|
||||
.cache_hit = false,
|
||||
.upstream = "9.9.9.9",
|
||||
});
|
||||
}
|
||||
|
||||
const poll_interval: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(5), .clock = .awake };
|
||||
|
||||
/// Waits for `counter` to reach `target`, up to `limit` polls of 5 ms. A
|
||||
/// deadline that passes is a failure rather than slowness: every case here is
|
||||
/// bounded well below the 2 s the spec allows.
|
||||
fn awaitCount(counter: *const std.atomic.Value(u64), target: u64, limit: usize) !void {
|
||||
var polls: usize = 0;
|
||||
while (counter.load(.monotonic) < target) : (polls += 1) {
|
||||
try testing.expect(polls < limit);
|
||||
try poll_interval.sleep(io);
|
||||
}
|
||||
}
|
||||
|
||||
fn writeRows(database: *db.Db, timestamps: []const i64, domain: []const u8) !void {
|
||||
var writer = try queries_repo.BatchWriter.init(database);
|
||||
defer writer.deinit();
|
||||
|
||||
var rows: [16]queries_repo.Row = undefined;
|
||||
for (timestamps, rows[0..timestamps.len]) |timestamp, *row| {
|
||||
row.* = .{
|
||||
.timestamp = timestamp,
|
||||
.domain = domain,
|
||||
.client_ip = "192.0.2.10",
|
||||
.qtype = 1,
|
||||
.blocked = false,
|
||||
.block_reason = null,
|
||||
.response_time_us = null,
|
||||
.cache_hit = null,
|
||||
.upstream = null,
|
||||
};
|
||||
}
|
||||
try writer.writeBatch(rows[0..timestamps.len]);
|
||||
}
|
||||
|
||||
/// The timestamps in the log, oldest first.
|
||||
fn readTimestamps(database: *db.Db, out: []i64) ![]i64 {
|
||||
var stmt = try database.prepare("SELECT timestamp FROM query_log ORDER BY timestamp");
|
||||
defer stmt.deinit();
|
||||
var n: usize = 0;
|
||||
while (try stmt.step()) : (n += 1) {
|
||||
if (n == out.len) return error.TestUnexpectedResult;
|
||||
out[n] = stmt.columnInt(0);
|
||||
}
|
||||
return out[0..n];
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// case 1-6: the query logger, retention and the disk gate on a real database
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
test "S8 case 1: the logger writes a real querylog.db end to end" {
|
||||
if (!build_options.integration) return error.SkipZigTest;
|
||||
|
||||
var f: Fixture = .init();
|
||||
defer f.deinit();
|
||||
|
||||
var log_db = try QueryLog.create(&f);
|
||||
defer log_db.deinit();
|
||||
try testing.expectEqual(querylog_schema.RecreateReason.missing, log_db.opened.recreated.?);
|
||||
|
||||
var queue_buf: [512]logger.Entry = undefined;
|
||||
var query_log: logger.Logger = .init(.{}, &queue_buf);
|
||||
|
||||
var future = try io.concurrent(logger.Logger.runWriter, .{
|
||||
&query_log,
|
||||
io,
|
||||
log_db.database(),
|
||||
@as(?*disk_monitor.Monitor, null),
|
||||
});
|
||||
|
||||
var name_buf: [32]u8 = undefined;
|
||||
for (0..250) |i| {
|
||||
const domain = try std.fmt.bufPrint(&name_buf, "d{d}.example", .{i % 10});
|
||||
query_log.log(io, entryAt(@intCast(i), domain));
|
||||
}
|
||||
query_log.shutdown(io);
|
||||
try future.await(io);
|
||||
|
||||
try testing.expectEqual(@as(u64, 0), query_log.queries_dropped.load(.monotonic));
|
||||
try testing.expectEqual(@as(u64, 250), query_log.rows_written.load(.monotonic));
|
||||
try testing.expectEqual(@as(i64, 250), try queries_repo.countRows(log_db.database()));
|
||||
try testing.expectEqual(@as(i64, 10), try queries_repo.countDomains(log_db.database()));
|
||||
try testing.expectEqual(
|
||||
@as(i64, querylog_schema.fingerprint),
|
||||
try log_db.database().queryInt("PRAGMA user_version"),
|
||||
);
|
||||
}
|
||||
|
||||
test "S8 case 2: a single entry reaches the file once the flush interval passes" {
|
||||
if (!build_options.integration) return error.SkipZigTest;
|
||||
|
||||
var f: Fixture = .init();
|
||||
defer f.deinit();
|
||||
|
||||
var log_db = try QueryLog.create(&f);
|
||||
defer log_db.deinit();
|
||||
|
||||
var queue_buf: [8]logger.Entry = undefined;
|
||||
var query_log: logger.Logger = .init(.{}, &queue_buf);
|
||||
|
||||
var future = try io.concurrent(logger.Logger.runWriter, .{
|
||||
&query_log,
|
||||
io,
|
||||
log_db.database(),
|
||||
@as(?*disk_monitor.Monitor, null),
|
||||
});
|
||||
|
||||
query_log.log(io, entryAt(1, "only.example"));
|
||||
|
||||
// Ten flush intervals of headroom: a row that has not landed by then is a
|
||||
// failure of the interval race, not a slow machine.
|
||||
const limit = 10 * logger.flush_interval_ms / 5;
|
||||
try awaitCount(&query_log.rows_written, 1, limit);
|
||||
try testing.expectEqual(@as(i64, 1), try queries_repo.countRows(log_db.database()));
|
||||
|
||||
query_log.shutdown(io);
|
||||
try future.await(io);
|
||||
}
|
||||
|
||||
test "S8 case 3: a full queue drops the oldest entries and the newest survive" {
|
||||
if (!build_options.integration) return error.SkipZigTest;
|
||||
|
||||
var f: Fixture = .init();
|
||||
defer f.deinit();
|
||||
|
||||
var log_db = try QueryLog.create(&f);
|
||||
defer log_db.deinit();
|
||||
|
||||
var root_buf: [path_buf_len]u8 = undefined;
|
||||
var monitor: disk_monitor.Monitor = .init(
|
||||
.{},
|
||||
f.tmp.dir,
|
||||
try f.rootZ(&root_buf),
|
||||
null,
|
||||
);
|
||||
monitor.state_raw.store(@intFromEnum(disk_monitor.State.critical), .monotonic);
|
||||
|
||||
var queue_buf: [8]logger.Entry = undefined;
|
||||
var query_log: logger.Logger = .init(.{}, &queue_buf);
|
||||
|
||||
// The whole burst is enqueued before the writer starts. A writer already
|
||||
// draining the queue would take entries out of it mid-burst and make the
|
||||
// number of drops depend on the scheduler.
|
||||
for (0..20) |i| query_log.log(io, entryAt(@intCast(i), "burst.example"));
|
||||
try testing.expectEqual(@as(u64, 12), query_log.queries_dropped.load(.monotonic));
|
||||
|
||||
var future = try io.concurrent(logger.Logger.runWriter, .{
|
||||
&query_log,
|
||||
io,
|
||||
log_db.database(),
|
||||
@as(?*disk_monitor.Monitor, &monitor),
|
||||
});
|
||||
|
||||
try awaitCount(&query_log.batches_gated, 1, 200);
|
||||
try testing.expectEqual(@as(i64, 0), try queries_repo.countRows(log_db.database()));
|
||||
|
||||
// Un-gate before the shutdown: a writer held by the disk gate holds its
|
||||
// batch, and `shutdown` alone would never release it.
|
||||
monitor.state_raw.store(@intFromEnum(disk_monitor.State.ok), .monotonic);
|
||||
try awaitCount(&query_log.rows_written, 8, 300);
|
||||
query_log.shutdown(io);
|
||||
try future.await(io);
|
||||
|
||||
var stamps: [16]i64 = undefined;
|
||||
const kept = try readTimestamps(log_db.database(), &stamps);
|
||||
try testing.expectEqual(@as(usize, 8), kept.len);
|
||||
for (kept, 0..) |stamp, i| {
|
||||
try testing.expectEqual(@as(i64, @intCast(i + 12)), stamp);
|
||||
}
|
||||
}
|
||||
|
||||
test "S8 case 4: the privacy transforms reach the stored rows" {
|
||||
if (!build_options.integration) return error.SkipZigTest;
|
||||
|
||||
var f: Fixture = .init();
|
||||
defer f.deinit();
|
||||
|
||||
var log_db = try QueryLog.create(&f);
|
||||
defer log_db.deinit();
|
||||
|
||||
var queue_buf: [64]logger.Entry = undefined;
|
||||
var query_log: logger.Logger = .init(
|
||||
.{ .hide_domains = true, .hide_client_ips = true },
|
||||
&queue_buf,
|
||||
);
|
||||
|
||||
var future = try io.concurrent(logger.Logger.runWriter, .{
|
||||
&query_log,
|
||||
io,
|
||||
log_db.database(),
|
||||
@as(?*disk_monitor.Monitor, null),
|
||||
});
|
||||
|
||||
var name_buf: [32]u8 = undefined;
|
||||
for (0..20) |i| {
|
||||
const domain = try std.fmt.bufPrint(&name_buf, "private{d}.example", .{i});
|
||||
query_log.log(io, entryAt(@intCast(i), domain));
|
||||
}
|
||||
query_log.shutdown(io);
|
||||
try future.await(io);
|
||||
|
||||
try testing.expectEqual(@as(i64, 20), try queries_repo.countRows(log_db.database()));
|
||||
// Every name collapsed onto the marker, so the dimension table holds one row.
|
||||
try testing.expectEqual(@as(i64, 1), try queries_repo.countDomains(log_db.database()));
|
||||
try testing.expectEqual(
|
||||
@as(i64, 20),
|
||||
try log_db.database().queryInt(
|
||||
\\SELECT count(*) FROM query_log q JOIN domains d ON d.id = q.domain_id
|
||||
\\ WHERE d.domain = 'hidden' AND q.client_ip = 'hidden'
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
test "S8 case 5: a retention pass prunes the old rows and truncates the write-ahead log" {
|
||||
if (!build_options.integration) return error.SkipZigTest;
|
||||
|
||||
var f: Fixture = .init();
|
||||
defer f.deinit();
|
||||
|
||||
var log_db = try QueryLog.create(&f);
|
||||
defer log_db.deinit();
|
||||
|
||||
const now = std.Io.Clock.real.now(io).toSeconds();
|
||||
const day = 86_400;
|
||||
try writeRows(log_db.database(), &.{ now - 40 * day, now - 31 * day }, "old.example");
|
||||
try writeRows(log_db.database(), &.{ now - 3 * day, now - 60 }, "fresh.example");
|
||||
try testing.expectEqual(@as(i64, 4), try queries_repo.countRows(log_db.database()));
|
||||
try testing.expect(try f.sizeOf("querylog.db-wal") > 0);
|
||||
|
||||
var pass: retention.Retention = .init(.{ .retention_days = 30 });
|
||||
pass.runOnce(io, log_db.database());
|
||||
|
||||
try testing.expectEqual(@as(u64, 1), pass.stats.passes);
|
||||
try testing.expectEqual(@as(u64, 2), pass.stats.rows_pruned);
|
||||
try testing.expectEqual(@as(u64, 1), pass.stats.checkpoints);
|
||||
try testing.expectEqual(@as(u64, 0), pass.stats.vacuums);
|
||||
try testing.expectEqual(@as(i64, 2), try queries_repo.countRows(log_db.database()));
|
||||
// Both names stay: the dimension table is not collected.
|
||||
try testing.expectEqual(@as(i64, 2), try queries_repo.countDomains(log_db.database()));
|
||||
|
||||
if (try f.exists("querylog.db-wal")) {
|
||||
try testing.expectEqual(@as(u64, 0), try f.sizeOf("querylog.db-wal"));
|
||||
}
|
||||
}
|
||||
|
||||
test "S8 case 6: a critical disk gates the flushes and recovery releases them" {
|
||||
if (!build_options.integration) return error.SkipZigTest;
|
||||
|
||||
var f: Fixture = .init();
|
||||
defer f.deinit();
|
||||
|
||||
var log_db = try QueryLog.create(&f);
|
||||
defer log_db.deinit();
|
||||
|
||||
var root_buf: [path_buf_len]u8 = undefined;
|
||||
const data_path = try f.rootZ(&root_buf);
|
||||
|
||||
// No filesystem holds this much free space, so the sample classifies
|
||||
// critical against the real `statvfs` reading rather than a stub.
|
||||
const unreachable_mb = std.math.maxInt(u32);
|
||||
var monitor: disk_monitor.Monitor = .init(
|
||||
.{ .min_free_mb = unreachable_mb, .warn_free_mb = unreachable_mb },
|
||||
f.tmp.dir,
|
||||
data_path,
|
||||
null,
|
||||
);
|
||||
monitor.sample(io);
|
||||
try testing.expectEqual(disk_monitor.State.critical, monitor.state());
|
||||
try testing.expect(!monitor.writesAllowed());
|
||||
try testing.expect(monitor.gauges().free_bytes > 0);
|
||||
try testing.expect(monitor.gauges().db_bytes > 0);
|
||||
|
||||
var queue_buf: [64]logger.Entry = undefined;
|
||||
var query_log: logger.Logger = .init(.{}, &queue_buf);
|
||||
for (0..5) |i| query_log.log(io, entryAt(@intCast(i), "gated.example"));
|
||||
|
||||
var future = try io.concurrent(logger.Logger.runWriter, .{
|
||||
&query_log,
|
||||
io,
|
||||
log_db.database(),
|
||||
@as(?*disk_monitor.Monitor, &monitor),
|
||||
});
|
||||
|
||||
try awaitCount(&query_log.batches_gated, 1, 200);
|
||||
try testing.expectEqual(@as(u64, 0), query_log.rows_written.load(.monotonic));
|
||||
try testing.expectEqual(@as(i64, 0), try queries_repo.countRows(log_db.database()));
|
||||
|
||||
monitor.cfg = .{ .min_free_mb = 0, .warn_free_mb = 0 };
|
||||
monitor.sample(io);
|
||||
try testing.expectEqual(disk_monitor.State.ok, monitor.state());
|
||||
try testing.expect(monitor.writesAllowed());
|
||||
|
||||
// The gate re-reads the monitor once per `gate_retry_s`, so the release
|
||||
// costs at most that one second.
|
||||
const limit = (logger.gate_retry_s * 1000 + 500) / 5;
|
||||
try awaitCount(&query_log.rows_written, 5, limit);
|
||||
|
||||
query_log.shutdown(io);
|
||||
try future.await(io);
|
||||
|
||||
try testing.expect(query_log.batches_gated.load(.monotonic) > 0);
|
||||
try testing.expectEqual(@as(i64, 5), try queries_repo.countRows(log_db.database()));
|
||||
try testing.expectEqual(@as(u64, 0), monitor.sample_failures.load(.monotonic));
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// case 7: the log sink against a real file
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
test "S8 case 7: the log sink appends, rotates and honours max_files" {
|
||||
if (!build_options.integration) return error.SkipZigTest;
|
||||
|
||||
var f: Fixture = .init();
|
||||
defer f.deinit();
|
||||
|
||||
var path_buf: [path_buf_len]u8 = undefined;
|
||||
const log_path = try f.path(&path_buf, "nxdns.log");
|
||||
|
||||
const before = logging.stats();
|
||||
|
||||
// `install` would take the 1 MiB floor of `max_size_mb`, which is a
|
||||
// megabyte of writes per generation; `installForTest` overrides that one
|
||||
// threshold and nothing else.
|
||||
const max_bytes = 256;
|
||||
logging.installForTest(io, .{
|
||||
.level = .info,
|
||||
.output = .file,
|
||||
.file_path = log_path,
|
||||
.max_files = 3,
|
||||
}, max_bytes);
|
||||
defer logging.deinstall();
|
||||
|
||||
// `logFn` is called directly: the test runner installs its own
|
||||
// `std_options`, so a `std.log` call here would never reach this sink.
|
||||
for (0..40) |i| logging.logFn(.warn, .s8_sink, "rotation line {d}", .{i});
|
||||
for (0..5) |i| logging.logFn(.info, .s8_sink, "tail line {d}", .{i});
|
||||
// Below the configured threshold, so it is filtered rather than written.
|
||||
logging.logFn(.debug, .s8_sink, "never written", .{});
|
||||
|
||||
logging.deinstall();
|
||||
|
||||
const after = logging.stats();
|
||||
try testing.expectEqual(@as(u64, 45), after.lines_written - before.lines_written);
|
||||
try testing.expect(after.rotations > before.rotations);
|
||||
try testing.expectEqual(@as(u64, 0), after.sink_errors - before.sink_errors);
|
||||
try testing.expectEqual(@as(u64, 0), after.lines_deduped - before.lines_deduped);
|
||||
|
||||
try testing.expect(try f.exists("nxdns.log"));
|
||||
try testing.expect(try f.sizeOf("nxdns.log") <= max_bytes);
|
||||
try testing.expect(try f.exists("nxdns.log.1"));
|
||||
try testing.expect(try f.exists("nxdns.log.2"));
|
||||
// `max_files` counts the live file, so generation 3 is never created.
|
||||
try testing.expect(!try f.exists("nxdns.log.3"));
|
||||
|
||||
// The live file holds the newest lines, and it is reopened rather than
|
||||
// truncated: a second install appends behind what is already there.
|
||||
const kept = try f.tmp.dir.readFileAlloc(io, "nxdns.log", testing.allocator, .limited(4096));
|
||||
defer testing.allocator.free(kept);
|
||||
try testing.expect(std.mem.count(u8, kept, "tail line 4") == 1);
|
||||
try testing.expect(std.mem.count(u8, kept, "(s8_sink)") >= 1);
|
||||
|
||||
const live_bytes = kept.len;
|
||||
logging.installForTest(io, .{
|
||||
.level = .info,
|
||||
.output = .file,
|
||||
.file_path = log_path,
|
||||
.max_files = 3,
|
||||
}, max_bytes * 16);
|
||||
logging.logFn(.info, .s8_sink, "after reopen", .{});
|
||||
logging.deinstall();
|
||||
|
||||
const reopened = try f.tmp.dir.readFileAlloc(io, "nxdns.log", testing.allocator, .limited(8192));
|
||||
defer testing.allocator.free(reopened);
|
||||
try testing.expect(reopened.len > live_bytes);
|
||||
try testing.expectEqualStrings(kept, reopened[0..live_bytes]);
|
||||
try testing.expect(std.mem.count(u8, reopened, "after reopen") == 1);
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// case 8-9: the pure components against real packets and real addresses
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// A NOERROR response for example.com A carrying one answer per TTL.
|
||||
fn buildAnswer(buf: []u8, ttls: []const u32) ![]u8 {
|
||||
const query = "\x12\x34\x01\x00\x00\x01\x00\x00\x00\x00\x00\x00" ++
|
||||
"\x07example\x03com\x00\x00\x01\x00\x01";
|
||||
const request = try packet.parse(query);
|
||||
const q = packet.firstQuestion(request).?;
|
||||
|
||||
var builder = try packet.ResponseBuilder.init(buf, request.header, q);
|
||||
for (ttls) |ttl| {
|
||||
try builder.addAnswer(q.name, .a, .in, ttl, "\x0a\x00\x00\x01");
|
||||
}
|
||||
return builder.finish();
|
||||
}
|
||||
|
||||
fn firstAnswerTtl(bytes: []const u8) !u32 {
|
||||
const p = try packet.parse(bytes);
|
||||
var it = packet.answers(p);
|
||||
return (try it.next()).?.ttl;
|
||||
}
|
||||
|
||||
test "S8 case 8: the cache ages a real response and expires it at the boundary" {
|
||||
if (!build_options.integration) return error.SkipZigTest;
|
||||
|
||||
var response_buf: [512]u8 = undefined;
|
||||
const response = try buildAnswer(&response_buf, &.{ 300, 600 });
|
||||
const class = dns_cache.classify(response, 3600).?;
|
||||
try testing.expectEqual(@as(u32, 300), class.ttl_seconds);
|
||||
try testing.expectEqual(false, class.negative);
|
||||
|
||||
var cache = try dns_cache.DnsCache.init(testing.allocator, .{ .size = 16, .negative_ttl_max = 3600 });
|
||||
defer cache.deinit();
|
||||
|
||||
var key_buf: [dns_cache.max_key_len]u8 = undefined;
|
||||
const key = dns_cache.buildKey(&key_buf, "example.com", 1, 1, false, null);
|
||||
|
||||
try cache.put(1000, key, response, class);
|
||||
try testing.expectEqual(@as(u32, 1), cache.len());
|
||||
|
||||
var out: [512]u8 = undefined;
|
||||
const hit = cache.get(1120, key, &out).?;
|
||||
try testing.expectEqual(response.len, hit.len);
|
||||
try testing.expectEqual(@as(u32, 180), try firstAnswerTtl(hit));
|
||||
|
||||
// The stored copy keeps its own age, so a later hit ages from the same base.
|
||||
const later = cache.get(1290, key, &out).?;
|
||||
try testing.expectEqual(@as(u32, 10), try firstAnswerTtl(later));
|
||||
|
||||
// The transaction ID is the caller's to set, and the aged bytes still parse.
|
||||
packet.setId(later, 0xbeef);
|
||||
try testing.expectEqual(@as(u16, 0xbeef), (try packet.parse(later)).header.id);
|
||||
|
||||
// The entry expires at stored_at + ttl, and that second is already too late.
|
||||
try testing.expectEqual(@as(?[]u8, null), cache.get(1300, key, &out));
|
||||
try testing.expectEqual(@as(u32, 0), cache.len());
|
||||
try testing.expectEqual(@as(u64, 2), cache.stats.hits);
|
||||
try testing.expectEqual(@as(u64, 1), cache.stats.expirations);
|
||||
}
|
||||
|
||||
test "S8 case 9: the limiter refuses the query past the limit and only that client" {
|
||||
if (!build_options.integration) return error.SkipZigTest;
|
||||
|
||||
var limiter = try rate_limiter.RateLimiter.init(
|
||||
testing.allocator,
|
||||
.{ .limit = 1000, .window_seconds = 60 },
|
||||
);
|
||||
defer limiter.deinit();
|
||||
|
||||
const mapped = address.NetAddress.fromIp(
|
||||
try std.Io.net.IpAddress.parse("::ffff:192.168.1.40", 53),
|
||||
).key();
|
||||
const plain = (try address.NetAddress.parse("192.168.1.40")).key();
|
||||
// One client, whichever family the socket reported it under.
|
||||
try testing.expectEqualSlices(u8, &plain, &mapped);
|
||||
|
||||
const start: std.Io.Timestamp = .{ .nanoseconds = 1 << 80 };
|
||||
for (0..1000) |i| {
|
||||
const now: std.Io.Timestamp = .{ .nanoseconds = start.nanoseconds + @as(i96, @intCast(i)) };
|
||||
try testing.expect(limiter.check(now, mapped));
|
||||
}
|
||||
try testing.expect(!limiter.check(start, plain));
|
||||
|
||||
try testing.expectEqual(@as(u64, 1000), limiter.stats.allowed);
|
||||
try testing.expectEqual(@as(u64, 1), limiter.stats.refused);
|
||||
try testing.expectEqual(@as(u64, 0), limiter.stats.untracked);
|
||||
try testing.expectEqual(@as(u32, 1), limiter.trackedClients());
|
||||
|
||||
const other = (try address.NetAddress.parse("fd00::40")).key();
|
||||
try testing.expect(limiter.check(start, other));
|
||||
try testing.expectEqual(@as(u32, 2), limiter.trackedClients());
|
||||
|
||||
// The next window admits the refused client again.
|
||||
const next: std.Io.Timestamp = .{ .nanoseconds = start.nanoseconds + 60 * std.time.ns_per_s };
|
||||
try testing.expect(limiter.check(next, mapped));
|
||||
try testing.expectEqual(@as(u64, 1), limiter.stats.refused);
|
||||
}
|
||||
@@ -0,0 +1,468 @@
|
||||
//! `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. 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 db = @import("../db.zig");
|
||||
|
||||
/// One `query_log` row. The logger applies the privacy transforms of PLAN
|
||||
/// §11.4 before it builds this, so `domain` and `client_ip` are already
|
||||
/// whatever the operator agreed to store.
|
||||
pub const Row = struct {
|
||||
timestamp: i64,
|
||||
domain: []const u8,
|
||||
client_ip: []const u8,
|
||||
qtype: ?u16,
|
||||
blocked: bool,
|
||||
block_reason: ?[]const u8,
|
||||
response_time_us: ?i64,
|
||||
cache_hit: ?bool,
|
||||
upstream: ?[]const u8,
|
||||
};
|
||||
|
||||
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, block_reason,
|
||||
\\ response_time_us, cache_hit, upstream)
|
||||
\\VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)
|
||||
;
|
||||
|
||||
/// 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.
|
||||
pub const BatchWriter = struct {
|
||||
database: *db.Db,
|
||||
insert_domain: db.Stmt,
|
||||
select_domain: db.Stmt,
|
||||
insert_row: db.Stmt,
|
||||
|
||||
pub fn init(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();
|
||||
const insert_row = try database.prepare(insert_row_sql);
|
||||
return .{
|
||||
.database = database,
|
||||
.insert_domain = insert_domain,
|
||||
.select_domain = select_domain,
|
||||
.insert_row = insert_row,
|
||||
};
|
||||
}
|
||||
|
||||
pub fn deinit(self: *BatchWriter) void {
|
||||
self.insert_row.deinit();
|
||||
self.select_domain.deinit();
|
||||
self.insert_domain.deinit();
|
||||
}
|
||||
|
||||
/// One transaction for the whole batch. 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;
|
||||
|
||||
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 tx.commit();
|
||||
}
|
||||
|
||||
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 stmt.bindTextOrNull(6, row.block_reason);
|
||||
try bindIntOrNull(stmt, 7, row.response_time_us);
|
||||
try bindIntOrNull(stmt, 8, if (row.cache_hit) |v| @as(i64, @intFromBool(v)) else null);
|
||||
try stmt.bindTextOrNull(9, row.upstream);
|
||||
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.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);
|
||||
}
|
||||
|
||||
/// Deletes every `query_log` row strictly older than `cutoff_ts` and returns
|
||||
/// how many went.
|
||||
///
|
||||
/// Orphaned `domains` rows stay: it is a dimension table, re-interning a name
|
||||
/// costs one indexed insert, and §11.3 asks for no collection.
|
||||
pub fn pruneOlderThan(database: *db.Db, cutoff_ts: i64) db.Error!i64 {
|
||||
var stmt = try database.prepare("DELETE FROM query_log WHERE timestamp < ?1");
|
||||
defer stmt.deinit();
|
||||
try stmt.bindInt(1, cutoff_ts);
|
||||
try stmt.exec();
|
||||
return database.changes();
|
||||
}
|
||||
|
||||
/// `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");
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// tests
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
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,
|
||||
.blocked = false,
|
||||
.block_reason = null,
|
||||
.response_time_us = 1200,
|
||||
.cache_hit = false,
|
||||
.upstream = "9.9.9.9",
|
||||
};
|
||||
}
|
||||
|
||||
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 "writeBatch inserts every row and interns each domain once" {
|
||||
var database = try openLog();
|
||||
defer database.close();
|
||||
var writer = try BatchWriter.init(&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(&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 "nullable columns round-trip a value and a null" {
|
||||
var database = try openLog();
|
||||
defer database.close();
|
||||
var writer = try BatchWriter.init(&database);
|
||||
defer writer.deinit();
|
||||
|
||||
try writer.writeBatch(&.{
|
||||
.{
|
||||
.timestamp = 10,
|
||||
.domain = "blocked.example",
|
||||
.client_ip = "2001:db8::1",
|
||||
.qtype = 28,
|
||||
.blocked = true,
|
||||
.block_reason = "blocklist",
|
||||
.response_time_us = 42,
|
||||
.cache_hit = true,
|
||||
.upstream = "dns.example",
|
||||
},
|
||||
.{
|
||||
.timestamp = 11,
|
||||
.domain = "quiet.example",
|
||||
.client_ip = "hidden",
|
||||
.qtype = null,
|
||||
.blocked = false,
|
||||
.block_reason = null,
|
||||
.response_time_us = null,
|
||||
.cache_hit = null,
|
||||
.upstream = null,
|
||||
},
|
||||
});
|
||||
|
||||
var stmt = try database.prepare(
|
||||
\\SELECT d.domain, q.client_ip, q.qtype, q.blocked, q.block_reason,
|
||||
\\ q.response_time_us, q.cache_hit, q.upstream
|
||||
\\ FROM query_log q JOIN domains d ON d.id = q.domain_id
|
||||
\\ ORDER BY q.timestamp
|
||||
);
|
||||
defer stmt.deinit();
|
||||
|
||||
try testing.expect(try stmt.step());
|
||||
try testing.expectEqualStrings("blocked.example", stmt.columnText(0));
|
||||
try testing.expectEqualStrings("2001:db8::1", stmt.columnText(1));
|
||||
try testing.expectEqual(@as(i64, 28), stmt.columnInt(2));
|
||||
try testing.expect(stmt.columnBool(3));
|
||||
try testing.expectEqualStrings("blocklist", stmt.columnText(4));
|
||||
try testing.expectEqual(@as(i64, 42), stmt.columnInt(5));
|
||||
try testing.expect(stmt.columnBool(6));
|
||||
try testing.expectEqualStrings("dns.example", stmt.columnText(7));
|
||||
|
||||
try testing.expect(try stmt.step());
|
||||
try testing.expectEqualStrings("quiet.example", stmt.columnText(0));
|
||||
try testing.expectEqualStrings("hidden", stmt.columnText(1));
|
||||
try testing.expect(stmt.isNull(2));
|
||||
try testing.expect(!stmt.columnBool(3));
|
||||
try testing.expect(stmt.isNull(4));
|
||||
try testing.expect(stmt.isNull(5));
|
||||
try testing.expect(stmt.isNull(6));
|
||||
try testing.expect(stmt.isNull(7));
|
||||
|
||||
try testing.expect(!try stmt.step());
|
||||
}
|
||||
|
||||
test "an empty batch writes nothing and opens no transaction" {
|
||||
var database = try openLog();
|
||||
defer database.close();
|
||||
var writer = try BatchWriter.init(&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(&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));
|
||||
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));
|
||||
}
|
||||
|
||||
test "pruneOlderThan leaves the domains dimension table intact" {
|
||||
var database = try openLog();
|
||||
defer database.close();
|
||||
var writer = try BatchWriter.init(&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));
|
||||
|
||||
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(&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(&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(&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));
|
||||
try checkpointTruncate(&database);
|
||||
try vacuum(&database);
|
||||
|
||||
try testing.expectEqual(@as(i64, 1), try countRows(&database));
|
||||
try testing.expectEqual(@as(i64, 2), try countDomains(&database));
|
||||
}
|
||||
@@ -0,0 +1,284 @@
|
||||
//! Query-log retention (PLAN §11.5): a daily pass over `querylog.db` that
|
||||
//! deletes rows older than `logging.retention_days`, truncates the WAL, and
|
||||
//! rewrites the file on every seventh pass.
|
||||
//!
|
||||
//! The pass touches `querylog.db` only. §3.6 walls `config.db` off from
|
||||
//! retention churn, and the `hand_edited=0` client rows of §7.2 are pruned by
|
||||
//! whatever creates them, which is Phase 7.
|
||||
//!
|
||||
//! Nothing here retries within a pass. A failed step logs at `warn` and the
|
||||
//! next pass, a day later, does the same work again against the same data.
|
||||
|
||||
const std = @import("std");
|
||||
|
||||
const db = @import("db.zig");
|
||||
const model = @import("../config/model.zig");
|
||||
const queries_repo = @import("repositories/queries_repo.zig");
|
||||
|
||||
const log = std.log.scoped(.retention);
|
||||
|
||||
/// A full `VACUUM` rewrites the whole database file. On the SD card of a
|
||||
/// household box that is the most expensive write this program makes, so it
|
||||
/// runs on every seventh pass rather than every night.
|
||||
pub const vacuum_every_passes = 7;
|
||||
|
||||
/// One day. `retention_days` is the finest granularity the configuration
|
||||
/// expresses, so a finer schedule would prune nothing new.
|
||||
pub const pass_interval_s = 86_400;
|
||||
|
||||
pub const Stats = struct {
|
||||
passes: u64 = 0,
|
||||
rows_pruned: u64 = 0,
|
||||
checkpoints: u64 = 0,
|
||||
vacuums: u64 = 0,
|
||||
};
|
||||
|
||||
pub const Retention = struct {
|
||||
cfg: model.Logging,
|
||||
stats: Stats,
|
||||
|
||||
pub fn init(cfg: model.Logging) Retention {
|
||||
return .{ .cfg = cfg, .stats = .{} };
|
||||
}
|
||||
|
||||
/// One pass: prune, checkpoint, and on every seventh pass vacuum.
|
||||
///
|
||||
/// The three steps are independent. A failed prune does not skip the
|
||||
/// checkpoint, because the WAL that the checkpoint truncates was filled by
|
||||
/// the query logger rather than by this pass.
|
||||
///
|
||||
/// Every failure is a database error, and every database error logs at
|
||||
/// `warn` and leaves the pass counted as done: a pass that returned early
|
||||
/// on the first failure would still be a day away from its retry.
|
||||
///
|
||||
/// `database` must be a connection no other task uses; see `run`.
|
||||
pub fn runOnce(self: *Retention, io: std.Io, database: *db.Db) void {
|
||||
self.stats.passes += 1;
|
||||
const cutoff = std.Io.Clock.real.now(io).toSeconds() - model.retentionSeconds(self.cfg);
|
||||
|
||||
if (queries_repo.pruneOlderThan(database, cutoff)) |deleted| {
|
||||
self.stats.rows_pruned += @intCast(deleted);
|
||||
} else |err| {
|
||||
log.warn("retention prune before {d} failed: {s}", .{ cutoff, @errorName(err) });
|
||||
}
|
||||
|
||||
if (queries_repo.checkpointTruncate(database)) {
|
||||
self.stats.checkpoints += 1;
|
||||
} else |err| {
|
||||
log.warn("retention checkpoint failed: {s}", .{@errorName(err)});
|
||||
}
|
||||
|
||||
if (self.stats.passes % vacuum_every_passes != 0) return;
|
||||
if (queries_repo.vacuum(database)) {
|
||||
self.stats.vacuums += 1;
|
||||
} else |err| {
|
||||
log.warn("retention vacuum failed: {s}", .{@errorName(err)});
|
||||
}
|
||||
}
|
||||
|
||||
/// Daily loop, first pass immediately. Phase 7 starts it.
|
||||
///
|
||||
/// `boot` rather than `awake`: a box that suspends overnight must still see
|
||||
/// its day elapse.
|
||||
///
|
||||
/// `database` must be a connection dedicated to retention: no other task
|
||||
/// may use the same handle while this loop runs. `FULLMUTEX` (`db.zig:218`)
|
||||
/// serializes one SQLite call against another, but a transaction is
|
||||
/// connection state, not call state. On a handle shared with the query
|
||||
/// logger's writer, a prune that lands between that writer's BEGIN and
|
||||
/// COMMIT runs inside the writer's transaction and commits or rolls back
|
||||
/// with the batch, and a checkpoint or a `VACUUM` can land inside a
|
||||
/// transaction that is still open.
|
||||
///
|
||||
/// Retention takes `database` per call and opens nothing itself; Phase 7
|
||||
/// opens the second connection. Isolation across the two connections is
|
||||
/// SQLite's own — WAL plus the `busy_timeout` of `db.zig`'s open options —
|
||||
/// so a pass that still loses a race sees `error.Busy` or `error.Locked`,
|
||||
/// logs at `warn`, and repeats the work on the next interval.
|
||||
pub fn run(self: *Retention, io: std.Io, database: *db.Db) std.Io.Cancelable!void {
|
||||
const interval: std.Io.Clock.Duration = .{
|
||||
.raw = .fromSeconds(pass_interval_s),
|
||||
.clock = .boot,
|
||||
};
|
||||
while (true) {
|
||||
self.runOnce(io, database);
|
||||
try interval.sleep(io);
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// tests
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
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 writeRows(database: *db.Db, timestamps: []const i64) !void {
|
||||
var writer = try queries_repo.BatchWriter.init(database);
|
||||
defer writer.deinit();
|
||||
var rows: [8]queries_repo.Row = undefined;
|
||||
for (timestamps, rows[0..timestamps.len]) |timestamp, *row| {
|
||||
row.* = .{
|
||||
.timestamp = timestamp,
|
||||
.domain = "example.com",
|
||||
.client_ip = "192.0.2.10",
|
||||
.qtype = 1,
|
||||
.blocked = false,
|
||||
.block_reason = null,
|
||||
.response_time_us = null,
|
||||
.cache_hit = null,
|
||||
.upstream = null,
|
||||
};
|
||||
}
|
||||
try writer.writeBatch(rows[0..timestamps.len]);
|
||||
}
|
||||
|
||||
test "a pass prunes the rows past the retention window and keeps the rest" {
|
||||
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
|
||||
var database = try openLog();
|
||||
defer database.close();
|
||||
|
||||
const now = std.Io.Clock.real.now(io).toSeconds();
|
||||
const day = 86_400;
|
||||
try writeRows(&database, &.{ now - 40 * day, now - 31 * day, now - 29 * day, now - 60 });
|
||||
|
||||
var retention: Retention = .init(.{ .retention_days = 30 });
|
||||
retention.runOnce(io, &database);
|
||||
|
||||
try testing.expectEqual(@as(i64, 2), try queries_repo.countRows(&database));
|
||||
try testing.expectEqual(@as(u64, 1), retention.stats.passes);
|
||||
try testing.expectEqual(@as(u64, 2), retention.stats.rows_pruned);
|
||||
try testing.expectEqual(@as(u64, 1), retention.stats.checkpoints);
|
||||
try testing.expectEqual(@as(u64, 0), retention.stats.vacuums);
|
||||
}
|
||||
|
||||
test "the cutoff follows retention_days" {
|
||||
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
|
||||
var database = try openLog();
|
||||
defer database.close();
|
||||
|
||||
const now = std.Io.Clock.real.now(io).toSeconds();
|
||||
const day = 86_400;
|
||||
// The same row is inside the window of one configuration and outside the
|
||||
// window of the other.
|
||||
try writeRows(&database, &.{now - 3 * day});
|
||||
|
||||
var keeps: Retention = .init(.{ .retention_days = 7 });
|
||||
keeps.runOnce(io, &database);
|
||||
try testing.expectEqual(@as(i64, 1), try queries_repo.countRows(&database));
|
||||
try testing.expectEqual(@as(u64, 0), keeps.stats.rows_pruned);
|
||||
|
||||
var prunes: Retention = .init(.{ .retention_days = 1 });
|
||||
prunes.runOnce(io, &database);
|
||||
try testing.expectEqual(@as(i64, 0), try queries_repo.countRows(&database));
|
||||
try testing.expectEqual(@as(u64, 1), prunes.stats.rows_pruned);
|
||||
}
|
||||
|
||||
test "the seventh pass vacuums and the six before it do not" {
|
||||
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
|
||||
var database = try openLog();
|
||||
defer database.close();
|
||||
|
||||
var retention: Retention = .init(.{});
|
||||
for (0..6) |_| {
|
||||
retention.runOnce(io, &database);
|
||||
try testing.expectEqual(@as(u64, 0), retention.stats.vacuums);
|
||||
}
|
||||
retention.runOnce(io, &database);
|
||||
|
||||
try testing.expectEqual(@as(u64, 7), retention.stats.passes);
|
||||
try testing.expectEqual(@as(u64, 1), retention.stats.vacuums);
|
||||
try testing.expectEqual(@as(u64, 7), retention.stats.checkpoints);
|
||||
|
||||
for (0..7) |_| retention.runOnce(io, &database);
|
||||
try testing.expectEqual(@as(u64, 14), retention.stats.passes);
|
||||
try testing.expectEqual(@as(u64, 2), retention.stats.vacuums);
|
||||
}
|
||||
|
||||
test "a pass over an empty database still counts" {
|
||||
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
|
||||
var database = try openLog();
|
||||
defer database.close();
|
||||
|
||||
var retention: Retention = .init(.{});
|
||||
retention.runOnce(io, &database);
|
||||
|
||||
try testing.expectEqual(@as(u64, 1), retention.stats.passes);
|
||||
try testing.expectEqual(@as(u64, 0), retention.stats.rows_pruned);
|
||||
try testing.expectEqual(@as(u64, 1), retention.stats.checkpoints);
|
||||
try testing.expectEqual(@as(i64, 0), try queries_repo.countRows(&database));
|
||||
}
|
||||
|
||||
test "a failing prune counts the pass and leaves the rows alone" {
|
||||
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
|
||||
var database = try openLog();
|
||||
defer database.close();
|
||||
|
||||
const now = std.Io.Clock.real.now(io).toSeconds();
|
||||
try writeRows(&database, &.{now - 40 * 86_400});
|
||||
try database.exec(
|
||||
\\CREATE TRIGGER refuse_delete BEFORE DELETE ON query_log
|
||||
\\BEGIN SELECT RAISE(ABORT, 'refused'); END;
|
||||
);
|
||||
|
||||
var retention: Retention = .init(.{ .retention_days = 30 });
|
||||
retention.runOnce(io, &database);
|
||||
|
||||
try testing.expectEqual(@as(i64, 1), try queries_repo.countRows(&database));
|
||||
try testing.expectEqual(@as(u64, 1), retention.stats.passes);
|
||||
try testing.expectEqual(@as(u64, 0), retention.stats.rows_pruned);
|
||||
// The checkpoint runs whether or not the prune did.
|
||||
try testing.expectEqual(@as(u64, 1), retention.stats.checkpoints);
|
||||
}
|
||||
|
||||
test "the next pass retries what the failed one could not do" {
|
||||
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
|
||||
var database = try openLog();
|
||||
defer database.close();
|
||||
|
||||
const now = std.Io.Clock.real.now(io).toSeconds();
|
||||
try writeRows(&database, &.{ now - 40 * 86_400, now - 39 * 86_400 });
|
||||
try database.exec(
|
||||
\\CREATE TRIGGER refuse_delete BEFORE DELETE ON query_log
|
||||
\\BEGIN SELECT RAISE(ABORT, 'refused'); END;
|
||||
);
|
||||
|
||||
var retention: Retention = .init(.{ .retention_days = 30 });
|
||||
retention.runOnce(io, &database);
|
||||
try testing.expectEqual(@as(i64, 2), try queries_repo.countRows(&database));
|
||||
|
||||
try database.exec("DROP TRIGGER refuse_delete;");
|
||||
retention.runOnce(io, &database);
|
||||
|
||||
try testing.expectEqual(@as(i64, 0), try queries_repo.countRows(&database));
|
||||
try testing.expectEqual(@as(u64, 2), retention.stats.passes);
|
||||
try testing.expectEqual(@as(u64, 2), retention.stats.rows_pruned);
|
||||
}
|
||||
Reference in New Issue
Block a user