//! 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 querylog_versions = @import("querylog_versions.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(testing.allocator, 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, .qclass = 1, .rcode = 0, .blocked = false, .response_time_us = null, .cache_hit = null, .upstream = null, .group_id = 1, .group_name = "default", .policy_action = .allow, .policy_reason = .no_match, .matched = null, .source_id = null, .source_name = null, .cname_target = null, .safe_search_target = null, .route_kind = .upstream, .forward_zone = null, }; } 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, testing.allocator, log_db.database(), @as(?*disk_monitor.Monitor, null), }); // Declared after the `log_db.deinit` defer so LIFO stops the writer first: // an early return below would otherwise close the database under a live // writer and leave the never-closed queue parking it forever. defer { query_log.shutdown(io); future.await(io) catch {}; } 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_versions.current_version), 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; // A one-second window, not the 60-second default: this case is about the // entry reaching the file on the interval alone, with nothing to close the // queue for it. const flush_interval_s = 1; var query_log: logger.Logger = .init(.{ .query_log_flush_interval_s = flush_interval_s }, &queue_buf); var future = try io.concurrent(logger.Logger.runWriter, .{ &query_log, io, testing.allocator, log_db.database(), @as(?*disk_monitor.Monitor, null), }); // Declared after the `log_db.deinit` defer so LIFO stops the writer first: // an early return below would otherwise close the database under a live // writer and leave the never-closed queue parking it forever. defer { query_log.shutdown(io); future.await(io) catch {}; } query_log.log(io, entryAt(1, "only.example")); // Three 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 = 3 * flush_interval_s * 1000 / 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; // Zero interval: this case is about the queue cap and the gate, so the // writer takes what is queued and goes straight to the gate. var query_log: logger.Logger = .init(.{ .query_log_flush_interval_s = 0 }, &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, testing.allocator, log_db.database(), @as(?*disk_monitor.Monitor, &monitor), }); // Declared after the `log_db.deinit` defer so LIFO stops the writer first: // an early return below would otherwise close the database under a live // writer and leave the never-closed queue parking it forever. defer { query_log.shutdown(io); future.await(io) catch {}; } 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 still held by the gate counts its // batch as dropped rather than writing it, which is a different case (the // logger's own suite covers 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, testing.allocator, log_db.database(), @as(?*disk_monitor.Monitor, null), }); // Declared after the `log_db.deinit` defer so LIFO stops the writer first: // an early return below would otherwise close the database under a live // writer and leave the never-closed queue parking it forever. defer { query_log.shutdown(io); future.await(io) catch {}; } 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_days: retention.RetentionDays = .init(30); var pass: retention.Retention = .init(&pass_days); pass.runOnce(io, log_db.database(), null, null); try testing.expectEqual(@as(u64, 1), pass.snapshotStats().passes); try testing.expectEqual(@as(u64, 2), pass.snapshotStats().rows_pruned); try testing.expectEqual(@as(u64, 1), pass.snapshotStats().checkpoints); try testing.expectEqual(@as(u64, 0), pass.snapshotStats().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, null, 0); 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; // Zero interval: the case is the gate's hold and release, so the writer // takes the five queued entries straight to it. var query_log: logger.Logger = .init(.{ .query_log_flush_interval_s = 0 }, &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, testing.allocator, log_db.database(), @as(?*disk_monitor.Monitor, &monitor), }); // Declared after the `log_db.deinit` defer so LIFO stops the writer first: // an early return below would otherwise close the database under a live // writer and leave the never-closed queue parking it forever. defer { query_log.shutdown(io); future.await(io) catch {}; } 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.setThresholds(.{ .min_free_mb = 0, .warn_free_mb = 0 }); monitor.sample(io, null, 0); 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, .sized(16)); 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); 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); }