nxdns/src/storage/logger.zig

//! 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 in `transformed`, before the entry is
//! enqueued, so nothing downstream — the database or the event stream — can
//! observe a value the operator asked to hide. `log` is the two halves in
//! order; `QuerySink` calls them separately so both of its consumers see the
//! one transformed entry.
//!
//! One transaction per `logging.query_log_flush_interval_s`, not one per query.
//! At household rates a per-query commit costs orders of magnitude more disk
//! writes than the rows are worth — a minute of batching is what keeps an SD
//! card alive. The window is also what a crash costs: while the writer is
//! healthy and the disk gate is open, a process that dies loses about one
//! interval of query history. It is not a ceiling — a batch the gate is
//! holding, or one waiting on a write lock, is older than that — and nothing
//! here promises one, because query history is the cheapest data on the box.
//!
//! 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 until it recovers — or, if the process is
//! already stopping, drops what it holds. 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 builtin = @import("builtin");

const db = @import("db.zig");
const disk_monitor = @import("disk_monitor.zig");
const events = @import("events.zig");
const limits = @import("../config/limits.zig");
const model = @import("../config/model.zig");
const provenance = @import("provenance.zig");
const queries_repo = @import("repositories/queries_repo.zig");
const regex = @import("../filter/regex.zig");
const safe_url = @import("../safe_url.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);

/// Rows per transaction. Comptime: the window, not the batch size, is what an
/// operator has a reason to move, and a batch this size already amortizes the
/// commit at any household rate.
pub const flush_batch = 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;

/// The widths of the query log's text columns, and the single source of them:
/// every producer that formats into one of these fields sizes its own buffer
/// from the constant here, so nothing can format wider than the row stores.
pub const max_domain_len = 253;
/// RFC 5952 text of any IPv6 address, zone identifier included.
pub const max_client_len = 45;

/// `matched` holds the rule that decided the query, and the widest rule the
/// configuration accepts is a regex pattern at `regex.max_pattern_len`. It does
/// not fit a `u8` length, which is why this one field carries a `u16`.
pub const max_matched_len = regex.max_pattern_len;

/// The redacted resolver identity of the exchange that actually happened. Two
/// bounds apply and the buffer takes the larger, so neither producer truncates:
/// the longest well-formed `scheme://host:port` with a maximal host, and
/// `safe_url.redact`'s own output bound (`max_len` plus the `...` it appends
/// when it truncates).
pub const max_upstream_len = @max(
    "https://".len + max_domain_len + ":65535".len,
    safe_url.max_len + 3,
);

/// `cname_target`, `safe_search_target` and `forward_zone` each hold a domain
/// name, so they are all the same width as `domain`.
const max_name_len = max_domain_len;

/// One row on its way to `query_log`, carrying its own bytes.
///
/// Every field is by value: `Io.Queue` copies elements as raw bytes, so nothing
/// here may borrow from the query that produced it. The buffer widths above are
/// therefore the row's real storage cost, multiplied by the queue capacity —
/// see `query_log_buffer_max`.
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,
    qclass: u16,
    /// The client-visible RCODE. Twelve bits, not four: an EDNS extended code
    /// carries eight more bits in the OPT record than the header's four. The
    /// type is the enforcement — the column's `CHECK` in `querylog_schema.ddl`
    /// bounds the same value against every other writer of the file.
    rcode: u12,
    blocked: bool,
    response_time_us: ?i64,
    cache_hit: ?bool,
    upstream_buf: [max_upstream_len]u8,
    upstream_len: u16,

    group_id: ?i64,
    group_buf: [limits.max_group_name_len]u8,
    group_len: u8,
    policy_action: provenance.PolicyAction,
    policy_reason: provenance.PolicyReason,
    matched_buf: [max_matched_len]u8,
    matched_len: u16,
    source_id: ?i64,
    source_buf: [limits.max_source_name_len]u8,
    source_len: u8,
    cname_buf: [max_name_len]u8,
    cname_len: u8,
    safe_search_buf: [max_name_len]u8,
    safe_search_len: u8,
    route_kind: provenance.RouteKind,
    forward_zone_buf: [max_name_len]u8,
    forward_zone_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.
    ///
    /// Every text field defaults to `""`, which reaches a nullable column as
    /// NULL. The three fields with no sensible empty value — the two enums and
    /// the route — default to what a query the pipeline has not yet explained
    /// would honestly say about itself.
    pub const Fields = struct {
        timestamp: i64,
        domain: []const u8,
        client_ip: []const u8,
        qtype: ?u16 = null,
        qclass: u16 = 0,
        rcode: u12 = 0,
        blocked: bool = false,
        response_time_us: ?i64 = null,
        cache_hit: ?bool = null,
        /// Empty means "no upstream was attempted", which reaches the database
        /// as NULL. Already redacted by the caller.
        upstream: []const u8 = "",

        group_id: ?i64 = null,
        group_name: []const u8 = "",
        policy_action: provenance.PolicyAction = .not_evaluated,
        policy_reason: provenance.PolicyReason = .no_match,
        matched: []const u8 = "",
        source_id: ?i64 = null,
        source_name: []const u8 = "",
        cname_target: []const u8 = "",
        safe_search_target: []const u8 = "",
        route_kind: provenance.RouteKind = .upstream,
        forward_zone: []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,
            .qclass = f.qclass,
            .rcode = f.rcode,
            .blocked = f.blocked,
            .response_time_us = f.response_time_us,
            .cache_hit = f.cache_hit,
            .upstream_buf = undefined,
            .upstream_len = 0,
            .group_id = f.group_id,
            .group_buf = undefined,
            .group_len = 0,
            .policy_action = f.policy_action,
            .policy_reason = f.policy_reason,
            .matched_buf = undefined,
            .matched_len = 0,
            .source_id = f.source_id,
            .source_buf = undefined,
            .source_len = 0,
            .cname_buf = undefined,
            .cname_len = 0,
            .safe_search_buf = undefined,
            .safe_search_len = 0,
            .route_kind = f.route_kind,
            .forward_zone_buf = undefined,
            .forward_zone_len = 0,
        };
        entry.setDomain(f.domain);
        entry.setClientIp(f.client_ip);
        copyInto(&entry.upstream_buf, &entry.upstream_len, f.upstream);
        copyInto(&entry.group_buf, &entry.group_len, f.group_name);
        entry.setMatched(f.matched);
        copyInto(&entry.source_buf, &entry.source_len, f.source_name);
        entry.setCnameTarget(f.cname_target);
        entry.setSafeSearchTarget(f.safe_search_target);
        copyInto(&entry.forward_zone_buf, &entry.forward_zone_len, f.forward_zone);
        return entry;
    }

    pub fn setDomain(self: *Entry, value: []const u8) void {
        copyInto(&self.domain_buf, &self.domain_len, value);
    }

    pub fn setClientIp(self: *Entry, value: []const u8) void {
        copyInto(&self.client_buf, &self.client_len, value);
    }

    pub fn setMatched(self: *Entry, value: []const u8) void {
        copyInto(&self.matched_buf, &self.matched_len, value);
    }

    pub fn setCnameTarget(self: *Entry, value: []const u8) void {
        copyInto(&self.cname_buf, &self.cname_len, value);
    }

    pub fn setSafeSearchTarget(self: *Entry, value: []const u8) void {
        copyInto(&self.safe_search_buf, &self.safe_search_len, 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 upstream(self: *const Entry) []const u8 {
        return self.upstream_buf[0..self.upstream_len];
    }

    pub fn groupName(self: *const Entry) []const u8 {
        return self.group_buf[0..self.group_len];
    }

    pub fn matched(self: *const Entry) []const u8 {
        return self.matched_buf[0..self.matched_len];
    }

    pub fn sourceName(self: *const Entry) []const u8 {
        return self.source_buf[0..self.source_len];
    }

    pub fn cnameTarget(self: *const Entry) []const u8 {
        return self.cname_buf[0..self.cname_len];
    }

    pub fn safeSearchTarget(self: *const Entry) []const u8 {
        return self.safe_search_buf[0..self.safe_search_len];
    }

    pub fn forwardZone(self: *const Entry) []const u8 {
        return self.forward_zone_buf[0..self.forward_zone_len];
    }
};

/// The memory budget the queue is allowed to occupy. `Entry` travels by value,
/// so the composition root allocates `query_log_buffer_max` of them in full at
/// boot (`app.zig`) and the SSE hub embeds a ring of them per subscriber.
const queue_budget_bytes = 64 * 1024 * 1024;

/// The ceiling `config/validate.zig` enforces on `logging.query_log_buffer_max`,
/// derived from the width of `Entry` rather than picked.
///
/// The provenance columns of milestone 28 roughly tripled `Entry`, so the bound
/// that matters is bytes, not entries: an operator who asks for a million
/// entries is asking for well over a gigabyte of queue. This is a sanity bound,
/// not a memory-fit guarantee — what actually fits depends on the box.
pub const query_log_buffer_max: u32 = @intCast(queue_budget_bytes / @sizeOf(Entry));

/// Copies as much of `value` as `buf` holds, and stores the length through
/// `len`. `len`'s type never bounds anything — `buf.len` does — so the same
/// helper serves the `u8` fields and the `u16` ones.
fn copyInto(buf: []u8, len: anytype, value: []const u8) void {
    const n = @min(buf.len, value.len);
    @memcpy(buf[0..n], value[0..n]);
    len.* = @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,
        .qclass = entry.qclass,
        .rcode = entry.rcode,
        .blocked = entry.blocked,
        .response_time_us = entry.response_time_us,
        .cache_hit = entry.cache_hit,
        .upstream = emptyAsNull(entry.upstream()),
        .group_id = entry.group_id,
        .group_name = emptyAsNull(entry.groupName()),
        .policy_action = entry.policy_action,
        .policy_reason = entry.policy_reason,
        .matched = emptyAsNull(entry.matched()),
        .source_id = entry.source_id,
        .source_name = emptyAsNull(entry.sourceName()),
        .cname_target = emptyAsNull(entry.cnameTarget()),
        .safe_search_target = emptyAsNull(entry.safeSearchTarget()),
        .route_kind = entry.route_kind,
        .forward_zone = emptyAsNull(entry.forwardZone()),
    };
}

fn emptyAsNull(value: []const u8) ?[]const u8 {
    return if (value.len == 0) null else value;
}

const EntryQueue = std.Io.Queue(Entry);

/// A database failure is not in here: a batch the database refuses is dropped,
/// counted and reported where it happens, and the writer carries on.
/// `GatedAtShutdown` is the one condition `flush` cannot settle by itself —
/// the disk gate is shut and the process is stopping, so the batch in hand is
/// lost and so is everything still queued behind it. Only `runWriter` can see
/// both, so it does the counting and files the single episode.
pub const FlushError = std.Io.Cancelable || error{GatedAtShutdown};

/// 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,
};

/// Where the disk gate stands right now, as a state rather than a tally.
///
/// `open` is the steady state. `gated` says the gate is holding writes back but
/// nothing has been lost to it yet, and `losing` says this episode has already
/// cost rows. A cumulative drop counter cannot say any of that: it only ever
/// grows, so a rollup computed from it would latch on the first overflow.
pub const GateEpisode = enum(u8) { open, gated, losing };

/// The episode state and the identity of the episode it belongs to, in one
/// word so a compare-and-swap can test both at once.
///
/// The identity is what the state alone cannot carry. A producer decides a row
/// is lost, stalls, and wakes after the gate has closed, reopened and closed
/// again; a bare `gated -> losing` swap would then mark an episode that has
/// cost nothing. `generation` rises every time an episode opens, so that swap
/// fails against the newer episode and the stale loss is discarded.
const Gate = packed struct(u64) {
    episode: GateEpisode,
    generation: u56,

    const initial: Gate = .{ .episode = .open, .generation = 0 };

    fn bits(self: Gate) u64 {
        return @bitCast(self);
    }

    fn of(bits_value: u64) Gate {
        return @bitCast(bits_value);
    }
};

/// Parks a producer immediately before it evicts an entry, so a test can move
/// the disk gate while that producer is still inside `enqueue` and the row it
/// is about to lose is still queued.
///
/// Nothing else reaches that point. `enqueue` never suspends, so from outside a
/// gate read before its loop and one read at the eviction give the same answer,
/// and no test could tell the two apart — which is exactly the difference that
/// decides whether a discard can join an episode that opened after the producer
/// started.
///
/// Before the eviction and not after it: parking after the row is already gone
/// would let a test open an episode and then file an earlier loss under it,
/// which is the misattribution the sampling rule exists to prevent. The loss
/// has to happen inside the episode for the episode to own it. The storage
/// exists in a test build only, and `park` reduces to nothing everywhere else —
/// the settings hash seam's shape (`web/handlers/settings.zig`).
const discard_stall = if (builtin.is_test) struct {
    var armed: bool = false;
    var parked: std.Io.Event = .unset;
    var release: std.Io.Event = .unset;

    fn park(io: std.Io) void {
        if (!armed) return;
        parked.set(io);
        release.waitUncancelable(io);
    }
} else struct {
    fn park(io: std.Io) void {
        _ = io;
    }
};

/// The §11.4 privacy policy: one value, never two. A producer decides the
/// domain fields and the client field from ONE load of `privacy_packed`, so no
/// entry can leave `transformed` with the domain redacted and the client
/// exposed, or the reverse, because a `setPrivacy` landed between the two
/// decisions.
pub const Privacy = packed struct(u8) {
    hide_domains: bool = false,
    hide_client_ips: bool = false,
    _reserved: u6 = 0,
};

pub const Logger = struct {
    /// Both privacy flags in one atomic byte, loaded once per entry.
    privacy_packed: std.atomic.Value(u8),
    /// Independent of the privacy policy: it governs when a batch commits, not
    /// what a row contains, so nothing pairs the two.
    flush_interval_s: std.atomic.Value(u16),
    queue: EntryQueue,
    queries_dropped: std.atomic.Value(u64),
    /// When the newest drop happened, in unix seconds; 0 means none yet. Read
    /// through `lastDropSeconds`, which is what turns the sentinel into a null.
    last_drop_s: std.atomic.Value(i64),
    rows_written: std.atomic.Value(u64),
    batches_gated: std.atomic.Value(u64),
    /// The gating episode `/api/health` reports as `query_history.losing`, as
    /// `Gate` bits. Moved only by `gateHolds`/`gateReopened`, which the writer
    /// calls as it observes the monitor, and raised to `losing` by a drop that
    /// carries the identity of the episode still holding.
    ///
    /// Read it through `gateEpisode` or `sampleGate`, never as a raw integer.
    gate: 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),
    /// Set by `shutdown`, read by the disk gate. The gate holds a batch for as
    /// long as the disk stays critical, which is right while the process is
    /// running and a hang once it is stopping — nothing will ever release it.
    draining: std.atomic.Value(bool),
    /// Wired by the composition root after `init`, following the
    /// `gate: ?*disk_monitor.Monitor` idiom. Null in every unit test here.
    diagnostics: ?*events.Store = null,

    /// `queue_buf.len` is the backpressure cap — the composition root
    /// (`app.zig:311`) allocates `cfg.logging.query_log_buffer_max` entries,
    /// which `config/validate.zig` bounds. 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 .{
            .privacy_packed = .init(@bitCast(Privacy{
                .hide_domains = cfg.hide_domains,
                .hide_client_ips = cfg.hide_client_ips,
            })),
            .flush_interval_s = .init(cfg.query_log_flush_interval_s),
            .queue = .init(queue_buf),
            .queries_dropped = .init(0),
            .last_drop_s = .init(0),
            .rows_written = .init(0),
            .batches_gated = .init(0),
            .gate = .init(Gate.initial.bits()),
            .writer_failed = .init(false),
            .draining = .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 {
        self.logTransformed(io, self.transformed(entry));
    }

    /// The §11.4 privacy transforms, on their own. `QuerySink` runs them once
    /// and hands the result to every consumer, so nothing downstream — the
    /// database or the event stream — can observe a value the operator asked
    /// to hide.
    /// `hide_domains` covers every field derived from the query name, not just
    /// `domain`: a matched wildcard, a CNAME target and a safe-search target
    /// each name the very thing the operator asked to keep out of the log.
    ///
    /// `forward_zone`, `group_name` and `source_name` stay visible. They are
    /// configuration labels the operator wrote, identical on every row that
    /// hits them, and they say nothing about which name a client looked up.
    pub fn transformed(self: *const Logger, entry: Entry) Entry {
        const policy = self.privacy();
        var out = entry;
        if (policy.hide_domains) {
            out.setDomain(hidden_marker);
            // Only where there is something to hide: an empty field means the
            // query had no such value, and writing a marker would claim it did.
            if (out.matched_len != 0) out.setMatched(hidden_marker);
            if (out.cname_len != 0) out.setCnameTarget(hidden_marker);
            if (out.safe_search_len != 0) out.setSafeSearchTarget(hidden_marker);
        }
        if (policy.hide_client_ips) out.setClientIp(hidden_marker);
        return out;
    }

    /// The live policy, from one load. Every producer decision about one entry
    /// must come from a single call to this.
    pub fn privacy(self: *const Logger) Privacy {
        return @bitCast(self.privacy_packed.load(.monotonic));
    }

    pub fn setPrivacy(self: *Logger, p: Privacy) void {
        self.privacy_packed.store(@bitCast(p), .monotonic);
    }

    pub fn setFlushInterval(self: *Logger, seconds: u16) void {
        self.flush_interval_s.store(seconds, .monotonic);
    }

    /// `log` without the transforms, for a caller that already applied them.
    pub fn logTransformed(self: *Logger, io: std.Io, entry: Entry) void {
        self.enqueue(io, entry);
    }

    /// 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;
            discard_stall.park(io);
            const got = self.queue.get(io, &oldest, 0) catch break;
            // Sampled after the eviction and not before the loop: the row is
            // lost on the line above, and a sample taken while the gate was
            // still open would let an episode that opened during `put` escape
            // being marked for a row it really cost. Reading it here cannot
            // misattribute in the other direction either — a sample the gate
            // outruns fails `countDropped`'s generation check.
            if (got == 1) self.countDropped(io, 1, self.sampleGate());
        }
        self.countDropped(io, 1, self.sampleGate());
    }

    /// 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.
    ///
    /// It must outlive the producers rather than share their lifetime: a
    /// cancellation that races the close decides at random whether the batch in
    /// hand is written or counted as dropped. `app.zig` spawns this outside the
    /// group it cancels for exactly that reason.
    ///
    /// `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);
            // No recovery path claims this episode: the writer is gone for the
            // life of the process, so the row stays active, which is the truth.
            self.reportWrite(io, "writer", "preparing the batch statements failed", @errorName(err), 0);
            self.queue.close(io);
            _ = self.dropRemaining(io, self.sampleGate());
            return;
        };
        defer writer.deinit();
        return self.runPrepared(io, &writer, monitor);
    }

    /// The writer loop over statements someone else prepared.
    ///
    /// `runWriter` prepares and then calls this. A logger generation created by
    /// a resize prepares separately, before anything is published, so that a
    /// statement failure is refused at prepare time instead of silently killing
    /// the writer of a queue producers are already filling
    /// (`logger_controller.zig`).
    pub fn runPrepared(
        self: *Logger,
        io: std.Io,
        writer: *queries_repo.BatchWriter,
        monitor: ?*disk_monitor.Monitor,
    ) std.Io.Cancelable!void {
        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,
            };
            // Before `fill`, not only inside `flush`: `fill` spends the whole
            // flush interval taking entries off the queue, and a producer that
            // overflows the queue during that wait is losing rows to the gate
            // just as surely as the held batch is. Observing here is what makes
            // the episode start cover those drops instead of misfiling them as
            // ordinary overflow.
            const at = self.observeGate(monitor);
            const deadline = self.flushDeadline(io);
            // `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, at) catch |err| {
                self.countDropped(io, n, at);
                return err;
            };
            self.flush(io, writer, batch[0..n], monitor) catch |err| switch (err) {
                error.Canceled => |e| {
                    self.countDropped(io, n, at);
                    return e;
                },
                // Nothing will open the gate now. Everything still queued is
                // lost with the batch in hand, so it is counted here and
                // announced once — a per-chunk report would write to the very
                // disk that is out of space, dozens of times, on the way out.
                error.GatedAtShutdown => {
                    // Re-sampled rather than reusing `at`: `flush` observed the
                    // gate again on its way to this error, so the episode that
                    // is costing these rows is the one holding now.
                    const gated_at = self.sampleGate();
                    self.countDropped(io, n, gated_at);
                    const lost = n + self.dropRemaining(io, gated_at);
                    scope.warn(
                        "query log: {d} rows dropped at shutdown, the disk gate was closed",
                        .{lost},
                    );
                    self.reportWrite(
                        io,
                        "batch",
                        "query log rows were dropped at shutdown",
                        "DiskCritical",
                        lost,
                    );
                    return;
                },
            };
        }
    }

    /// When the batch that starts now must be committed. `.boot` and not
    /// `.awake`: a suspended box would otherwise stretch the window by however
    /// long it slept, and the rows are already in memory waiting.
    ///
    /// An interval of 0 yields a deadline that has already passed, which is
    /// exactly the documented sentinel — `fill` then takes what is queued and
    /// returns without waiting for anything.
    fn flushDeadline(self: *const Logger, io: std.Io) std.Io.Clock.Timestamp {
        return .fromNow(io, .{
            .raw = .fromSeconds(self.flush_interval_s.load(.monotonic)),
            .clock = .boot,
        });
    }

    /// Counts every entry left in a closed queue as dropped, and returns how
    /// many. 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, at: Gate) usize {
        var total: usize = 0;
        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(io, n, at);
            total += n;
        }
        return total;
    }

    /// Closes the queue and tells the writer to stop waiting on anything that
    /// may never arrive. `log` drops from here on, and `runWriter` returns once
    /// it has flushed what was left — a blocked `getOne` on a closed queue
    /// returns immediately, so the interval is never waited out at shutdown.
    ///
    /// The close comes first: a writer that sees `draining` set must be able to
    /// drain the queue to the end, and only a closed queue reports its end.
    ///
    /// Every producer must be stopped and joined before this is called
    /// (`app.zig`): an entry enqueued after the close is a dropped entry.
    pub fn shutdown(self: *Logger, io: std.Io) void {
        self.queue.close(io);
        self.draining.store(true, .release);
    }

    /// `shutdown` without `draining`: this generation is being replaced, not
    /// the process stopped.
    ///
    /// The flag is what turns a gate-held batch into a counted loss
    /// (`flush`'s `GatedAtShutdown`), and a retired writer must not take that
    /// path — the disk can still recover, and the rows it is holding are still
    /// going to be written when it does. Every producer of this generation
    /// must have released it before the close, exactly as at shutdown.
    pub fn retire(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,
        at: Gate,
    ) 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, at) 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,
        at: Gate,
    ) 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(io, 1, at);
            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,
    ) FlushError!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.gateHolds();
                // Waiting for the disk to recover is right while the process
                // runs and wrong once it is stopping: nothing is going to free
                // space during shutdown, so the batch is lost either way and
                // the only choice left is between losing it counted and
                // hanging the exit. The counting and the one report belong to
                // `runWriter`, which knows the rest of the queue is lost too.
                if (self.draining.load(.acquire)) return error.GatedAtShutdown;
                _ = self.batches_gated.fetchAdd(1, .monotonic);
                try pause.sleep(io);
            }
            self.gateReopened();
        }

        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) });
            // Sampled here, so a batch the gate already let through is counted
            // against whatever episode is open now — a database failure is not
            // a gating loss.
            self.countDropped(io, entries.len, self.sampleGate());
            self.reportWrite(io, "batch", "a query log batch was dropped", @errorName(err), entries.len);
            return;
        };
        _ = self.rows_written.fetchAdd(entries.len, .monotonic);
        if (self.diagnostics) |store| {
            store.resolve(io, std.Io.Clock.real.now(io).toSeconds(), .query_log_write, "batch");
        }
    }

    /// An error, not a warning: dropped query rows are gone, and a writer that
    /// never started means every later row is gone too.
    fn reportWrite(
        self: *Logger,
        io: std.Io,
        operation: []const u8,
        message: []const u8,
        error_name: []const u8,
        rows: usize,
    ) void {
        const store = self.diagnostics orelse return;
        var buf: [events.Store.max_detail_len]u8 = undefined;
        const detail = std.fmt.bufPrint(&buf, "{s}: {s} ({d} rows)", .{
            message,
            error_name,
            rows,
        }) catch buf[0..];
        store.report(
            io,
            std.Io.Clock.real.now(io).toSeconds(),
            .query_log_write,
            operation,
            operation,
            .@"error",
            detail,
        );
    }

    /// Counts `n` lost rows, stamps the loss, and raises the episode `at` to
    /// `losing` if that episode is still the current one.
    ///
    /// `at` is sampled where the rows were lost, not here: see `Gate`. A caller
    /// that lost rows outside any episode passes what `sampleGate` gave it and
    /// the raise is simply a no-op.
    ///
    /// The stamp is a second atomic rather than a field beside the count, so a
    /// reader can briefly see the new total against the previous timestamp. The
    /// health contract says so: `dropped_total` above zero with a null
    /// `last_drop_s` is a legal, momentary answer.
    fn countDropped(self: *Logger, io: std.Io, n: usize, at: Gate) void {
        _ = self.queries_dropped.fetchAdd(n, .monotonic);
        self.stampDrop(std.Io.Clock.real.now(io).toSeconds());
        if (at.episode != .gated) return;
        const losing: Gate = .{ .episode = .losing, .generation = at.generation };
        _ = self.gate.cmpxchgStrong(at.bits(), losing.bits(), .acq_rel, .monotonic);
    }

    /// Moves the stamp forward only. Two producers can reach `countDropped` out
    /// of order, and a plain store would let the older one publish its
    /// timestamp over the newer drop's — a `last_drop_s` that walks backwards
    /// while drops are still arriving.
    fn stampDrop(self: *Logger, at_s: i64) void {
        var seen = self.last_drop_s.load(.monotonic);
        while (at_s > seen) {
            seen = self.last_drop_s.cmpxchgWeak(seen, at_s, .monotonic, .monotonic) orelse return;
        }
    }

    /// When the newest drop happened, or null while nothing has been dropped.
    pub fn lastDropSeconds(self: *const Logger) ?i64 {
        const stamped = self.last_drop_s.load(.monotonic);
        return if (stamped == 0) null else stamped;
    }

    /// Where the gate stands right now, for a reader that only wants the state.
    pub fn gateEpisode(self: *const Logger) GateEpisode {
        return Gate.of(self.gate.load(.acquire)).episode;
    }

    /// The identity a caller must carry with rows it loses.
    ///
    /// Take it where the loss actually happens. A producer discarding one entry
    /// samples at the discard; the writer samples once at `observeGate`,
    /// because the batch it is holding belongs to the episode that was open
    /// while it filled. Sampling earlier than the loss hides episodes that
    /// opened in between; sampling later cannot misattribute, because the
    /// generation makes an outrun sample fail its swap.
    fn sampleGate(self: *const Logger) Gate {
        return Gate.of(self.gate.load(.acquire));
    }

    /// Opens a gating episode under a fresh generation, or leaves one that is
    /// already holding alone — `losing` must not fall back to `gated`, and a
    /// second observation of the same gate must not look like a new episode.
    fn gateHolds(self: *Logger) void {
        var current = self.sampleGate();
        while (current.episode == .open) {
            const next: Gate = .{ .episode = .gated, .generation = current.generation +% 1 };
            const raced = self.gate.cmpxchgWeak(current.bits(), next.bits(), .acq_rel, .acquire) orelse
                return;
            current = Gate.of(raced);
        }
    }

    /// Ends the episode, keeping its generation so the next one gets a number
    /// no stale producer holds. A drop sampled during the episode that lands
    /// after this reopen finds its generation gone and is discarded.
    fn gateReopened(self: *Logger) void {
        var current = self.sampleGate();
        while (current.episode != .open) {
            const next: Gate = .{ .episode = .open, .generation = current.generation };
            const raced = self.gate.cmpxchgWeak(current.bits(), next.bits(), .acq_rel, .acquire) orelse
                return;
            current = Gate.of(raced);
        }
    }

    /// Moves the episode to wherever the monitor says the gate is, and returns
    /// the episode the caller's rows now belong to. A null monitor is no gating
    /// at all, so the episode stays `open`.
    fn observeGate(self: *Logger, monitor: ?*disk_monitor.Monitor) Gate {
        const m = monitor orelse return self.sampleGate();
        if (m.writesAllowed()) self.gateReopened() else self.gateHolds();
        return self.sampleGate();
    }
};

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 events_fixture = @import("events_fixture.zig");
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",
    });
}

/// Every provenance field set to a distinct recognisable value, so a test that
/// loses one loses it visibly.
fn fullFields(timestamp: i64) Entry.Fields {
    return .{
        .timestamp = timestamp,
        .domain = "ads.example.com",
        .client_ip = "2001:db8::1",
        .qtype = 28,
        .qclass = 1,
        .rcode = 3,
        .blocked = true,
        .response_time_us = 42,
        .cache_hit = true,
        .upstream = "https://dns.example/dns-query",
        .group_id = 7,
        .group_name = "kids",
        .policy_action = .block,
        .policy_reason = .blocklist_wildcard,
        .matched = "*.ads.example",
        .source_id = 3,
        .source_name = "steven black",
        .cname_target = "tracker.cdn.example",
        .safe_search_target = "forcesafesearch.google.com",
        .route_kind = .blocked,
        .forward_zone = "home.arpa",
    };
}

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(fullFields(1700000000));

    try testing.expectEqualStrings("ads.example.com", entry.domain());
    try testing.expectEqualStrings("2001:db8::1", entry.clientIp());
    try testing.expectEqualStrings("https://dns.example/dns-query", entry.upstream());
    try testing.expectEqual(@as(?u16, 28), entry.qtype);
    try testing.expectEqual(@as(u16, 1), entry.qclass);
    try testing.expectEqual(@as(u12, 3), entry.rcode);
    try testing.expect(entry.blocked);
    try testing.expectEqual(@as(?i64, 42), entry.response_time_us);
    try testing.expectEqual(@as(?bool, true), entry.cache_hit);

    try testing.expectEqual(@as(?i64, 7), entry.group_id);
    try testing.expectEqualStrings("kids", entry.groupName());
    try testing.expectEqual(provenance.PolicyAction.block, entry.policy_action);
    try testing.expectEqual(provenance.PolicyReason.blocklist_wildcard, entry.policy_reason);
    try testing.expectEqualStrings("*.ads.example", entry.matched());
    try testing.expectEqual(@as(?i64, 3), entry.source_id);
    try testing.expectEqualStrings("steven black", entry.sourceName());
    try testing.expectEqualStrings("tracker.cdn.example", entry.cnameTarget());
    try testing.expectEqualStrings("forcesafesearch.google.com", entry.safeSearchTarget());
    try testing.expectEqual(provenance.RouteKind.blocked, entry.route_kind);
    try testing.expectEqualStrings("home.arpa", entry.forwardZone());
}

test "an oversize string is truncated to what its buffer holds" {
    const entry: Entry = .init(.{
        .timestamp = 1,
        .domain = "a" ** 400,
        .client_ip = "c" ** 80,
        .upstream = "u" ** 600,
        .group_name = "g" ** 200,
        .matched = "m" ** 600,
        .source_name = "s" ** 200,
        .cname_target = "n" ** 400,
        .safe_search_target = "f" ** 400,
        .forward_zone = "z" ** 400,
    });

    try testing.expectEqual(@as(usize, max_domain_len), entry.domain().len);
    try testing.expectEqual(@as(usize, max_client_len), entry.clientIp().len);
    try testing.expectEqual(@as(usize, max_upstream_len), entry.upstream().len);
    try testing.expectEqual(@as(usize, limits.max_group_name_len), entry.groupName().len);
    try testing.expectEqual(@as(usize, max_matched_len), entry.matched().len);
    try testing.expectEqual(@as(usize, limits.max_source_name_len), entry.sourceName().len);
    try testing.expectEqual(@as(usize, max_name_len), entry.cnameTarget().len);
    try testing.expectEqual(@as(usize, max_name_len), entry.safeSearchTarget().len);
    try testing.expectEqual(@as(usize, max_name_len), entry.forwardZone().len);
    try testing.expectEqualStrings("a" ** max_domain_len, entry.domain());
}

test "a 256-byte matched pattern is stored whole" {
    // The widest rule the configuration accepts is a regex at
    // `regex.max_pattern_len`, and it does not fit a `u8` length — which is the
    // whole reason `matched_len` is a `u16`.
    const widest = "p" ** regex.max_pattern_len;
    const entry: Entry = .init(.{
        .timestamp = 1,
        .domain = "example.com",
        .client_ip = "192.0.2.1",
        .matched = widest,
    });
    try testing.expectEqualStrings(widest, entry.matched());
    try testing.expectEqual(@as(u16, regex.max_pattern_len), entry.matched_len);
}

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(?u16, null), bare_row.qtype);
    try testing.expectEqual(@as(?bool, null), bare_row.cache_hit);
    // Every optional text field of an entry nothing filled in reaches its
    // column as NULL rather than as an empty string.
    try testing.expectEqual(@as(?[]const u8, null), bare_row.upstream);
    try testing.expectEqual(@as(?[]const u8, null), bare_row.group_name);
    try testing.expectEqual(@as(?[]const u8, null), bare_row.matched);
    try testing.expectEqual(@as(?[]const u8, null), bare_row.source_name);
    try testing.expectEqual(@as(?[]const u8, null), bare_row.cname_target);
    try testing.expectEqual(@as(?[]const u8, null), bare_row.safe_search_target);
    try testing.expectEqual(@as(?[]const u8, null), bare_row.forward_zone);

    const full: Entry = .init(fullFields(8));
    const full_row = toRow(&full);
    try testing.expect(full_row.blocked);
    try testing.expectEqual(@as(u16, 1), full_row.qclass);
    try testing.expectEqual(@as(u12, 3), full_row.rcode);
    try testing.expectEqualStrings("https://dns.example/dns-query", full_row.upstream.?);
    try testing.expectEqual(@as(?i64, 7), full_row.group_id);
    try testing.expectEqualStrings("kids", full_row.group_name.?);
    try testing.expectEqual(provenance.PolicyAction.block, full_row.policy_action);
    try testing.expectEqual(provenance.PolicyReason.blocklist_wildcard, full_row.policy_reason);
    try testing.expectEqualStrings("*.ads.example", full_row.matched.?);
    try testing.expectEqual(@as(?i64, 3), full_row.source_id);
    try testing.expectEqualStrings("steven black", full_row.source_name.?);
    try testing.expectEqualStrings("tracker.cdn.example", full_row.cname_target.?);
    try testing.expectEqualStrings("forcesafesearch.google.com", full_row.safe_search_target.?);
    try testing.expectEqual(provenance.RouteKind.blocked, full_row.route_kind);
    try testing.expectEqualStrings("home.arpa", full_row.forward_zone.?);
}

test "an entry with every provenance field set survives the queue, toRow, insert and detailById" {
    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);

    // The widest `matched` the configuration accepts, carried the whole way:
    // 256 bytes does not fit the `u8` length every other text field uses.
    const widest_matched = "p" ** max_matched_len;
    var fields = fullFields(1234);
    fields.matched = widest_matched;
    logger.log(io, .init(fields));

    const queued = try logger.queue.getOne(io);
    try logger.flush(io, &writer, &.{queued}, null);

    var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
    defer arena_state.deinit();
    const stored = (try queries_repo.detailById(&database, arena_state.allocator(), 1)).?;

    try testing.expectEqual(@as(i64, 1234), stored.ts);
    try testing.expectEqualStrings("ads.example.com", stored.domain);
    try testing.expectEqualStrings("2001:db8::1", stored.client_ip);
    try testing.expectEqual(@as(?u16, 28), stored.qtype);
    try testing.expectEqual(@as(u16, 1), stored.qclass);
    try testing.expectEqual(@as(u12, 3), stored.rcode);
    try testing.expect(stored.blocked);
    try testing.expectEqual(@as(?i64, 42), stored.response_time_us);
    try testing.expectEqual(@as(?bool, true), stored.cache_hit);
    try testing.expectEqualStrings("https://dns.example/dns-query", stored.upstream);
    try testing.expectEqual(@as(?i64, 7), stored.group_id);
    try testing.expectEqualStrings("kids", stored.group_name);
    try testing.expectEqual(provenance.PolicyAction.block, stored.policy_action);
    try testing.expectEqual(provenance.PolicyReason.blocklist_wildcard, stored.policy_reason);
    try testing.expectEqualStrings(widest_matched, stored.matched);
    try testing.expectEqual(@as(?i64, 3), stored.source_id);
    try testing.expectEqualStrings("steven black", stored.source_name);
    try testing.expectEqualStrings("tracker.cdn.example", stored.cname_target);
    try testing.expectEqualStrings("forcesafesearch.google.com", stored.safe_search_target);
    try testing.expectEqual(provenance.RouteKind.blocked, stored.route_kind);
    try testing.expectEqualStrings("home.arpa", stored.forward_zone);
}

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 "hide_domains hides every query-derived name and leaves the labels alone" {
    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 }, &buf);
    logger.log(io, .init(fullFields(1)));
    const hidden = try logger.queue.getOne(io);

    // Every field derived from the name the client asked for.
    try testing.expectEqualStrings(hidden_marker, hidden.domain());
    try testing.expectEqualStrings(hidden_marker, hidden.matched());
    try testing.expectEqualStrings(hidden_marker, hidden.cnameTarget());
    try testing.expectEqualStrings(hidden_marker, hidden.safeSearchTarget());

    // The client is governed by `hide_client_ips`, not by this flag.
    try testing.expectEqualStrings("2001:db8::1", hidden.clientIp());

    // Configuration labels the operator wrote. They are identical on every row
    // that hits them and say nothing about which name a client looked up.
    try testing.expectEqualStrings("kids", hidden.groupName());
    try testing.expectEqualStrings("steven black", hidden.sourceName());
    try testing.expectEqualStrings("home.arpa", hidden.forwardZone());
    try testing.expectEqualStrings("https://dns.example/dns-query", hidden.upstream());
}

test "hide_client_ips hides the client and nothing else" {
    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_client_ips = true }, &buf);
    logger.log(io, .init(fullFields(1)));
    const hidden = try logger.queue.getOne(io);

    try testing.expectEqualStrings(hidden_marker, hidden.clientIp());
    try testing.expectEqualStrings("ads.example.com", hidden.domain());
    try testing.expectEqualStrings("*.ads.example", hidden.matched());
    try testing.expectEqualStrings("tracker.cdn.example", hidden.cnameTarget());
    try testing.expectEqualStrings("forcesafesearch.google.com", hidden.safeSearchTarget());
    try testing.expectEqualStrings("kids", hidden.groupName());
    try testing.expectEqualStrings("steven black", hidden.sourceName());
    try testing.expectEqualStrings("home.arpa", hidden.forwardZone());
}

test "hide_domains writes no marker into a field the query never had" {
    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 }, &buf);
    // An ordinary allowed query: no rule matched, no CNAME was uncloaked, no
    // safe-search rewrite happened. Marking those "hidden" would claim the
    // query had values it did not.
    logger.log(io, sampleEntry(1, "plain.example"));
    const hidden = try logger.queue.getOne(io);

    try testing.expectEqualStrings(hidden_marker, hidden.domain());
    try testing.expectEqualStrings("", hidden.matched());
    try testing.expectEqualStrings("", hidden.cnameTarget());
    try testing.expectEqualStrings("", hidden.safeSearchTarget());
}

test "the entry queue's worst case stays inside its byte budget" {
    // The bound `config/validate.zig` enforces is derived from this, so the
    // budget is what a maximal configuration can actually cost.
    try testing.expect(@as(usize, query_log_buffer_max) * @sizeOf(Entry) <= queue_budget_bytes);
    // One more entry than the ceiling would exceed it, so the ceiling is the
    // largest value that fits rather than a round number under it.
    try testing.expect((@as(usize, query_log_buffer_max) + 1) * @sizeOf(Entry) > queue_budget_bytes);

    // The shipped default has to be comfortably inside the budget, or the
    // out-of-the-box configuration is the one that spends it. At the widths
    // above it costs about 17 MiB, roughly a quarter of the ceiling.
    const default_max: usize = (model.Logging{}).query_log_buffer_max;
    try testing.expect(default_max <= query_log_buffer_max);
    try testing.expect(default_max * @sizeOf(Entry) <= queue_budget_bytes / 2);
}

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());
}

/// The rendezvous that forces the flip to land BETWEEN two producer entries
/// rather than whenever the scheduler feels like it.
const PrivacyFlip = struct {
    logger: *Logger,
    /// Set by the producer once its pre-flip entry is enqueued.
    before_done: std.Io.Event = .unset,
    /// Set by the flipper once `setPrivacy` has returned.
    flipped: std.Io.Event = .unset,

    fn run(self: *PrivacyFlip, io: std.Io) void {
        self.before_done.wait(io) catch return;
        self.logger.setPrivacy(.{ .hide_domains = true, .hide_client_ips = true });
        self.flipped.set(io);
    }
};

test "a privacy flip redacts every entry after it and none before it" {
    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);
    var flip: PrivacyFlip = .{ .logger = &logger };

    var future = try io.concurrent(PrivacyFlip.run, .{ &flip, io });
    logger.log(io, sampleEntry(1, "before.example"));
    flip.before_done.set(io);
    try flip.flipped.wait(io);
    logger.log(io, sampleEntry(2, "after.example"));
    future.await(io);

    const before = try logger.queue.getOne(io);
    try testing.expectEqualStrings("before.example", before.domain());
    try testing.expectEqualStrings("192.0.2.10", before.clientIp());

    const after = try logger.queue.getOne(io);
    try testing.expectEqualStrings(hidden_marker, after.domain());
    try testing.expectEqualStrings(hidden_marker, after.clientIp());
}

fn flipPrivacyRepeatedly(logger: *Logger, rounds: usize) void {
    for (0..rounds) |i| {
        logger.setPrivacy(if (i % 2 == 0)
            .{}
        else
            .{ .hide_domains = true, .hide_client_ips = true });
    }
}

test "no producer observes a privacy policy that redacts one field and not the other" {
    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);
    const source = sampleEntry(1, "tracker.example");

    const rounds = 20_000;
    var flipper = try io.concurrent(flipPrivacyRepeatedly, .{ &logger, rounds });
    // Recorded, not asserted, while the flipper runs: an assertion that
    // returned here would leave `Threaded.deinit` joining a task nothing ends.
    var mixed = false;
    for (0..rounds) |_| {
        const out = logger.transformed(source);
        const domain_hidden = std.mem.eql(u8, out.domain(), hidden_marker);
        const client_hidden = std.mem.eql(u8, out.clientIp(), hidden_marker);
        if (domain_hidden != client_hidden) mixed = true;
    }
    flipper.await(io);
    try testing.expect(!mixed);
}

test "the split halves reproduce log byte for byte" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    const configs = [_]model.Logging{
        .{},
        .{ .hide_domains = true },
        .{ .hide_client_ips = true },
        .{ .hide_domains = true, .hide_client_ips = true },
    };

    for (configs) |cfg| {
        var buf: [4]Entry = undefined;
        var logger: Logger = .init(cfg, &buf);
        const source = sampleEntry(100, "tracker.example");

        logger.log(io, source);
        logger.logTransformed(io, logger.transformed(source));

        const from_log = try logger.queue.getOne(io);
        const from_halves = try logger.queue.getOne(io);
        try testing.expectEqualSlices(
            u8,
            std.mem.asBytes(&from_log),
            std.mem.asBytes(&from_halves),
        );
    }
}

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 "shutdown writes the batch the writer holds and the rest of the queue" {
    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;
    // An hour: nothing here can be explained by the window expiring. Every row
    // that lands does so because the close released it.
    var logger: Logger = .init(.{ .query_log_flush_interval_s = 3600 }, &buf);

    var future = try io.concurrent(Logger.runWriter, .{
        &logger,
        io,
        &database,
        @as(?*disk_monitor.Monitor, null),
    });
    // Declared after the `database.close` defer so LIFO stops the writer first:
    // an early return anywhere below would otherwise close the handle under a
    // live writer and leave `Threaded.deinit` joining a task nothing ends.
    defer {
        logger.shutdown(io);
        future.await(io) catch {};
    }

    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));
    }

    // The app.zig order: the only producer is this task and it is done, so the
    // close cannot lose an entry, and the writer is awaited rather than
    // canceled.
    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 "entries that arrive inside one window reach the database in one batch" {
    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: [16]Entry = undefined;
    var logger: Logger = .init(.{ .query_log_flush_interval_s = 3600 }, &buf);

    for (0..6) |i| logger.log(io, sampleEntry(@intCast(i), "batched.example"));
    // No further producer exists, so the close is what ends the fill — the same
    // thing that ends it at shutdown.
    logger.shutdown(io);

    var batch: [flush_batch]Entry = undefined;
    batch[0] = try logger.queue.getOne(io);
    var n: usize = 1;
    try logger.fill(io, &batch, logger.flushDeadline(io), &n, logger.sampleGate());
    try testing.expectEqual(@as(usize, 6), n);

    // One `writeBatch` call, which is one transaction (`queries_repo.zig`).
    try logger.flush(io, &writer, batch[0..n], null);
    try testing.expectEqual(@as(u64, 6), logger.rows_written.load(.monotonic));
    try testing.expectEqual(@as(i64, 6), try queries_repo.countRows(&database));
}

test "a zero interval takes what is queued and waits for nothing" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var buf: [8]Entry = undefined;
    var logger: Logger = .init(.{ .query_log_flush_interval_s = 0 }, &buf);

    logger.log(io, sampleEntry(1, "now.example"));
    logger.log(io, sampleEntry(2, "now.example"));

    var batch: [flush_batch]Entry = undefined;
    batch[0] = try logger.queue.getOne(io);
    var n: usize = 1;
    // The queue is open and the batch has room: any non-zero interval blocks
    // here until it expires. Zero returns with what was already queued.
    try logger.fill(io, &batch, logger.flushDeadline(io), &n, logger.sampleGate());
    try testing.expectEqual(@as(usize, 2), n);
}

test "the writer holds an entry for the length of the flush interval" {
    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(.{ .query_log_flush_interval_s = 1 }, &buf);

    var future = try io.concurrent(Logger.runWriter, .{
        &logger,
        io,
        &database,
        @as(?*disk_monitor.Monitor, null),
    });
    // See the note in "shutdown writes the batch the writer holds": this must
    // run before the deferred `database.close`.
    defer {
        logger.shutdown(io);
        future.await(io) catch {};
    }

    logger.log(io, sampleEntry(1, "only.example"));

    // Sampled, not asserted, while the writer runs: an assertion that fails
    // here would return before the writer is stopped, and `Threaded.deinit`
    // then waits on a task nothing will ever end.
    const quarter: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(250), .clock = .awake };
    try quarter.sleep(io);
    const written_at_a_quarter = logger.rows_written.load(.monotonic);

    // Three times the interval. A flush that has not happened by then is a
    // failure, not slowness.
    const poll: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(5), .clock = .awake };
    var waited: usize = 0;
    while (logger.rows_written.load(.monotonic) == 0 and waited < 600) : (waited += 1) {
        try poll.sleep(io);
    }

    logger.shutdown(io);
    try future.await(io);

    // A writer that commits per query has already written at a quarter of the
    // window; this one has not.
    try testing.expectEqual(@as(u64, 0), written_at_a_quarter);
    try testing.expect(waited < 600);
    try testing.expectEqual(@as(u64, 1), logger.rows_written.load(.monotonic));
    try testing.expectEqual(@as(i64, 1), try queries_repo.countRows(&database));
}

test "a full batch flushes without waiting for the interval" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var database = try openLog();
    defer database.close();

    var buf: [256]Entry = undefined;
    var logger: Logger = .init(.{ .query_log_flush_interval_s = 3600 }, &buf);

    var future = try io.concurrent(Logger.runWriter, .{
        &logger,
        io,
        &database,
        @as(?*disk_monitor.Monitor, null),
    });
    // See the note in "shutdown writes the batch the writer holds": this must
    // run before the deferred `database.close`.
    defer {
        logger.shutdown(io);
        future.await(io) catch {};
    }

    for (0..150) |i| logger.log(io, sampleEntry(@intCast(i), "burst.example"));

    // The window is an hour away, so only a full batch can release a flush.
    const poll: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(5), .clock = .awake };
    var waited: usize = 0;
    while (logger.rows_written.load(.monotonic) < flush_batch and waited < 400) : (waited += 1) {
        try poll.sleep(io);
    }
    // Sampled with the writer still running, asserted once it has stopped.
    const written_before_shutdown = logger.rows_written.load(.monotonic);

    logger.shutdown(io);
    try future.await(io);

    // Exactly one batch went out early, and the 50 behind it waited for the
    // close rather than for the hour.
    try testing.expect(waited < 400);
    try testing.expectEqual(@as(u64, flush_batch), written_before_shutdown);
    try testing.expectEqual(@as(i64, 150), try queries_repo.countRows(&database));
}

test "the writer's next cycle uses the interval set since its last one" {
    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;
    // Zero: every cycle commits what it has and goes straight back to `getOne`,
    // so the first entry proves the writer is running and parked between cycles.
    var logger: Logger = .init(.{ .query_log_flush_interval_s = 0 }, &buf);

    var future = try io.concurrent(Logger.runWriter, .{
        &logger,
        io,
        &database,
        @as(?*disk_monitor.Monitor, null),
    });
    // See the note in "shutdown writes the batch the writer holds": this must
    // run before the deferred `database.close`.
    defer {
        logger.shutdown(io);
        future.await(io) catch {};
    }

    logger.log(io, sampleEntry(1, "first.example"));
    const poll: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(5), .clock = .awake };
    var waited: usize = 0;
    while (logger.rows_written.load(.monotonic) == 0 and waited < 400) : (waited += 1) {
        try poll.sleep(io);
    }

    // An hour, installed while the writer is parked: the cycle the next entry
    // starts must wait it out instead of committing at once.
    logger.setFlushInterval(3600);
    logger.log(io, sampleEntry(2, "second.example"));
    const quarter: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(250), .clock = .awake };
    try quarter.sleep(io);
    const written_under_the_new_interval = logger.rows_written.load(.monotonic);

    logger.shutdown(io);
    try future.await(io);

    try testing.expect(waited < 400);
    // The first entry, and only the first: the second is still held.
    try testing.expectEqual(@as(u64, 1), written_under_the_new_interval);
    // The close releases it, which is what makes the hold a hold and not a loss.
    try testing.expectEqual(@as(i64, 2), 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),
    });
    // This task is `flush`, not `runWriter`: no queue shutdown can release it,
    // so only cancellation ends it on an early return. Declared after the
    // `writer.deinit`/`database.close` defers so LIFO runs it first.
    defer {
        _ = future.cancel(io) catch {};
    }

    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 queue overflow stamps the drop and leaves the gate episode open" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    // No writer and no monitor: nothing consumes the queue, so the third entry
    // has to displace the oldest, and no gate is involved in the loss.
    var buf: [2]Entry = undefined;
    var logger: Logger = .init(.{}, &buf);

    try testing.expectEqual(@as(?i64, null), logger.lastDropSeconds());

    for (0..3) |i| logger.log(io, sampleEntry(@intCast(i), "overflow.example"));

    try testing.expectEqual(@as(u64, 1), logger.queries_dropped.load(.monotonic));
    const stamped = logger.lastDropSeconds() orelse return error.TestUnexpectedResult;
    try testing.expect(stamped > 1_700_000_000);
    // Cumulative loss is not a current fault: the episode never opened.
    try testing.expectEqual(GateEpisode.open, logger.gateEpisode());
}

test "the gating episode opens on the gate, turns losing on a drop, and clears on recovery" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var database = try openLog();
    defer database.close();

    // Small enough that the entries logged below have to displace each other,
    // which is the gate-caused overflow this episode is meant to catch.
    var buf: [4]Entry = undefined;
    var logger: Logger = .init(.{ .query_log_flush_interval_s = 0 }, &buf);

    var monitor: disk_monitor.Monitor = .init(.{}, std.Io.Dir.cwd(), ".", null);
    monitor.state_raw.store(@intFromEnum(disk_monitor.State.critical), .monotonic);

    logger.log(io, sampleEntry(1, "held.example"));

    var future = try io.concurrent(Logger.runWriter, .{
        &logger,
        io,
        &database,
        @as(?*disk_monitor.Monitor, &monitor),
    });
    // See the note in "shutdown writes the batch the writer holds": this must
    // run before the deferred `database.close`.
    defer {
        logger.shutdown(io);
        future.await(io) catch {};
    }

    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);
    }

    // The gate is holding and nothing has been lost yet, which is not a fault:
    // the batch is still going to be written if the disk recovers.
    try testing.expectEqual(GateEpisode.gated, logger.gateEpisode());
    try testing.expectEqual(@as(u64, 0), logger.queries_dropped.load(.monotonic));

    // Now overflow the queue behind the held batch. These drops belong to the
    // episode, and that is what turns it from held to losing.
    for (0..32) |i| logger.log(io, sampleEntry(@intCast(i + 2), "queued.example"));
    try testing.expect(logger.queries_dropped.load(.monotonic) > 0);
    try testing.expectEqual(GateEpisode.losing, logger.gateEpisode());
    try testing.expect(logger.lastDropSeconds() != null);

    // The disk recovers: the held batch goes out and the episode ends.
    monitor.state_raw.store(@intFromEnum(disk_monitor.State.ok), .monotonic);
    waited = 0;
    // The write is the completion condition, not the episode: `flush` reopens
    // the gate before it calls `writeBatch`, so a poll on the episode alone
    // returns while the row is still in flight and `rows_written` is still 0.
    while (logger.rows_written.load(.monotonic) == 0) : (waited += 1) {
        try testing.expect(waited < 400);
        try poll.sleep(io);
    }
    try testing.expectEqual(GateEpisode.open, logger.gateEpisode());
    // The count keeps the history the state does not.
    try testing.expect(logger.queries_dropped.load(.monotonic) > 0);

    logger.shutdown(io);
    try future.await(io);
}

// The two tests below drive the gate primitives directly. A threaded test
// cannot prove the absence of the race they close — it can only fail to hit it
// — so they pin the mechanism instead: the generation a drop must match, and
// the direction the stamp may move.

test "a drop sampled in one episode cannot mark the next one losing" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var queue_buf: [2]Entry = undefined;
    var logger: Logger = .init(.{}, &queue_buf);

    logger.gateHolds();
    // What a producer holds while it stalls: episode one, still holding.
    const stalled_in = logger.sampleGate();
    try testing.expectEqual(GateEpisode.gated, stalled_in.episode);

    // The gate opens and closes again while that producer is descheduled.
    logger.gateReopened();
    logger.gateHolds();
    const episode_two = logger.sampleGate();
    try testing.expectEqual(GateEpisode.gated, episode_two.episode);
    try testing.expect(episode_two.generation != stalled_in.generation);

    // The stale drop still counts as a lost row — it was one — but episode two
    // has cost nothing and must not be told it has.
    logger.countDropped(io, 1, stalled_in);
    try testing.expectEqual(@as(u64, 1), logger.queries_dropped.load(.monotonic));
    try testing.expectEqual(GateEpisode.gated, logger.gateEpisode());

    // A drop that really belongs to episode two does raise it.
    logger.countDropped(io, 1, episode_two);
    try testing.expectEqual(GateEpisode.losing, logger.gateEpisode());

    // And a second hold of a gate that never opened is the same episode, not a
    // new one: `losing` must not fall back to `gated`.
    logger.gateHolds();
    try testing.expectEqual(GateEpisode.losing, logger.gateEpisode());
    try testing.expectEqual(episode_two.generation, logger.sampleGate().generation);
}

fn logOne(logger: *Logger, io: std.Io) void {
    logger.log(io, sampleEntry(2, "second.example"));
}

test "a gate that closes mid-enqueue still gets the discard that follows it" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    // Capacity one and no writer, so the second entry has to evict the first
    // and the producer discards inside `enqueue`.
    var queue_buf: [1]Entry = undefined;
    var logger: Logger = .init(.{}, &queue_buf);

    logger.log(io, sampleEntry(1, "first.example"));
    try testing.expectEqual(GateEpisode.open, logger.gateEpisode());

    discard_stall.parked = .unset;
    discard_stall.release = .unset;
    discard_stall.armed = true;
    defer discard_stall.armed = false;

    // The producer enters `enqueue` with the gate open — which is the reading a
    // sample taken before the loop would keep for the rest of the call.
    var producer = try io.concurrent(logOne, .{ &logger, io });
    // The producer parks inside `enqueue` and only `release` frees it: an early
    // return below would otherwise leave `Threaded.deinit` joining it forever.
    defer {
        discard_stall.armed = false;
        discard_stall.release.set(io);
        producer.await(io);
    }
    discard_stall.parked.waitUncancelable(io);

    // The producer is parked with the row it will evict still on the queue, so
    // the episode opens strictly before the loss rather than after it. Nothing
    // has been dropped yet, and that is what makes the row this episode's.
    try testing.expectEqual(@as(u64, 0), logger.queries_dropped.load(.monotonic));
    logger.gateHolds();
    discard_stall.armed = false;
    discard_stall.release.set(io);
    producer.await(io);

    try testing.expectEqual(@as(u64, 1), logger.queries_dropped.load(.monotonic));
    // The assertion the pre-loop sample fails: it would still be holding
    // `open`, `countDropped` would return before its swap, and the episode
    // would sit at `gated` having silently cost a row.
    try testing.expectEqual(GateEpisode.losing, logger.gateEpisode());
}

test "the drop stamp only ever moves forward" {
    var queue_buf: [2]Entry = undefined;
    var logger: Logger = .init(.{}, &queue_buf);

    try testing.expectEqual(@as(?i64, null), logger.lastDropSeconds());

    logger.stampDrop(1_700_000_100);
    try testing.expectEqual(@as(?i64, 1_700_000_100), logger.lastDropSeconds());

    // A producer that read its clock earlier but arrives later: the newer drop
    // already published its time and must keep it.
    logger.stampDrop(1_700_000_050);
    try testing.expectEqual(@as(?i64, 1_700_000_100), logger.lastDropSeconds());

    logger.stampDrop(1_700_000_200);
    try testing.expectEqual(@as(?i64, 1_700_000_200), logger.lastDropSeconds());
}

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;
    // Zero interval: the writer reaches the gate with what is queued instead of
    // waiting a minute for a producer that no longer exists.
    var logger: Logger = .init(.{ .query_log_flush_interval_s = 0 }, &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),
    });
    // Cancellation is this case's subject, so the guard is the same operation
    // the body performs; declared after the `database.close` defer so LIFO ends
    // the writer before the handle goes away.
    defer {
        _ = future.cancel(io) catch {};
    }

    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 "a disk-gated writer drops what it holds at shutdown instead of hanging" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var database = try openLog();
    defer database.close();

    var fx: events_fixture.Fixture = .{};
    try fx.init(io, 1000);
    defer fx.deinit();

    var buf: [512]Entry = undefined;
    var logger: Logger = .init(.{ .query_log_flush_interval_s = 0 }, &buf);
    logger.diagnostics = &fx.store;

    var monitor: disk_monitor.Monitor = .init(.{}, std.Io.Dir.cwd(), ".", null);
    monitor.state_raw.store(@intFromEnum(disk_monitor.State.critical), .monotonic);

    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),
    });
    // See the note in "shutdown writes the batch the writer holds": this must
    // run before the deferred `fx.deinit` and `database.close`.
    defer {
        logger.shutdown(io);
        future.await(io) catch {};
    }

    const poll: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(5), .clock = .awake };
    var waited: usize = 0;
    while (logger.batches_gated.load(.monotonic) == 0 and waited < 400) : (waited += 1) {
        try poll.sleep(io);
    }

    // Two and a half chunks wait behind the batch the gate is holding, so a
    // drain that reported per chunk would file three episodes' worth of writes
    // to the disk that is out of space.
    const queued = 250;
    for (0..queued) |i| logger.log(io, sampleEntry(@intCast(i + 3), "queued.example"));

    // The disk never recovers. The wait still ends, and every entry is counted.
    logger.shutdown(io);
    try future.await(io);

    try testing.expect(waited < 400);
    try testing.expectEqual(@as(u64, queued + 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));

    // One episode, reported once, and its detail carries the whole loss rather
    // than the size of whichever chunk was last.
    try testing.expectEqual(@as(i64, 1), try fx.count("SELECT count(*) FROM operational_events"));
    try testing.expectEqual(@as(i64, 1), try fx.count("SELECT occurrences FROM operational_events"));
    try testing.expectEqualStrings("batch", try fx.text("SELECT subject_key FROM operational_events"));
    try testing.expectEqualStrings(
        "query log rows were dropped at shutdown: DiskCritical (252 rows)",
        try fx.text("SELECT detail FROM operational_events"),
    );
}

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));
}

test "a dropped batch opens an error episode the next good batch closes" {
    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 fx: events_fixture.Fixture = .{};
    try fx.init(io, 1000);
    defer fx.deinit();

    var writer = try queries_repo.BatchWriter.init(&database);
    defer writer.deinit();

    var buf: [4]Entry = undefined;
    var logger: Logger = .init(.{}, &buf);
    logger.diagnostics = &fx.store;

    var doomed = sampleEntry(10, "poison.example");
    doomed.setClientIp("boom");
    const bad = [_]Entry{doomed};
    try logger.flush(io, &writer, &bad, null);

    try testing.expectEqualStrings("query_log.write", try fx.text("SELECT code FROM operational_events"));
    try testing.expectEqualStrings("batch", try fx.text("SELECT subject_key FROM operational_events"));
    try testing.expectEqualStrings("error", try fx.text("SELECT severity FROM operational_events"));

    const good = [_]Entry{sampleEntry(11, "next.example")};
    try logger.flush(io, &writer, &good, null);

    try testing.expectEqual(@as(i64, 1), try fx.count("SELECT count(*) FROM operational_events"));
    try testing.expectEqual(
        @as(i64, 0),
        try fx.count("SELECT count(*) FROM operational_events WHERE resolved_at IS NULL"),
    );
}

test "a writer that cannot prepare leaves an episode no recovery path claims" {
    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 fx: events_fixture.Fixture = .{};
    try fx.init(io, 1000);
    defer fx.deinit();

    var buf: [8]Entry = undefined;
    var logger: Logger = .init(.{}, &buf);
    logger.diagnostics = &fx.store;

    try logger.runWriter(io, &database, null);

    try testing.expectEqualStrings("writer", try fx.text(
        "SELECT subject_key FROM operational_events WHERE resolved_at IS NULL",
    ));
    try testing.expectEqualStrings("error", try fx.text(
        "SELECT severity FROM operational_events WHERE resolved_at IS NULL",
    ));

    // The writer returned, so nothing can ever close this. A second run finds
    // the queue closed and adds no second episode.
    try logger.runWriter(io, &database, null);
    try testing.expectEqual(
        @as(i64, 1),
        try fx.count("SELECT count(*) FROM operational_events WHERE resolved_at IS NULL"),
    );
}