//! The blocklist manager (PLAN §4): the compiled files under
//! `<data_dir>/blocklists/`, the refresh that produces them, the metadata
//! columns it writes back, the snapshot built from them and the swap that
//! publishes it.
//!
//! This is the only file in this milestone that touches both the database and
//! the filesystem. Everything it composes — the parsers, the compiler, the
//! domain sets, the rule sets and the snapshot — is pure and testable without
//! either.
//!
//! **The swap is an `std.Io.RwLock`, not a lock-free pointer.** PLAN §7.3 says
//! "readers lock-free"; this is a deliberate deviation. Freeing the old
//! snapshot without a lock needs epoch-based reclamation or hazard pointers: a
//! class of code that is very hard to get right and impossible to test
//! convincingly, bought for a household resolver whose target is 100 qps. A
//! shared lock held for the microseconds of one `evaluate` costs an uncontended
//! atomic pair; the writer takes the exclusive lock only on a swap, which
//! happens on refresh. The old snapshot is freed *after* `unlock` returns, and
//! the `Handle` API makes "do not retain the pointer" the only shape a caller
//! can write.
//!
//! The same lock guards the status table, which is written from the refresh
//! task and read by the API. Both critical sections are short and hold no
//! socket and no file, so the uncancelable lock forms are used: a lock this
//! code takes is always released within a few instructions.
//!
//! Two more locks serialize the writers, and they divide the work by how long
//! it takes.
//!
//! `writer_lock` covers what a writer does to the *published* state: build a
//! snapshot, install the compiled files, write the runtime columns, record a
//! status. Two concurrent reloads would otherwise compute the same generation
//! and each destroy a snapshot the other had just published. Every section it
//! guards is bounded by local work — a read of the compiled files at worst —
//! so a web mutation that ends in `reload` never waits out a download.
//!
//! `refresh_lock` covers what a writer does *before* it has anything to
//! publish: the download of one source, at up to 300 s each, and the compile
//! that follows it. It also covers blocklist-directory maintenance, because
//! those stages are the only writers of `.raw.tmp` / `.list.tmp` /
//! `.wild.tmp` / `.allow.tmp` and `pruneOrphans` must not sweep the
//! temporaries of a refresh that is still running. Two concurrent refreshes
//! would share the fetcher's buffers and, for one source, the same temporary
//! paths.
//!
//! **Lock ordering: `refresh_lock` is never acquired while `writer_lock` is
//! held.** A path that needs both takes `refresh_lock` first. The public entry
//! points take what they need; the `*Locked` bodies assume it and never take
//! it again, because neither mutex is reentrant.

const std = @import("std");
const Allocator = std.mem.Allocator;

const model = @import("../config/model.zig");
const safe_url = @import("../safe_url.zig");
const db = @import("../storage/db.zig");
const clients_repo = @import("../storage/repositories/clients_repo.zig");
const groups_repo = @import("../storage/repositories/groups_repo.zig");
const rules_repo = @import("../storage/repositories/rules_repo.zig");
const sources_repo = @import("../storage/repositories/sources_repo.zig");
const disk_monitor = @import("../storage/disk_monitor.zig");
const events = @import("../storage/events.zig");
const compiler = @import("compiler.zig");
const fetcher = @import("fetcher.zig");
const matcher = @import("matcher.zig");
const parsers = @import("parsers.zig");

const log = std.log.scoped(.blocklist_manager);
const Sha256 = std.crypto.hash.sha2.Sha256;

/// `SourceStatus.last_error` is fixed-size so the failure path allocates
/// nothing.
pub const max_error_len: usize = 128;

/// `SourceStatus.url` is fixed-size so a copied status borrows nothing. A
/// blocklist url longer than this is truncated in the status only; the row
/// keeps it whole.
///
/// The log form of a url is bounded separately by `safe_url.max_len`. The two
/// numbers agree today and answer different questions; neither follows the
/// other.
pub const max_url_len: usize = 255;

/// A compiled body larger than this is refused at load. A source that reaches
/// it produced more than `fetcher.max_body_bytes` of names, which cannot
/// happen from a download this fetcher performed.
pub const max_compiled_bytes: usize = 128 * 1024 * 1024;

/// Buffer size for every file stream this file opens. One buffer is live per
/// stage, and the stages do not overlap.
const io_buf_len: usize = 64 * 1024;

/// Holds the sniff sample: `parsers.sample_lines` lines of at most
/// `compiler.max_line_len` bytes, each with its newline. A fixed byte window
/// would be spent by a handful of legal 4096-byte comment lines and the format
/// would then be decided by almost no data.
const sample_buf_len: usize = parsers.sample_lines * (compiler.max_line_len + 1);

/// `<id>` is at most 20 characters and the longest suffix is `.allow.tmp`.
const name_buf_len: usize = 48;

/// How one blocklist source is named in a log line: by its row id and its name,
/// which are its own identity, and by its redacted url, which says where it
/// points and nothing more.
///
/// The url used to carry the identity here on its own. It cannot: `safe_url`
/// drops the path, because a path segment is a place an operator's token lives,
/// and two sources on one host are told apart by exactly that path. The id and
/// the name are on the row every one of these lines already holds, they are
/// what the API and the web UI show, and neither can leak what the url holds.
/// The name is escaped for the same reason the url is — both are database text
/// and a newline in either would forge a log line. It carries its own quotes,
/// out of `safe_url.quoteText`, because a quote this format string added would
/// be a quote the name could close: `ads' (https://decoy.example) --` would then
/// read as a source pointing somewhere it does not.
const SourceLabel = struct {
    id: i64,
    name: []const u8,
    url: []const u8,

    fn of(row: sources_repo.SourceRow) SourceLabel {
        return .{ .id = row.id, .name = row.name, .url = row.url };
    }

    pub fn format(self: SourceLabel, w: *std.Io.Writer) std.Io.Writer.Error!void {
        try w.print("source {d} {f} {f}", .{
            self.id,
            safe_url.quoteText(self.name),
            safe_url.redactQuoted(self.url),
        });
    }
};

pub const Paths = struct {
    /// `<data_dir>`, owned by the caller and left open for the manager's life.
    dir: std.Io.Dir,
    subdir: []const u8 = "blocklists",
};

/// Where one source stands. `.never_fetched` is the state of a source that has
/// no compiled files and no stored checksum, which is a fresh install rather
/// than a failure. `.no_valid_entries` is a download that compiled cleanly and
/// yielded nothing usable — an error page or a compressed body, not a
/// blocklist.
pub const State = enum {
    ok,
    never_fetched,
    fetch_failed,
    compile_failed,
    no_valid_entries,
    load_failed,

    /// Whether this state was recorded by a refresh rather than by a load.
    /// A load outcome never overwrites one: the files a reload just read are
    /// exactly the files the failed refresh could not replace, and the operator
    /// still has to see why the update did not land. `SourceStatus.loaded`
    /// carries the other half — whether the source is filtering at all.
    pub fn isRefreshFailure(self: State) bool {
        return switch (self) {
            .fetch_failed, .compile_failed, .no_valid_entries => true,
            .ok, .never_fetched, .load_failed => false,
        };
    }
};

/// The `blocklist.storage` operations, each its own episode subject.
///
/// A fixed set on purpose: several of these fail once per file in a pass, and
/// one slot per operation is what turns that into one report per pass instead
/// of an unbounded list of them.
pub const StorageOp = enum { sweep, directory_read, create_dir, open_dir, delete };

/// One operation's outcome across one pass. `detail` keeps the last failure,
/// and `failures` says how many that pass held — the row's `occurrences` counts
/// failing passes, so the count belongs in the text.
const Aggregate = struct {
    failures: u32 = 0,
    succeeded: bool = false,
    detail: [events.Store.max_detail_len]u8 = @splat(0),
    detail_len: u16 = 0,

    fn detailText(self: *const Aggregate) []const u8 {
        return self.detail[0..self.detail_len];
    }
};

/// What a locked body observed, held until `Manager.flushDiagnostics` can
/// report it with no manager lock held.
///
/// Two of this file's operations cannot report from where they stand:
/// `publishRefresh` runs under `writer_lock` by contract and `pruneOrphans`
/// holds both writer mutexes through its filesystem work. Collect-then-flush is
/// what keeps their outcomes without holding a lock across a store call, and
/// nothing here can grow: the storage slots are an enum array and a refresh
/// outcome rides the status entry the source already has.
///
/// The flush still happens inside the lock that serializes passes — a pass
/// drains its own outcomes before it releases `refresh_lock` (or, for a
/// standalone `reload`, `writer_lock`). What collect-then-flush avoids is
/// holding a *manager* lock across a store call, not deferring the report until
/// the next pass could merge into it.
const Pending = struct {
    mutex: std.Io.Mutex = .init,
    storage: std.EnumArray(StorageOp, Aggregate) = .initFill(.{}),
    /// Null until a pass observes a snapshot outcome at all.
    snapshot_failed: ?bool = null,
    snapshot_detail: [events.Store.max_detail_len]u8 = @splat(0),
    snapshot_detail_len: u16 = 0,
};

/// A status is a value with no borrowed memory, so a copy handed to the API
/// outlives every reload. The url is held inline for that reason.
pub const SourceStatus = struct {
    id: i64,
    state: State = .never_fetched,
    /// Whether this source's compiled files were read into the most recent
    /// snapshot build — that is, whether it is filtering right now. `state`
    /// describes the most recent attempt to *produce* those files, which is a
    /// different fact: a source whose refresh failed keeps serving what the
    /// refresh did not replace, and reads `.fetch_failed` with `loaded` set.
    loaded: bool = false,
    last_attempt: i64 = 0,
    last_success: i64 = 0,
    counts: compiler.Counts = .{},
    /// A display copy of the source url, truncated at `max_url_len`. The whole
    /// url is in the `blocklist_sources` row this status shares an `id` with.
    url: [max_url_len]u8 = @splat(0),
    url_len: u8 = 0,
    last_error: [max_error_len]u8 = @splat(0),
    last_error_len: u8 = 0,
    /// The diagnostics identity of this source, canonicalized from the WHOLE
    /// url by `setUrl`. `url` above is a display copy truncated at
    /// `max_url_len`, and two urls sharing a 255-byte prefix would share one
    /// episode if that copy were the key.
    event_key: [events.Store.max_subject_key_len]u8 = @splat(0),
    event_key_len: u16 = 0,
    /// Diagnostics accounting for the pass in progress, cleared by every
    /// `flushDiagnostics`. `pass_outcome` says this source recorded one at all;
    /// `pass_failures` counts the failing ones, which a pass can hold more than
    /// one of (a refresh that failed, then the reload that could not load the
    /// files it did not write). One flush reports one `blocklist.refresh`
    /// occurrence per source, so `occurrences` counts failing passes rather
    /// than flushes, and the detail carries how many failures the pass held.
    ///
    /// These two fields live in exactly one copy of the status table at a time,
    /// which is what makes that count right while reloads replace the table
    /// underneath: a candidate built by `mergeStatuses` carries none of them,
    /// `installStatuses` folds the live table's in as it swaps, and the flush
    /// claims an entry by copying it and zeroing both fields in one locked
    /// step. Copy them anywhere else and the outcome gets reported twice.
    pass_outcome: bool = false,
    pass_failures: u16 = 0,

    pub fn eventKey(self: *const SourceStatus) []const u8 {
        return self.event_key[0..self.event_key_len];
    }

    pub fn errorText(self: *const SourceStatus) []const u8 {
        return self.last_error[0..self.last_error_len];
    }

    pub fn urlText(self: *const SourceStatus) []const u8 {
        return self.url[0..self.url_len];
    }

    fn setUrl(self: *SourceStatus, url: []const u8) void {
        const kept = @min(url.len, max_url_len);
        @memcpy(self.url[0..kept], url[0..kept]);
        @memset(self.url[kept..], 0);
        self.url_len = @intCast(kept);
        self.event_key_len = @intCast(events.canonicalKey(url, &self.event_key).len);
    }

    fn fail(self: *SourceStatus, state: State, text: []const u8) void {
        self.pass_failures +|= 1;
        self.pass_outcome = true;
        self.state = state;
        const kept = @min(text.len, max_error_len);
        @memcpy(self.last_error[0..kept], text[0..kept]);
        @memset(self.last_error[kept..], 0);
        self.last_error_len = @intCast(kept);
    }

    fn succeed(self: *SourceStatus, at: i64, counts: compiler.Counts) void {
        self.pass_outcome = true;
        self.state = .ok;
        self.counts = counts;
        self.last_success = at;
        self.last_error = @splat(0);
        self.last_error_len = 0;
    }
};

/// The header every compiled file carries, ahead of the body. The `sha256`
/// covers the `.list` body, then the `.wild` body, then the `.allow` body, and
/// **not** the header, so it stays stable across a refetch of unchanged content
/// while `fetched_at` moves.
///
/// Each body is followed by a separator byte, so the digest identifies which
/// body a name sits in rather than only which names were written.
pub const Header = struct {
    url: []const u8,
    format: parsers.Format,
    fetched_at: i64,
    counts: compiler.Counts,
    /// 64 lowercase hex characters.
    checksum: []const u8,

    pub fn write(self: Header, w: *std.Io.Writer) std.Io.Writer.Error!void {
        try w.writeAll("# nxdns blocklist\n");
        try w.print("# url {s}\n", .{self.url});
        try w.print("# format {s}\n", .{@tagName(self.format)});
        try w.print("# fetched_at {d}\n", .{self.fetched_at});
        try w.print("# domains {d}\n", .{self.counts.domains});
        try w.print("# wildcards {d}\n", .{self.counts.wildcards});
        try w.print("# exceptions {d}\n", .{self.counts.exceptions});
        try w.print("# skipped_regex {d}\n", .{self.counts.skipped_regex});
        try w.print("# skipped_unsupported {d}\n", .{self.counts.skipped_unsupported});
        try w.print("# invalid {d}\n", .{self.counts.invalid});
        try w.print("# sha256 {s}\n", .{self.checksum});
    }
};

/// The body of a compiled file: everything after the leading `#` lines. A file
/// with no header is all body, which is what makes a hand-written fixture a
/// legal compiled file.
pub fn stripHeader(bytes: []const u8) []const u8 {
    var rest = bytes;
    while (rest.len != 0 and rest[0] == '#') {
        const newline = std.mem.indexOfScalar(u8, rest, '\n') orelse return rest[rest.len..];
        rest = rest[newline + 1 ..];
    }
    return rest;
}

/// The scheduler's two time operations, behind a seam. Validated intervals are
/// at least an hour, so a test that used the real clock would either sleep an
/// hour or prove nothing; a test installs its own step clock instead.
pub const ScheduleClock = struct {
    ctx: ?*anyopaque = null,
    /// Seconds on a monotonic clock. Only differences matter.
    nowFn: *const fn (ctx: ?*anyopaque, io: std.Io) i64,
    /// Returns when `deadline_s` arrives or `event` is set, whichever comes
    /// first; a null deadline waits for the event alone. A spurious early
    /// return is allowed — the caller rechecks both the version and the clock.
    waitFn: *const fn (
        ctx: ?*anyopaque,
        io: std.Io,
        event: *std.Io.Event,
        deadline_s: ?i64,
    ) std.Io.Cancelable!void,

    pub const real: ScheduleClock = .{ .nowFn = realNow, .waitFn = realWait };

    /// Test seam: the loop only ever exits on shutdown, so a test that wants
    /// `runScheduler` to run its startup pass and return installs this and
    /// gets `error.Canceled` at the first park.
    pub const shutdown_at_first_park: ScheduleClock = .{ .nowFn = realNow, .waitFn = cancelWait };

    fn cancelWait(_: ?*anyopaque, _: std.Io, _: *std.Io.Event, _: ?i64) std.Io.Cancelable!void {
        return error.Canceled;
    }

    /// `boot` rather than `awake`: a box that suspends overnight should still
    /// see its daily interval elapse.
    fn realNow(_: ?*anyopaque, io: std.Io) i64 {
        return std.Io.Clock.boot.now(io).toSeconds();
    }

    fn realWait(
        _: ?*anyopaque,
        io: std.Io,
        event: *std.Io.Event,
        deadline_s: ?i64,
    ) std.Io.Cancelable!void {
        const timeout: std.Io.Timeout = if (deadline_s) |seconds| .{ .deadline = .{
            .raw = .{ .nanoseconds = @as(i96, seconds) * std.time.ns_per_s },
            .clock = .boot,
        } } else .none;
        event.waitTimeout(io, timeout) catch |err| switch (err) {
            error.Timeout => {},
            error.Canceled => return error.Canceled,
        };
    }
};

pub const Manager = struct {
    gpa: Allocator,
    database: *db.Db,
    paths: Paths,
    fetcher: *fetcher.Fetcher,
    /// Read and written only under `schedule_mutex`; `setSchedule` replaces it
    /// while the scheduler is parked.
    update: model.BlocklistUpdate,
    /// Guards `update`, `schedule_version` and `schedule_anchor_s`.
    ///
    /// Lock ordering: innermost. `needsRefresh` takes it while `refresh_lock`
    /// is held, and nothing that holds it takes another manager lock.
    schedule_mutex: std.Io.Mutex,
    /// Bumped by every `setSchedule`. The scheduler reads it before it parks
    /// and again after it wakes: a change that lands in that window is what the
    /// recheck catches, so no wake is lost and none is mistaken for a deadline.
    schedule_version: u64,
    /// When the last refresh pass that RAN completed, on `ScheduleClock`'s
    /// clock. Success, failure and a disk-gate skip all advance it — the
    /// scheduled slot is spent either way and is not retried early. Null until
    /// the startup pass finishes.
    schedule_anchor_s: ?i64,
    /// Sticky once set, so `setSchedule` can never signal into a gap. The loop
    /// resets it under `schedule_mutex` before it recomputes its deadline.
    schedule_event: std.Io.Event,
    schedule_clock: ScheduleClock,
    /// Bounds one download. `std.http.Client` has no per-request deadline, so
    /// the fetch runs under `io.concurrent` against a sleep of this length.
    total_budget: std.Io.Clock.Duration,

    lock: std.Io.RwLock,
    /// Serializes everything that changes the published state — the snapshot,
    /// the compiled files, the runtime columns, the status table — against
    /// every other writer. Never taken by a reader.
    ///
    /// Lock ordering: `refresh_lock` is never acquired while this is held. A
    /// caller that needs both takes `refresh_lock` first.
    writer_lock: std.Io.Mutex,
    /// Serializes refresh passes against each other, and against the
    /// blocklist-directory maintenance in `pruneOrphans`. Held across a
    /// download and a compile, which `writer_lock` deliberately is not, so an
    /// unrelated `reload` never waits out a 300-second fetch.
    ///
    /// Lock ordering: this is taken first, and never while `writer_lock` is
    /// held.
    refresh_lock: std.Io.Mutex,
    current: ?*matcher.Snapshot,
    generation: u64,
    statuses: []SourceStatus,
    /// Owns the `statuses` table. The entries themselves borrow nothing.
    status_arena: std.heap.ArenaAllocator,

    /// The §11.6 disk gate (ruling 17). Set by the composition root after
    /// `init` and before `runScheduler` starts; null disables gating, which is
    /// what every test and `nxdns check` want. Only the scheduler consults it —
    /// see `refreshGated`.
    monitor: ?*disk_monitor.Monitor = null,
    /// The diagnostics store, wired the same way as `monitor` and null
    /// everywhere else. Never touched while a manager lock is held: see
    /// `flushDiagnostics`.
    diagnostics: ?*events.Store = null,
    /// What the locked bodies observed and could not report from where they
    /// stood. Bounded by construction — one slot per storage operation, one
    /// snapshot outcome — and drained by `flushDiagnostics`.
    pending: Pending = .{},
    /// Scheduled refresh passes skipped by the disk gate. The `/api/health`
    /// rollup reads it through `refreshesGated`.
    refreshes_gated: std.atomic.Value(u64) = .init(0),

    pub const Error = error{
        OutOfMemory,
        Canceled,
        /// A filesystem operation on the blocklist directory failed. The
        /// concrete cause is logged at `warn` where it happens: this taxonomy
        /// would otherwise carry two dozen members no caller can act on
        /// differently.
        FileSystem,
        /// The `groups` table changed between listing the groups and reading
        /// their ids. Retrying the reload is the answer, and the caller is the
        /// only one that can decide to.
        GroupSetChanged,
    } || db.Error || matcher.Snapshot.Error;

    /// The result is not copyable afterwards: `status_arena` and `lock` are
    /// addressed through `self`.
    pub fn init(
        gpa: Allocator,
        database: *db.Db,
        paths: Paths,
        fetcher_ptr: *fetcher.Fetcher,
        update: model.BlocklistUpdate,
        total_budget: std.Io.Clock.Duration,
    ) Error!Manager {
        return .{
            .gpa = gpa,
            .database = database,
            .paths = paths,
            .fetcher = fetcher_ptr,
            .update = update,
            .schedule_mutex = .init,
            .schedule_version = 0,
            .schedule_anchor_s = null,
            .schedule_event = .unset,
            .schedule_clock = .real,
            .total_budget = total_budget,
            .lock = .init,
            .writer_lock = .init,
            .refresh_lock = .init,
            .current = null,
            .generation = 0,
            .statuses = &.{},
            .status_arena = .init(gpa),
        };
    }

    pub fn deinit(self: *Manager, io: std.Io) void {
        self.lock.lockUncancelable(io);
        const old = self.current;
        self.current = null;
        self.statuses = &.{};
        self.lock.unlock(io);

        if (old) |snapshot| destroySnapshot(self.gpa, snapshot);
        self.status_arena.deinit();
        self.* = undefined;
    }

    /// Reader side of the swap. The handle holds a shared lock: release it, and
    /// do not retain `snapshot` afterwards.
    pub const Handle = struct {
        snapshot: *const matcher.Snapshot,
        manager: *Manager,

        pub fn release(self: Handle, io: std.Io) void {
            self.manager.lock.unlockShared(io);
        }
    };

    /// `null` before the first successful `reload`. The caller answers
    /// SERVFAIL, or forwards unfiltered, on its own policy — this file does not
    /// decide that.
    pub fn acquire(self: *Manager, io: std.Io) ?Handle {
        self.lock.lockSharedUncancelable(io);
        const snapshot = self.current orelse {
            self.lock.unlockShared(io);
            return null;
        };
        return .{ .snapshot = snapshot, .manager = self };
    }

    /// Copies the status table for the API and for `nxdns check`. Returns the
    /// number of entries written, which is `min(out.len, source count)`.
    ///
    /// The copies are self-contained: `SourceStatus` holds its url and its
    /// error text inline, so the caller may keep them for as long as it likes
    /// and a concurrent reload cannot pull memory out from under them.
    pub fn statusSnapshot(self: *Manager, io: std.Io, out: []SourceStatus) usize {
        self.lock.lockSharedUncancelable(io);
        defer self.lock.unlockShared(io);
        const kept = @min(out.len, self.statuses.len);
        @memcpy(out[0..kept], self.statuses[0..kept]);
        return kept;
    }

    // -----------------------------------------------------------------------
    // diagnostics
    // -----------------------------------------------------------------------

    /// Records one storage operation's failure. Callable from anywhere,
    /// including under both writer mutexes: it touches `pending` only.
    fn noteStorageFailure(
        self: *Manager,
        io: std.Io,
        op: StorageOp,
        comptime fmt: []const u8,
        args: anytype,
    ) void {
        if (self.diagnostics == null) return;
        self.pending.mutex.lockUncancelable(io);
        defer self.pending.mutex.unlock(io);
        const slot = self.pending.storage.getPtr(op);
        slot.failures +|= 1;
        var w: std.Io.Writer = .fixed(&slot.detail);
        w.print(fmt, args) catch {};
        slot.detail_len = @intCast(w.end);
    }

    fn noteStorageSuccess(self: *Manager, io: std.Io, op: StorageOp) void {
        if (self.diagnostics == null) return;
        self.pending.mutex.lockUncancelable(io);
        defer self.pending.mutex.unlock(io);
        self.pending.storage.getPtr(op).succeeded = true;
    }

    /// Records whether a snapshot was published. `reason` null is the post-swap
    /// success; anything else is the pass that could not publish one.
    fn noteSnapshot(self: *Manager, io: std.Io, reason: ?[]const u8) void {
        if (self.diagnostics == null) return;
        self.pending.mutex.lockUncancelable(io);
        defer self.pending.mutex.unlock(io);
        self.pending.snapshot_failed = reason != null;
        const text = reason orelse "";
        const kept = @min(text.len, self.pending.snapshot_detail.len);
        @memcpy(self.pending.snapshot_detail[0..kept], text[0..kept]);
        self.pending.snapshot_detail_len = @intCast(kept);
    }

    /// Drains `pending` and the status table into the store, holding no manager
    /// lock across a store call.
    ///
    /// Called by every pass that can fill either one, and *before that pass
    /// releases the lock serializing it* — `refresh_lock` for a refresh pass,
    /// `writer_lock` for a standalone `reload`. Draining after the release
    /// would let the next pass record its own outcomes on the same entries
    /// first, and two failing passes would reach the store as one occurrence.
    /// The `defer` that calls this is registered after the unlock `defer` for
    /// that reason; defers run last-registered-first.
    ///
    /// It is idempotent: a drained collector reports nothing.
    pub fn flushDiagnostics(self: *Manager, io: std.Io) void {
        const store = self.diagnostics orelse return;
        const now_s = std.Io.Clock.real.now(io).toSeconds();

        var storage: std.EnumArray(StorageOp, Aggregate) = undefined;
        var snapshot_failed: ?bool = null;
        var snapshot_detail: [events.Store.max_detail_len]u8 = undefined;
        var snapshot_detail_len: u16 = 0;
        {
            self.pending.mutex.lockUncancelable(io);
            defer self.pending.mutex.unlock(io);
            storage = self.pending.storage;
            snapshot_failed = self.pending.snapshot_failed;
            snapshot_detail = self.pending.snapshot_detail;
            snapshot_detail_len = self.pending.snapshot_detail_len;
            self.pending.storage = .initFill(.{});
            self.pending.snapshot_failed = null;
            self.pending.snapshot_detail_len = 0;
        }

        var it = storage.iterator();
        while (it.next()) |kv| {
            const op = @tagName(kv.key);
            if (kv.value.failures != 0) {
                var buf: [events.Store.max_detail_len]u8 = undefined;
                const detail = std.fmt.bufPrint(&buf, "{s} ({d} this pass)", .{
                    kv.value.detailText(),
                    kv.value.failures,
                }) catch buf[0..];
                store.report(io, now_s, .blocklist_storage, op, op, .warning, detail);
            } else if (kv.value.succeeded) {
                store.resolve(io, now_s, .blocklist_storage, op);
            }
        }

        if (snapshot_failed) |failed| {
            if (failed) {
                store.report(
                    io,
                    now_s,
                    .blocklist_snapshot,
                    snapshot_key,
                    "blocklist snapshot",
                    .@"error",
                    snapshot_detail[0..snapshot_detail_len],
                );
            } else {
                store.resolve(io, now_s, .blocklist_snapshot, snapshot_key);
            }
        }

        self.flushSourceDiagnostics(io, store, now_s);
    }

    /// One `blocklist.refresh` episode per source, from the status table.
    ///
    /// The table IS the per-source collection the collect-then-flush rule asks
    /// for: `prepareRefresh`, `publishRefresh` and the reload's load outcomes
    /// all write their result into the entry, under locks this cannot take. So
    /// one entry is copied out at a time under the exclusive lock and the store
    /// is called with nothing held.
    ///
    /// The walk is a drain, not an index scan: a reload can replace the whole
    /// table between two iterations, and an index into the table it replaced
    /// would skip or repeat entries. Each round takes the lock, claims the
    /// first entry that still carries pass accounting by copying it out and
    /// zeroing the two fields, and reports it with nothing held. Claiming and
    /// clearing are one locked step, so an outcome is reported once: a table
    /// swapped in mid-drain carries the entries this flush has not claimed yet,
    /// and `installStatuses` folded them in for exactly that reason. The drain
    /// ends when a scan finds nothing left to claim.
    ///
    /// The drain reaches only the sources the table still holds, so the sweep
    /// below is what closes the episode of one that is gone.
    fn flushSourceDiagnostics(self: *Manager, io: std.Io, store: *events.Store, now_s: i64) void {
        drain: while (true) {
            var status: SourceStatus = undefined;
            {
                self.lock.lockUncancelable(io);
                defer self.lock.unlock(io);
                const claimed = for (self.statuses) |*entry| {
                    if (!entry.pass_outcome) continue;
                    status = entry.*;
                    entry.pass_outcome = false;
                    entry.pass_failures = 0;
                    // A source with no diagnostics identity has nothing to
                    // report under, but its accounting is cleared all the same:
                    // left set, it would make every later scan claim it and the
                    // drain would never end.
                    if (entry.event_key_len == 0) continue;
                    break true;
                } else false;
                if (!claimed) break :drain;
            }
            if (!status.state.isRefreshFailure() and status.state != .load_failed) {
                store.resolve(io, now_s, .blocklist_refresh, status.eventKey());
                continue;
            }
            var buf: [events.Store.max_detail_len]u8 = undefined;
            const detail = std.fmt.bufPrint(&buf, "{t}: {s} ({d} this pass)", .{
                status.state,
                status.errorText(),
                status.pass_failures,
            }) catch buf[0..];
            var label_buf: [events.Store.max_subject_label_len]u8 = undefined;
            const label = std.fmt.bufPrint(&label_buf, "{f}", .{
                safe_url.redact(status.urlText()),
            }) catch &label_buf;
            store.report(io, now_s, .blocklist_refresh, status.eventKey(), label, .warning, detail);
        }

        self.resolveDeletedSources(io, store, now_s);
    }

    /// Closes the `blocklist.refresh` episode of a source that no longer exists.
    ///
    /// Nothing else can. An episode of this code is closed by its source
    /// succeeding, and a source deleted through the API or dropped by a config
    /// import never succeeds again: the drain above walks the status table, the
    /// deleted source has no entry in it, and the resolved-row pruning never
    /// touches an active row. Without this the operator keeps a warning about a
    /// list they removed on purpose, and no restart clears it.
    ///
    /// The status table holds every source at every flush site — `refreshAll`
    /// syncs it before it refreshes anything and a reload rebuilds it from the
    /// rows — so its keys are exactly the episodes that may stay open.
    fn resolveDeletedSources(self: *Manager, io: std.Io, store: *events.Store, now_s: i64) void {
        var storage: [events.Store.max_kept_keys][events.Store.max_subject_key_len]u8 = undefined;
        var lens: [events.Store.max_kept_keys]u16 = undefined;
        var len: usize = 0;
        {
            // Shared: this reads the table and changes nothing in it. The keys
            // are copied out because the arena they live in is freed by the
            // next `installStatuses`, and the store is called below with
            // nothing held.
            self.lock.lockSharedUncancelable(io);
            defer self.lock.unlockShared(io);

            // An empty table before the first published snapshot means "no
            // source set has been read yet", not "every source was deleted".
            // Sweeping on it would close every episode the last run left open,
            // and the pass that follows would reopen each one as a new episode
            // with its history reset.
            if (self.generation == 0) return;

            for (self.statuses) |*entry| {
                if (entry.event_key_len == 0) continue;
                // `resolveExcept` refuses a kept list longer than
                // `max_kept_keys`, because it canonicalizes onto the stack.
                // Over that many keyed sources the sweep is skipped whole: the
                // alternative is a truncated kept list, which would close
                // episodes that are still true. A source deleted while the
                // household is over the cap keeps its episode until the count
                // falls back under it.
                if (len == storage.len) return;
                const key = entry.eventKey();
                @memcpy(storage[len][0..key.len], key);
                lens[len] = entry.event_key_len;
                len += 1;
            }
        }

        var kept: [events.Store.max_kept_keys][]const u8 = undefined;
        for (0..len) |i| kept[i] = storage[i][0..lens[i]];
        store.resolveExcept(io, now_s, .blocklist_refresh, kept[0..len]);
    }

    // -----------------------------------------------------------------------
    // reload
    // -----------------------------------------------------------------------

    /// Reads the database and every compiled file, builds a snapshot and swaps
    /// it in.
    ///
    /// A source whose compiled files are missing, unreadable or checksum
    /// mismatched is marked `.load_failed` and left out of the snapshot rather
    /// than failing the whole reload: one bad file must not cost the operator
    /// every other list. `runScheduler` refreshes exactly those sources, so the
    /// state is recorded, surfaced and repaired, never silently accepted.
    ///
    /// A body that is present and checksum-clean but malformed fails the build
    /// (`error.NotSorted`), and the previously published snapshot keeps
    /// serving: nothing is swapped until the new snapshot exists. The status
    /// table keeps describing that snapshot too — the table is rebuilt off to
    /// the side and the load findings are written into it there, so a reload
    /// that never publishes changes neither.
    ///
    /// A standalone reload is its own pass, and `writer_lock` is what serializes
    /// it against every other writer of the status table. So it flushes inside
    /// that lock: the load outcomes it wrote at the swap are drained before any
    /// other pass can add its own to the same entries, which is what keeps two
    /// failing passes two occurrences instead of one.
    pub fn reload(self: *Manager, io: std.Io) Error!void {
        // Must not be entered with `writer_lock` held.
        self.writer_lock.lockUncancelable(io);
        defer self.writer_lock.unlock(io);
        // Registered after the unlock so it runs before it, and `defer` and not
        // straight-line code after the call: a reload that fails has already
        // collected the outcomes that explain why, and leaving them pending
        // would hold them until some later pass flushed them under the wrong
        // timestamp.
        defer self.flushDiagnostics(io);
        return self.reloadLocked(io);
    }

    /// `reload` without the flush, for a caller that is inside a pass with a
    /// flush of its own. One pass flushes once: flushing here as well would
    /// split the pass's outcomes across two reports.
    fn reloadCollecting(self: *Manager, io: std.Io) Error!void {
        // Must not be entered with `writer_lock` held.
        self.writer_lock.lockUncancelable(io);
        defer self.writer_lock.unlock(io);
        try self.reloadLocked(io);
    }

    fn reloadLocked(self: *Manager, io: std.Io) Error!void {
        var rows = try sources_repo.listSourceRows(self.database, self.gpa);
        defer rows.deinit(self.gpa);
        defer sources_repo.freeSourceRows(self.gpa, rows.items);

        // The table this reload will publish, built where no reader can see it.
        // It is installed in the swap below or freed unpublished, so a reload
        // that fails leaves the previous table describing the previous
        // snapshot — including the entry of a source deleted from the database,
        // which that snapshot still enforces.
        var candidate: ?StatusTable = try self.buildStatusTable(io, rows.items);
        errdefer if (candidate) |*table| table.deinit();

        var dir = try self.openDir(io, .{});
        defer dir.close(io);

        const sources = try self.gpa.alloc(model.BlocklistSource, rows.items.len);
        defer self.gpa.free(sources);
        const source_ids = try self.gpa.alloc(i64, rows.items.len);
        defer self.gpa.free(source_ids);
        const compiled = try self.gpa.alloc(?matcher.Snapshot.Compiled, rows.items.len);
        defer self.gpa.free(compiled);

        // The file contents outlive the header stripping and are freed once the
        // snapshot has copied what it needs into its own arena.
        var bodies: std.ArrayList([]u8) = .empty;
        defer {
            for (bodies.items) |body| self.gpa.free(body);
            bodies.deinit(self.gpa);
        }

        // What this reload found, per source. It is applied to the status table
        // only if the snapshot it describes is published: everything below here
        // can still fail, and a status table describing a snapshot nobody
        // serves is worse than one describing the previous one.
        const outcomes = try self.gpa.alloc(LoadOutcome, rows.items.len);
        defer self.gpa.free(outcomes);

        var loaded: usize = 0;
        for (rows.items, sources, source_ids, compiled, outcomes) |row, *source, *source_id, *slot, *outcome| {
            source_id.* = row.id;
            // `is_suggested` is a UI hint the snapshot never reads.
            source.* = .{ .url = row.url, .name = row.name, .enabled = row.enabled };
            outcome.* = if (row.enabled) try self.loadSource(io, dir, row, &bodies) else .disabled;
            switch (outcome.*) {
                .loaded => |body| {
                    slot.* = body;
                    loaded += 1;
                },
                // Not loadable and therefore not enforced. Saying so here is
                // what keeps `Snapshot.build`'s `MissingCompiledSource` for the
                // case it is meant for: a caller that forgot to read a body.
                .disabled, .failed => {
                    slot.* = null;
                    source.enabled = false;
                },
            }
        }

        var groups = try groups_repo.listGroups(self.database, self.gpa);
        defer groups.deinit(self.gpa);
        defer groups_repo.freeGroups(self.gpa, groups.items);

        const group_ids = try self.groupIds(groups.items);
        defer self.gpa.free(group_ids);

        var group_sources = try groups_repo.listGroupSources(self.database, self.gpa);
        defer group_sources.deinit(self.gpa);
        defer groups_repo.freeGroupSources(self.gpa, group_sources.items);

        var rule_rows = try rules_repo.listRules(self.database, self.gpa);
        defer rule_rows.deinit(self.gpa);
        defer rules_repo.freeRules(self.gpa, rule_rows.items);

        var clients = try clients_repo.listClients(self.database, self.gpa);
        defer clients.deinit(self.gpa);
        defer clients_repo.freeClients(self.gpa, clients.items);

        var prefixes = try clients_repo.listClientPrefixes(self.database, self.gpa);
        defer prefixes.deinit(self.gpa);
        defer clients_repo.freeClientPrefixes(self.gpa, prefixes.items);

        // Query names are attacker-supplied, so a fixed seed would make
        // probe-chain flooding computable offline.
        var seed_bytes: [8]u8 = undefined;
        io.random(&seed_bytes);

        const generation = self.generation + 1;
        const snapshot = try self.gpa.create(matcher.Snapshot);
        errdefer self.gpa.destroy(snapshot);
        snapshot.* = try matcher.Snapshot.build(self.gpa, .{
            .groups = groups.items,
            .group_ids = group_ids,
            .group_sources = group_sources.items,
            .sources = sources,
            .source_ids = source_ids,
            .rules = rule_rows.items,
            .clients = clients.items,
            .prefixes = prefixes.items,
            .compiled = compiled,
            .seed = std.mem.readInt(u64, &seed_bytes, .little),
            .generation = generation,
        });

        // Read before the swap: once `current` points at it, this snapshot
        // belongs to the readers and to whichever writer replaces it next.
        const memory_bytes = snapshot.memoryBytes();

        // The snapshot, the status table and the load facts land together, so a
        // reader never sees a status table describing anything but the
        // published snapshot.
        self.lock.lockUncancelable(io);
        const old = self.current;
        self.current = snapshot;
        self.generation = generation;
        applyLoadOutcomes(candidate.?.items, rows.items, outcomes);
        self.installStatuses(candidate.?);
        candidate = null;
        self.lock.unlock(io);

        // After `unlock`: no reader can still hold the old snapshot here, and
        // `writer_lock` keeps every other writer out of this sequence.
        if (old) |previous| destroySnapshot(self.gpa, previous);

        log.info("blocklist snapshot generation {d}: {d} of {d} sources loaded, {d} bytes", .{
            generation,
            loaded,
            rows.items.len,
            memory_bytes,
        });
        self.noteSnapshot(io, null);
    }

    /// What one enabled source contributes to the snapshot being built. Nothing
    /// here touches the status table: the outcome is data until the swap
    /// commits it.
    fn loadSource(
        self: *Manager,
        io: std.Io,
        dir: std.Io.Dir,
        row: sources_repo.SourceRow,
        bodies: *std.ArrayList([]u8),
    ) Error!LoadOutcome {
        const stored = row.checksum orelse
            // No stored checksum means no successful compile has ever
            // happened. A fresh install is here on every source.
            return .{ .failed = .{ .state = .never_fetched, .text = "" } };

        var list_buf: [name_buf_len]u8 = undefined;
        var wild_buf: [name_buf_len]u8 = undefined;
        var allow_buf: [name_buf_len]u8 = undefined;
        const list_name = compiledName(&list_buf, row.id, ".list");
        const wild_name = compiledName(&wild_buf, row.id, ".wild");
        const allow_name = compiledName(&allow_buf, row.id, ".allow");

        // Reserved before the reads, so no buffer can be orphaned by a failing
        // append: `bodies` owns each one from the moment it is read.
        try bodies.ensureUnusedCapacity(self.gpa, 3);

        // `error.Canceled` is the one-shot signal that this task is being torn
        // down, and it is consumed by whoever catches it. Recording it as a
        // load failure would spend it on a status row that reads "Canceled",
        // publish a snapshot with this source missing, and let the shutdown
        // carry on as if nothing had asked it to stop.
        const list_bytes = dir.readFileAlloc(io, list_name, self.gpa, .limited(max_compiled_bytes)) catch |err| {
            if (err == error.OutOfMemory) return error.OutOfMemory;
            if (err == error.Canceled) return error.Canceled;
            return loadFailure(row, list_name, err);
        };
        bodies.appendAssumeCapacity(list_bytes);

        const wild_bytes = dir.readFileAlloc(io, wild_name, self.gpa, .limited(max_compiled_bytes)) catch |err| {
            if (err == error.OutOfMemory) return error.OutOfMemory;
            if (err == error.Canceled) return error.Canceled;
            return loadFailure(row, wild_name, err);
        };
        bodies.appendAssumeCapacity(wild_bytes);

        // A missing `.allow` file is an empty allow body, not a failure. The
        // digest still covers three bodies, the third of them empty, so a
        // source whose list carries no `@@` line matches whether its empty
        // `.allow` file survived or not.
        const allow_bytes: []const u8 = blk: {
            const read = dir.readFileAlloc(io, allow_name, self.gpa, .limited(max_compiled_bytes)) catch |err| {
                if (err == error.OutOfMemory) return error.OutOfMemory;
                if (err == error.Canceled) return error.Canceled;
                if (err == error.FileNotFound) break :blk "";
                return loadFailure(row, allow_name, err);
            };
            bodies.appendAssumeCapacity(read);
            break :blk read;
        };

        const list_body = stripHeader(list_bytes);
        const wild_body = stripHeader(wild_bytes);
        const allow_body = stripHeader(allow_bytes);

        // The checksum covers the three bodies together, so a crash between the
        // `replace` calls — a new `.list` beside an old `.wild` — is caught
        // here and refreshed, not served as a half-updated list.
        if (!std.mem.eql(u8, stored, &bodyChecksum(list_body, wild_body, allow_body))) {
            log.warn("blocklist {f}: compiled files do not match the stored checksum", .{SourceLabel.of(row)});
            return .{ .failed = .{ .state = .load_failed, .text = "ChecksumMismatch" } };
        }

        return .{ .loaded = .{
            .list_body = list_body,
            .wild_body = wild_body,
            .allow_body = allow_body,
        } };
    }

    // -----------------------------------------------------------------------
    // refresh
    // -----------------------------------------------------------------------

    /// Downloads, compiles and atomically replaces the compiled files of one
    /// source, then writes its runtime columns.
    ///
    /// Returns `true` only when the compiled files were replaced. Unchanged
    /// content (equal checksum) and every recorded failure return `false`; the
    /// reason for a failure is in the status table, not in the return value.
    ///
    /// The status entry is found by row id, so a source added since the last
    /// `reload` has nowhere to record its outcome. `refreshAll` syncs the table
    /// before it refreshes anything, which is why `POST /api/blocklists/update`
    /// and the scheduler both enter through `refreshAll`.
    pub fn refreshSource(self: *Manager, io: std.Io, row: sources_repo.SourceRow) Error!bool {
        self.refresh_lock.lockUncancelable(io);
        defer self.refresh_lock.unlock(io);
        // Registered after the unlock, so it runs before it: a pass drains its
        // own outcomes while it still holds `refresh_lock`. `defer` at all, so
        // a refresh that fails outright still reports what it collected instead
        // of leaving it for an unrelated later flush.
        defer self.flushDiagnostics(io);
        return self.refreshSourceLocked(io, row);
    }

    /// Assumes `refresh_lock`. Takes `writer_lock` itself, for the publish half
    /// alone.
    fn refreshSourceLocked(self: *Manager, io: std.Io, row: sources_repo.SourceRow) Error!bool {
        // The previous entry describes the compiled files that are still on
        // disk, and a failed refresh leaves them serving. Starting from a blank
        // status would erase `last_success` and the counters of a list that is
        // still being enforced.
        var status: SourceStatus = self.priorStatus(io, row.id) orelse .{ .id = row.id };
        status.setUrl(row.url);
        status.last_attempt = std.Io.Clock.real.now(io).toSeconds();

        var dir = try self.openDir(io, .{});
        defer dir.close(io);

        var raw_buf: [name_buf_len]u8 = undefined;
        var list_tmp_buf: [name_buf_len]u8 = undefined;
        var wild_tmp_buf: [name_buf_len]u8 = undefined;
        var allow_tmp_buf: [name_buf_len]u8 = undefined;
        const raw_name = compiledName(&raw_buf, row.id, ".raw.tmp");
        const tmp: TempNames = .{
            .list = compiledName(&list_tmp_buf, row.id, ".list.tmp"),
            .wild = compiledName(&wild_tmp_buf, row.id, ".wild.tmp"),
            .allow = compiledName(&allow_tmp_buf, row.id, ".allow.tmp"),
        };

        // Installed before the calls that create these files, not after: an
        // `error.Canceled` or `error.OutOfMemory` returned straight out of
        // `download` or `compileTo` would outrun a later `defer` and leave a
        // temporary behind. Deleting a name that was never created is a no-op.
        defer self.deleteQuietly(io, dir, raw_name);
        defer self.deleteQuietly(io, dir, tmp.list);
        defer self.deleteQuietly(io, dir, tmp.wild);
        defer self.deleteQuietly(io, dir, tmp.allow);

        // The half that takes the time: one download of up to `total_budget`
        // and one compile of everything it returned. `refresh_lock` alone is
        // held here, so a rule save, a settings change or any other web
        // mutation that ends in `reload` runs beside it instead of behind it.
        const prepared = try self.prepareRefresh(io, dir, row, &status, raw_name, tmp);

        // The half that publishes. The compiled files, the runtime columns and
        // the status entry land under one `writer_lock`, so a reload never
        // reads new files beside a status entry describing the previous ones.
        self.writer_lock.lockUncancelable(io);
        defer self.writer_lock.unlock(io);

        const replaced = try self.publishRefresh(io, dir, row, &status, prepared, tmp);
        self.commitStatus(io, status);
        return replaced;
    }

    /// Every enabled source, one at a time, then one `reload`. A failing source
    /// never stops the pass: it would hide every source behind it.
    ///
    /// This returns an error only when nothing could be done at all — out of
    /// memory, an unreachable database, an unusable blocklist directory. A
    /// source that failed to download or compile is a successful pass with a
    /// non-`ok` status.
    pub fn refreshAll(self: *Manager, io: std.Io) Error!void {
        self.refresh_lock.lockUncancelable(io);
        defer self.refresh_lock.unlock(io);
        // Inside `refresh_lock`, by being registered after the unlock: see
        // `refreshSource`.
        defer self.flushDiagnostics(io);

        var rows = try sources_repo.listSourceRows(self.database, self.gpa);
        defer rows.deinit(self.gpa);
        defer sources_repo.freeSourceRows(self.gpa, rows.items);

        try self.syncStatuses(io, rows.items);

        for (rows.items) |row| {
            if (!row.enabled) continue;
            _ = try self.refreshSourceLocked(io, row);
        }
        // `reload` takes `writer_lock`, which the pass has been careful not to
        // hold: the order is `refresh_lock` first, always. The collecting
        // variant, because the `defer` above is this pass's one flush.
        return self.reloadCollecting(io);
    }

    /// The three temporary files one refresh compiles into, before the header
    /// is prepended and each is renamed over the file it replaces.
    const TempNames = struct {
        list: []const u8,
        wild: []const u8,
        allow: []const u8,
    };

    /// What the fetch-and-compile half of a refresh produced. `.failed` needs
    /// no publish and has already recorded why in the status entry.
    const Prepared = union(enum) {
        failed,
        compiled: struct {
            format: parsers.Format,
            result: compiler.Result,
        },
    };

    /// Downloads one source and compiles it into the temporary files.
    ///
    /// Assumes `refresh_lock` and must not be called with `writer_lock` held:
    /// this is the part that takes seconds, and nothing here touches the
    /// published state.
    fn prepareRefresh(
        self: *Manager,
        io: std.Io,
        dir: std.Io.Dir,
        row: sources_repo.SourceRow,
        status: *SourceStatus,
        raw_name: []const u8,
        tmp: TempNames,
    ) Error!Prepared {
        self.download(io, dir, raw_name, row) catch |err| switch (err) {
            error.OutOfMemory => return error.OutOfMemory,
            error.Canceled => return error.Canceled,
            else => {
                self.reportFetchFailure(row, status, err);
                return .failed;
            },
        };

        const format = self.detectFormat(io, dir, raw_name) catch |err| switch (err) {
            error.OutOfMemory => return error.OutOfMemory,
            error.Canceled => return error.Canceled,
            else => {
                self.reportCompileFailure(row, status, err);
                return .failed;
            },
        };

        const result = self.compileTo(io, dir, raw_name, format, tmp) catch |err| switch (err) {
            error.OutOfMemory => return error.OutOfMemory,
            error.Canceled => return error.Canceled,
            else => {
                self.reportCompileFailure(row, status, err);
                return .failed;
            },
        };

        if (rejectedWithoutEntries(result.counts)) {
            self.reportEmptyCompile(row, status, result.counts);
            return .failed;
        }

        return .{ .compiled = .{ .format = format, .result = result } };
    }

    /// Installs what `prepareRefresh` produced and writes the runtime columns.
    /// Returns `true` only when the compiled files were replaced.
    ///
    /// Assumes `writer_lock`. Reading the files on disk belongs here and not in
    /// the half above: they are published under this lock.
    fn publishRefresh(
        self: *Manager,
        io: std.Io,
        dir: std.Io.Dir,
        row: sources_repo.SourceRow,
        status: *SourceStatus,
        prepared: Prepared,
        tmp: TempNames,
    ) Error!bool {
        const compiled = switch (prepared) {
            .failed => return false,
            .compiled => |value| value,
        };

        const now = std.Io.Clock.real.now(io).toSeconds();

        // Recompiling identical content into new files would invalidate the
        // snapshot for nothing. The files on disk must actually carry that
        // content: if a reload found them corrupt and excluded the source, a
        // re-download of unchanged upstream bytes is the one chance to repair
        // them, and skipping the rewrite here would leave filtering off for
        // good.
        if (row.checksum) |stored| {
            if (std.mem.eql(u8, stored, &compiled.result.checksum) and
                self.diskBodiesMatch(io, dir, row.id, stored))
            {
                // Every count comes from the compile that just ran, not from
                // the row. The two skip counts have to: a skipped line lands in
                // no body, so a list that changed only its regex or
                // browser-syntax lines reaches here with a stale row. The three
                // written counts equal the row's anyway once the digest is
                // framed, so reading them from the compile costs nothing and
                // leaves no field whose freshness rests on an argument about
                // what the checksum covers.
                try sources_repo.updateSourceStats(self.database, row.id, .{
                    .last_updated = now,
                    .domain_count = compiled.result.counts.domains,
                    .wildcard_count = compiled.result.counts.wildcards,
                    .exception_count = compiled.result.counts.exceptions,
                    .skipped_regex_count = compiled.result.counts.skipped_regex,
                    .skipped_unsupported_count = compiled.result.counts.skipped_unsupported,
                    .checksum = stored,
                });
                status.succeed(now, compiled.result.counts);
                return false;
            }
        }

        const header: Header = .{
            .url = row.url,
            .format = compiled.format,
            .fetched_at = now,
            .counts = compiled.result.counts,
            .checksum = &compiled.result.checksum,
        };
        self.publish(io, dir, row.id, header, tmp) catch |err| switch (err) {
            error.OutOfMemory => return error.OutOfMemory,
            error.Canceled => return error.Canceled,
            else => {
                self.reportCompileFailure(row, status, err);
                return false;
            },
        };

        try sources_repo.updateSourceStats(self.database, row.id, .{
            .last_updated = now,
            .domain_count = compiled.result.counts.domains,
            .wildcard_count = compiled.result.counts.wildcards,
            .exception_count = compiled.result.counts.exceptions,
            .skipped_regex_count = compiled.result.counts.skipped_regex,
            .skipped_unsupported_count = compiled.result.counts.skipped_unsupported,
            .checksum = &compiled.result.checksum,
        });
        status.succeed(now, compiled.result.counts);
        return true;
    }

    /// The body goes to a temporary file, never to memory: `max_body_bytes` is
    /// 64 MiB and the memory budget has no room for it beside two snapshots.
    ///
    /// It takes the whole row rather than the url alone because its two log
    /// lines name the source by its id and name, which only the row carries.
    fn download(
        self: *Manager,
        io: std.Io,
        dir: std.Io.Dir,
        raw_name: []const u8,
        row: sources_repo.SourceRow,
    ) !void {
        const file = try dir.createFile(io, raw_name, .{ .permissions = .fromMode(0o600) });
        defer file.close(io);

        const buffer = try self.gpa.alloc(u8, io_buf_len);
        defer self.gpa.free(buffer);

        var fw = file.writer(io, buffer);
        const result = self.fetchWithin(io, row.url, &fw.interface) catch |err| {
            // `fetcher.Error.Unexpected` is what a failing sink surfaces as;
            // the concrete cause is on this writer, which the fetcher does not
            // own.
            if (fw.err) |cause| return cause;
            if (err == error.HttpStatus) {
                if (self.fetcher.last_status) |status| {
                    log.warn("blocklist {f}: http status {d}", .{ SourceLabel.of(row), @intFromEnum(status) });
                }
            }
            return err;
        };
        try fw.interface.flush();
        // The compile reads this file back; the bytes must be there, not in a
        // buffer this function is about to drop.
        try file.sync(io);

        log.debug("blocklist {f}: downloaded {d} bytes", .{ SourceLabel.of(row), result.bytes_read });
    }

    /// `std.http.Client` has no per-request deadline, so the whole exchange
    /// races a sleep and the loser is canceled — milestone 3's pattern.
    fn fetchWithin(
        self: *Manager,
        io: std.Io,
        url: []const u8,
        w: *std.Io.Writer,
    ) fetcher.Error!fetcher.Result {
        var outcomes: [2]Outcome = undefined;
        var race: std.Io.Select(Outcome) = .init(io, &outcomes);
        defer race.cancelDiscard();

        race.concurrent(.fetch, fetcher.Fetcher.fetch, .{ self.fetcher, io, url, w }) catch |err| switch (err) {
            error.ConcurrencyUnavailable => return error.SystemResources,
        };
        race.concurrent(.expiry, expire, .{ io, self.total_budget }) catch |err| switch (err) {
            error.ConcurrencyUnavailable => return error.SystemResources,
        };

        switch (try race.await()) {
            .fetch => |result| return result,
            .expiry => |result| {
                // A canceled sleep means this task is being torn down, not that
                // the download is slow.
                try result;
                return error.Timeout;
            },
        }
    }

    fn detectFormat(
        self: *Manager,
        io: std.Io,
        dir: std.Io.Dir,
        raw_name: []const u8,
    ) !parsers.Format {
        const file = try dir.openFile(io, raw_name, .{});
        defer file.close(io);

        const buffers = try self.gpa.alloc(u8, io_buf_len + sample_buf_len);
        defer self.gpa.free(buffers);

        var fr = file.reader(io, buffers[0..io_buf_len]);
        var sample: std.Io.Writer = .fixed(buffers[io_buf_len..]);
        collectSample(&fr.interface, &sample) catch |err| switch (err) {
            error.ReadFailed => return fr.err orelse err,
            // `sample_buf_len` holds every line `collectSample` can emit, so a
            // full buffer means the sample is complete.
            error.WriteFailed => {},
        };
        return parsers.detectFormat(sample.buffered());
    }

    /// Compiles into three plain temporary files. The compiled bodies cannot go
    /// straight into the final files: the header carries counts that only exist
    /// once the whole input has been compiled, and the loader requires the
    /// header first.
    fn compileTo(
        self: *Manager,
        io: std.Io,
        dir: std.Io.Dir,
        raw_name: []const u8,
        format: parsers.Format,
        tmp: TempNames,
    ) !compiler.Result {
        const raw = try dir.openFile(io, raw_name, .{});
        defer raw.close(io);
        const list_file = try dir.createFile(io, tmp.list, .{ .permissions = .fromMode(0o600) });
        defer list_file.close(io);
        const wild_file = try dir.createFile(io, tmp.wild, .{ .permissions = .fromMode(0o600) });
        defer wild_file.close(io);
        const allow_file = try dir.createFile(io, tmp.allow, .{ .permissions = .fromMode(0o600) });
        defer allow_file.close(io);

        const buffers = try self.gpa.alloc(u8, 4 * io_buf_len);
        defer self.gpa.free(buffers);

        var fr = raw.reader(io, buffers[0..io_buf_len]);
        var list_w = list_file.writer(io, buffers[io_buf_len .. 2 * io_buf_len]);
        var wild_w = wild_file.writer(io, buffers[2 * io_buf_len .. 3 * io_buf_len]);
        var allow_w = allow_file.writer(io, buffers[3 * io_buf_len ..]);

        const result = compiler.compile(
            self.gpa,
            &fr.interface,
            format,
            &list_w.interface,
            &wild_w.interface,
            &allow_w.interface,
        ) catch |err| switch (err) {
            // `compiler.Error` names the direction; the concrete cause is on
            // the stream that failed.
            error.ReadFailed => return fr.err orelse err,
            error.WriteFailed => return list_w.err orelse
                (wild_w.err orelse (allow_w.err orelse err)),
            else => return err,
        };

        try list_w.interface.flush();
        try wild_w.interface.flush();
        try allow_w.interface.flush();
        try list_file.sync(io);
        try wild_file.sync(io);
        try allow_file.sync(io);
        return result;
    }

    /// Writes header + body into each final file through `createFileAtomic` +
    /// `replace`, so a crash mid-write can never leave a half-list that would
    /// load as a valid, shorter blocklist.
    fn publish(
        self: *Manager,
        io: std.Io,
        dir: std.Io.Dir,
        id: i64,
        header: Header,
        tmp: TempNames,
    ) !void {
        const buffers = try self.gpa.alloc(u8, 2 * io_buf_len);
        defer self.gpa.free(buffers);

        var list_buf: [name_buf_len]u8 = undefined;
        var wild_buf: [name_buf_len]u8 = undefined;
        var allow_buf: [name_buf_len]u8 = undefined;
        try publishOne(io, dir, compiledName(&list_buf, id, ".list"), tmp.list, header, buffers);
        try publishOne(io, dir, compiledName(&wild_buf, id, ".wild"), tmp.wild, header, buffers);
        try publishOne(io, dir, compiledName(&allow_buf, id, ".allow"), tmp.allow, header, buffers);
    }

    fn publishOne(
        io: std.Io,
        dir: std.Io.Dir,
        dest: []const u8,
        body_name: []const u8,
        header: Header,
        buffers: []u8,
    ) !void {
        const body = try dir.openFile(io, body_name, .{});
        defer body.close(io);

        var af = try dir.createFileAtomic(io, dest, .{
            .permissions = .fromMode(0o600),
            .replace = true,
        });
        defer af.deinit(io);

        var fr = body.reader(io, buffers[0..io_buf_len]);
        var fw = af.file.writer(io, buffers[io_buf_len..]);

        header.write(&fw.interface) catch return fw.err orelse error.WriteFailed;
        _ = fr.interface.streamRemaining(&fw.interface) catch
            return fr.err orelse (fw.err orelse error.WriteFailed);
        fw.interface.flush() catch return fw.err orelse error.WriteFailed;

        // Before `replace`, which closes the file: the rename must publish
        // durable bytes, not an empty file with the content still in the page
        // cache.
        try af.file.sync(io);
        try af.replace(io);
    }

    /// Whether the compiled files on disk hash to `expected`. A missing,
    /// unreadable or corrupt file answers false, which sends the caller down
    /// the rewrite path — the only path that can repair it.
    ///
    /// A missing `.allow` file is the one exception, and it is the same one
    /// `loadSource` makes: it reads as an empty allow body, which is what a list
    /// with no `@@` line compiles to anyway. Answering false there would rewrite
    /// such a list on every refresh for no change in content.
    fn diskBodiesMatch(self: *Manager, io: std.Io, dir: std.Io.Dir, id: i64, expected: []const u8) bool {
        var list_buf: [name_buf_len]u8 = undefined;
        var wild_buf: [name_buf_len]u8 = undefined;
        var allow_buf: [name_buf_len]u8 = undefined;
        const limit: std.Io.Limit = .limited(max_compiled_bytes);

        const list_bytes = dir.readFileAlloc(io, compiledName(&list_buf, id, ".list"), self.gpa, limit) catch
            return false;
        defer self.gpa.free(list_bytes);
        const wild_bytes = dir.readFileAlloc(io, compiledName(&wild_buf, id, ".wild"), self.gpa, limit) catch
            return false;
        defer self.gpa.free(wild_bytes);

        const allow_bytes = dir.readFileAlloc(io, compiledName(&allow_buf, id, ".allow"), self.gpa, limit) catch |err|
            if (err == error.FileNotFound) @as([]u8, &.{}) else return false;
        defer self.gpa.free(allow_bytes);

        return compiledBodiesMatch(list_bytes, wild_bytes, allow_bytes, expected);
    }

    fn reportFetchFailure(
        self: *Manager,
        row: sources_repo.SourceRow,
        status: *SourceStatus,
        err: anyerror,
    ) void {
        _ = self;
        log.warn("blocklist {f}: download failed: {s}", .{ SourceLabel.of(row), @errorName(err) });
        status.fail(.fetch_failed, @errorName(err));
    }

    fn reportCompileFailure(
        self: *Manager,
        row: sources_repo.SourceRow,
        status: *SourceStatus,
        err: anyerror,
    ) void {
        _ = self;
        log.warn("blocklist {f}: compile failed: {s}", .{ SourceLabel.of(row), @errorName(err) });
        status.fail(.compile_failed, @errorName(err));
    }

    fn reportEmptyCompile(
        self: *Manager,
        row: sources_repo.SourceRow,
        status: *SourceStatus,
        counts: compiler.Counts,
    ) void {
        _ = self;
        var buf: [max_error_len]u8 = undefined;
        const text = std.fmt.bufPrint(
            &buf,
            "NoValidEntries invalid={d} unsupported={d} long_lines={d}",
            .{ counts.invalid, counts.skipped_unsupported, counts.long_lines },
        ) catch "NoValidEntries";
        log.warn("blocklist {f}: {s}", .{ SourceLabel.of(row), text });
        status.fail(.no_valid_entries, text);
    }

    // -----------------------------------------------------------------------
    // scheduling
    // -----------------------------------------------------------------------

    /// Loads at startup, refreshes only what needs it, then sleeps
    /// `update.interval_hours` between full passes. Returns on
    /// `error.Canceled`.
    ///
    /// A cold restart must not re-download every list and a boot loop must not
    /// become a download loop, so the startup pass refreshes a source only when
    /// it has no usable compiled files or its `last_updated` is older than the
    /// interval.
    ///
    /// `update.enabled == false` parks after the startup pass; manual refresh
    /// through `refreshAll` still works, and a later `setSchedule` wakes the
    /// loop rather than needing a restart.
    pub fn runScheduler(self: *Manager, io: std.Io) std.Io.Cancelable!void {
        // Ahead of the pass, not after it. This is the sweep that collects what
        // a killed process left behind: a `.raw.tmp` as large as the body the
        // dead refresh was writing, and the compiled files of a source deleted
        // while the server was down. Both are bytes the pass below is about to
        // ask the same filesystem for.
        try self.sweepOrphans(io);

        // `startupPass` flushes its own outcomes before it releases
        // `refresh_lock`, so the only flush left here is the one the failure
        // note below needs.
        self.startupPass(io) catch |err| switch (err) {
            error.Canceled => return error.Canceled,
            else => {
                log.warn("blocklist startup pass failed: {s}", .{@errorName(err)});
                self.noteSnapshot(io, @errorName(err));
                self.flushDiagnostics(io);
            },
        };
        // The startup pass ran, so it anchors the schedule — including when
        // updates are disabled, so a later enable measures its first interval
        // from real work rather than from the moment the operator flipped the
        // switch.
        self.anchorNow(io);

        while (true) {
            // One hold: read the version, reset the sticky event, and take the
            // schedule the deadline is computed from. A `setSchedule` that
            // lands after this reset completes the wait below at once, and the
            // version recheck decides whether the wake meant anything.
            self.schedule_mutex.lockUncancelable(io);
            const version = self.schedule_version;
            self.schedule_event.reset();
            const enabled = self.update.enabled;
            const interval_s = model.updateIntervalSeconds(self.update);
            const anchor = self.schedule_anchor_s;
            self.schedule_mutex.unlock(io);

            const now_s = self.schedule_clock.nowFn(self.schedule_clock.ctx, io);
            // Disabled parks on the event alone. The task still exits only on
            // shutdown, exactly as it did when it returned here.
            const deadline_s: ?i64 = if (enabled) (anchor orelse now_s) + interval_s else null;

            if (deadline_s == null or now_s < deadline_s.?) {
                try self.schedule_clock.waitFn(self.schedule_clock.ctx, io, &self.schedule_event, deadline_s);
                // Either the schedule changed under us or the wait was
                // spurious; recompute from the top rather than guess.
                if (self.scheduleVersion(io) != version) continue;
                if (deadline_s == null) continue;
                if (self.schedule_clock.nowFn(self.schedule_clock.ctx, io) < deadline_s.?) continue;
            }

            try self.scheduledPass(io);
            self.anchorNow(io);
        }
    }

    /// Installs a new blocklist-update schedule and wakes the scheduler. The
    /// anchor is untouched: the next refresh is due one NEW interval after the
    /// last pass that ran, which the loop refreshes immediately when that
    /// moment is already past.
    pub fn setSchedule(self: *Manager, io: std.Io, enabled: bool, interval_hours: u16) void {
        self.schedule_mutex.lockUncancelable(io);
        self.update = .{ .enabled = enabled, .interval_hours = interval_hours };
        self.schedule_version += 1;
        self.schedule_mutex.unlock(io);

        self.schedule_event.set(io);
    }

    /// The live schedule. Every reader outside the scheduler loop goes through
    /// here, so none of them reads `update` while `setSchedule` writes it.
    pub fn schedule(self: *Manager, io: std.Io) model.BlocklistUpdate {
        self.schedule_mutex.lockUncancelable(io);
        defer self.schedule_mutex.unlock(io);
        return self.update;
    }

    fn scheduleVersion(self: *Manager, io: std.Io) u64 {
        self.schedule_mutex.lockUncancelable(io);
        defer self.schedule_mutex.unlock(io);
        return self.schedule_version;
    }

    fn anchorNow(self: *Manager, io: std.Io) void {
        const now_s = self.schedule_clock.nowFn(self.schedule_clock.ctx, io);
        self.schedule_mutex.lockUncancelable(io);
        self.schedule_anchor_s = now_s;
        self.schedule_mutex.unlock(io);
    }

    /// What one elapsed interval does. Split from the loop above so a test can
    /// run the pass without waiting the interval out; nothing in production
    /// calls it but `runScheduler`.
    pub fn scheduledPass(self: *Manager, io: std.Io) std.Io.Cancelable!void {
        // Ahead of the gate as well as ahead of the refresh: the sweep only
        // unlinks, so it is the one thing here that can give a critically full
        // disk room back, and gating it would keep the residue that helped fill
        // the disk in the first place.
        try self.sweepOrphans(io);
        if (self.refreshGated()) return;
        // `refreshAll` flushes the pass itself, so the only flush left here is
        // the one the failure note below needs: flushing unconditionally would
        // report every outcome of the pass a second time.
        self.refreshAll(io) catch |err| switch (err) {
            error.Canceled => return error.Canceled,
            else => {
                log.warn("blocklist refresh pass failed: {s}", .{@errorName(err)});
                self.noteSnapshot(io, @errorName(err));
                self.flushDiagnostics(io);
            },
        };
    }

    /// `pruneOrphans` with its failure absorbed. Leftover bytes under
    /// `<data_dir>/blocklists/` are not an outage, and a sweep that could not
    /// read the directory must not cost the household the refresh pass behind
    /// it — let alone the server. Cancellation is the one outcome that
    /// propagates, because it means shutdown.
    ///
    /// Taken from outside every `*Locked` body: `pruneOrphans` takes both
    /// writer mutexes itself and neither is reentrant.
    fn sweepOrphans(self: *Manager, io: std.Io) std.Io.Cancelable!void {
        if (self.pruneOrphans(io)) {
            self.noteStorageSuccess(io, .sweep);
        } else |err| switch (err) {
            error.Canceled => return error.Canceled,
            else => {
                log.warn("pruning orphaned blocklist files failed: {s}", .{@errorName(err)});
                self.noteStorageFailure(io, .sweep, "pruning orphaned blocklist files failed: {s}", .{@errorName(err)});
            },
        }
        self.flushDiagnostics(io);
    }

    /// The §11.6 gate, consulted by scheduled passes only (ruling 17). A
    /// download writes tens of megabytes into the blocklist directory and the
    /// compile writes as much again, which is exactly the "non-essential write"
    /// a critically full disk must not take.
    ///
    /// `reload` and `refreshAll` are deliberately not gated: both are operator
    /// actions (the composition root's startup load, `POST
    /// /api/blocklists/update`),
    /// and an operator who asks for a refresh on a full disk has asked for it.
    ///
    /// Counting happens here, so a caller cannot skip a pass without recording
    /// it. One `warn` line per skipped pass — at a 24-hour interval that is one
    /// line a day, and the disk monitor already logs the state change itself.
    fn refreshGated(self: *Manager) bool {
        const monitor = self.monitor orelse return false;
        if (monitor.writesAllowed()) return false;
        _ = self.refreshes_gated.fetchAdd(1, .monotonic);
        log.warn("free space is critical; skipping the scheduled blocklist refresh", .{});
        return true;
    }

    /// Scheduled refresh passes the disk gate has skipped.
    pub fn refreshesGated(self: *const Manager) u64 {
        return self.refreshes_gated.load(.monotonic);
    }

    fn startupPass(self: *Manager, io: std.Io) Error!void {
        self.refresh_lock.lockUncancelable(io);
        defer self.refresh_lock.unlock(io);
        // Inside `refresh_lock`, by being registered after the unlock: see
        // `refreshSource`.
        defer self.flushDiagnostics(io);

        // Ahead of the gate on purpose: loading the compiled files that already
        // exist is a read. A full disk must not cost the household its
        // filtering as well as its downloads.
        try self.reloadCollecting(io);

        if (self.refreshGated()) return;

        var rows = try sources_repo.listSourceRows(self.database, self.gpa);
        defer rows.deinit(self.gpa);
        defer sources_repo.freeSourceRows(self.gpa, rows.items);

        const now = std.Io.Clock.real.now(io).toSeconds();
        var refreshed = false;
        for (rows.items) |row| {
            if (!row.enabled) continue;
            if (!self.needsRefresh(io, row, now)) continue;
            if (try self.refreshSourceLocked(io, row)) refreshed = true;
        }
        if (refreshed) try self.reloadCollecting(io);
    }

    fn needsRefresh(self: *Manager, io: std.Io, row: sources_repo.SourceRow, now: i64) bool {
        self.lock.lockSharedUncancelable(io);
        const state: State = state: {
            for (self.statuses) |status| {
                if (status.id == row.id) break :state status.state;
            }
            break :state .never_fetched;
        };
        self.lock.unlockShared(io);

        if (state != .ok) return true;
        const last = row.last_updated orelse return true;
        // The Pi has no RTC, so a fetch stamped while the clock ran ahead of
        // real time (a pre-NTP boot, a restored image) leaves a `last_updated`
        // in the future. Plain interval arithmetic would then suspend every
        // refresh until real time caught up with the poison stamp, and the
        // reconcile engine preserves runtime columns faithfully, so nothing
        // else would ever clear it. A stamp from the future is not evidence of
        // a recent fetch.
        if (last > now) return true;
        return now - last >= model.updateIntervalSeconds(self.schedule(io));
    }

    // -----------------------------------------------------------------------
    // orphans
    // -----------------------------------------------------------------------

    /// Deletes the compiled files and the leftover temporaries whose id is no
    /// longer a `blocklist_sources` row. Files of a live source are left alone,
    /// whatever their state.
    ///
    /// Three callers, and between them they cover every way an orphan is made:
    /// `runScheduler` sweeps once before its startup pass — the residue of a
    /// process that was killed mid-refresh, and of a source deleted while the
    /// server was down — and again before each scheduled pass; the
    /// `DELETE /api/blocklists/{id}` handler sweeps as soon as it has removed
    /// the row, so the directory follows the table an operator can see instead
    /// of waiting out `blocklist_update.interval_hours`.
    ///
    /// It is safe to call on a fresh install: `openDir` creates
    /// `<data_dir>/blocklists/` if nothing has yet, and an empty directory
    /// sweeps to nothing.
    pub fn pruneOrphans(self: *Manager, io: std.Io) Error!void {
        defer self.flushDiagnostics(io);
        return self.pruneOrphansLocked(io);
    }

    /// Assumes nothing and takes both writer mutexes itself. Split from
    /// `pruneOrphans` so the diagnostics flush above happens with neither held.
    fn pruneOrphansLocked(self: *Manager, io: std.Io) Error!void {
        // `refresh_lock` first, and for the reason it exists: the download and
        // the compile are the only writers of `.raw.tmp`, `.list.tmp`,
        // `.wild.tmp` and `.allow.tmp`, and they hold it for as long as they
        // run. Without it here, a source deleted through the API would sweep
        // the temporaries of a refresh still writing them — the row is gone, so
        // nothing else in this function would spare them — and the pass would
        // fail on a raw file that vanished under it.
        //
        // `writer_lock` second, in the one order this file ever takes them,
        // because the rows this reads and the compiled files it deletes are
        // what a reload is reading.
        self.refresh_lock.lockUncancelable(io);
        defer self.refresh_lock.unlock(io);
        self.writer_lock.lockUncancelable(io);
        defer self.writer_lock.unlock(io);

        var rows = try sources_repo.listSourceRows(self.database, self.gpa);
        defer rows.deinit(self.gpa);
        defer sources_repo.freeSourceRows(self.gpa, rows.items);

        var dir = try self.openDir(io, .{ .iterate = true });
        defer dir.close(io);

        // The names are collected first: `Entry.name` is invalidated by the
        // next `next`, and deleting under an open cursor is not defined.
        var doomed: std.ArrayList([]u8) = .empty;
        defer {
            for (doomed.items) |item| self.gpa.free(item);
            doomed.deinit(self.gpa);
        }

        var it = dir.iterate();
        while (true) {
            const entry = it.next(io) catch |err| switch (err) {
                error.Canceled => return error.Canceled,
                else => {
                    log.warn("pruning blocklists: reading the directory failed: {s}", .{@errorName(err)});
                    self.noteStorageFailure(io, .directory_read, "reading the blocklist directory failed: {s}", .{@errorName(err)});
                    return error.FileSystem;
                },
            } orelse break;
            if (entry.kind != .file) continue;
            const id = sourceFileId(entry.name) orelse continue;
            if (containsId(rows.items, id)) continue;
            try doomed.append(self.gpa, try self.gpa.dupe(u8, entry.name));
        }

        self.noteStorageSuccess(io, .directory_read);

        for (doomed.items) |name| {
            self.deleteQuietly(io, dir, name);
            log.info("pruned orphaned blocklist file {s}", .{name});
        }
    }

    // -----------------------------------------------------------------------
    // internals
    // -----------------------------------------------------------------------

    /// Builds the status table for `rows`, carrying every existing entry over
    /// by row id so a recorded failure survives. Nothing is published: the
    /// caller either installs the result or frees it, which is what lets
    /// `reloadLocked` decide only once its snapshot exists.
    fn buildStatusTable(
        self: *Manager,
        io: std.Io,
        rows: []const sources_repo.SourceRow,
    ) Error!StatusTable {
        var fresh: std.heap.ArenaAllocator = .init(self.gpa);
        errdefer fresh.deinit();
        const arena = fresh.allocator();

        // The published table is copied under the lock. Reading it unlocked
        // would race the writer that replaces it — and the arena its entries
        // live in is freed by whoever installs what this builds.
        const previous = previous: {
            self.lock.lockSharedUncancelable(io);
            defer self.lock.unlockShared(io);
            break :previous try arena.dupe(SourceStatus, self.statuses);
        };

        const table = try arena.alloc(SourceStatus, rows.len);
        mergeStatuses(table, rows, previous);
        return .{ .arena = fresh, .items = table };
    }

    /// Publishes a built table and frees the one it replaces. The caller holds
    /// the exclusive lock, so no reader is inside the old table and no flush is
    /// half way through draining it.
    ///
    /// The pass accounting the live table still holds is folded into the
    /// incoming entry of the same id first. `mergeStatuses` left the candidate
    /// carrying none, so an outcome recorded after the candidate was built —
    /// and any a flush has not drained yet — survives the swap exactly once. An
    /// outcome a flush already reported is zero in the live table, so nothing
    /// here resurrects it.
    fn installStatuses(self: *Manager, table: StatusTable) void {
        for (table.items) |*incoming| {
            const live = entryFor(self.statuses, incoming.id) orelse continue;
            incoming.pass_failures +|= live.pass_failures;
            incoming.pass_outcome = incoming.pass_outcome or live.pass_outcome;
        }
        self.status_arena.deinit();
        self.status_arena = table.arena;
        self.statuses = table.items;
    }

    /// Rebuilds and publishes the status table from the current source set.
    ///
    /// `refreshAll` calls this before it refreshes anything: a source added
    /// since the last reload needs an entry to record its outcome in, and the
    /// API has to see the pass advance while it runs. `reloadLocked` does not
    /// call it — a reload publishes its table together with the snapshot that
    /// table describes.
    fn syncStatuses(self: *Manager, io: std.Io, rows: []const sources_repo.SourceRow) Error!void {
        const table = try self.buildStatusTable(io, rows);
        self.lock.lockUncancelable(io);
        defer self.lock.unlock(io);
        self.installStatuses(table);
    }

    /// The recorded entry for one source, or `null` when the table has none.
    fn priorStatus(self: *Manager, io: std.Io, id: i64) ?SourceStatus {
        self.lock.lockSharedUncancelable(io);
        defer self.lock.unlockShared(io);
        for (self.statuses) |entry| {
            if (entry.id == id) return entry;
        }
        return null;
    }

    fn commitStatus(self: *Manager, io: std.Io, status: SourceStatus) void {
        self.lock.lockUncancelable(io);
        defer self.lock.unlock(io);
        for (self.statuses) |*entry| {
            if (entry.id != status.id) continue;
            // `loaded` is the reload's fact, not the refresh's: the files this
            // refresh wrote are not in a snapshot until the next reload reads
            // them.
            const loaded = entry.loaded;
            entry.* = status;
            entry.loaded = loaded;
            return;
        }
        // No entry of that id: the source was added or deleted between the
        // table this pass started from and this commit. The outcome is lost
        // either way — the next reload rebuilds the table from the rows — but
        // a failure that disappears without a line is the one thing milestone
        // 5 says never happens.
        log.warn("blocklist source {d}: no status entry to record the refresh outcome in", .{status.id});
    }

    /// One id per group, in `listGroups` order.
    fn groupIds(self: *Manager, groups: []const model.Group) Error![]i64 {
        const out = try self.gpa.alloc(i64, groups.len);
        errdefer self.gpa.free(out);
        for (out, groups) |*slot, group| {
            slot.* = try groups_repo.groupId(self.database, group.name) orelse
                return error.GroupSetChanged;
        }
        return out;
    }

    fn openDir(self: *Manager, io: std.Io, options: std.Io.Dir.OpenOptions) Error!std.Io.Dir {
        _ = self.paths.dir.createDirPathStatus(io, self.paths.subdir, .fromMode(0o700)) catch |err| switch (err) {
            error.Canceled => return error.Canceled,
            else => {
                log.warn("creating {s} failed: {s}", .{ self.paths.subdir, @errorName(err) });
                self.noteStorageFailure(io, .create_dir, "creating {s} failed: {s}", .{ self.paths.subdir, @errorName(err) });
                return error.FileSystem;
            },
        };
        self.noteStorageSuccess(io, .create_dir);

        const dir = self.paths.dir.openDir(io, self.paths.subdir, options) catch |err| switch (err) {
            error.Canceled => return error.Canceled,
            else => {
                log.warn("opening {s} failed: {s}", .{ self.paths.subdir, @errorName(err) });
                self.noteStorageFailure(io, .open_dir, "opening {s} failed: {s}", .{ self.paths.subdir, @errorName(err) });
                return error.FileSystem;
            },
        };
        self.noteStorageSuccess(io, .open_dir);
        return dir;
    }

    /// A temporary that cannot be removed is not a failure of the operation
    /// that made it, but it is not nothing either: it is left visible.
    fn deleteQuietly(self: *Manager, io: std.Io, dir: std.Io.Dir, name: []const u8) void {
        if (dir.deleteFile(io, name)) {
            self.noteStorageSuccess(io, .delete);
        } else |err| switch (err) {
            // A name that was never created is the ordinary case: the temporary
            // deletes are installed before the files exist.
            error.FileNotFound => self.noteStorageSuccess(io, .delete),
            else => {
                log.warn("deleting {s} failed: {s}", .{ name, @errorName(err) });
                self.noteStorageFailure(io, .delete, "deleting {s} failed: {s}", .{ name, @errorName(err) });
            },
        }
    }
};

/// The one subject `blocklist.snapshot` ever has: a box publishes exactly one
/// snapshot, and every source that failed to load is its own
/// `blocklist.refresh` episode.
const snapshot_key = "snapshot";

/// A status table and the arena holding it. Until `installStatuses` takes it,
/// it is a candidate nobody can see, and `deinit` frees it whole.
const StatusTable = struct {
    arena: std.heap.ArenaAllocator,
    items: []SourceStatus,

    fn deinit(self: *StatusTable) void {
        self.arena.deinit();
        self.items = &.{};
    }
};

/// Fills `table` with one entry per row, carrying an entry of the same row id
/// over from `previous`. A source deleted since `previous` was built is gone; a
/// source added since starts blank. `previous` is only read, so the caller's
/// published table is untouched by this.
///
/// The pass accounting is *not* carried: it lives in exactly one table copy at
/// a time. `previous` is a snapshot of the published table taken outside the
/// swap, so copying its counters here would leave the same outcomes in two
/// tables — the live one for a flush to drain, and this candidate for the
/// reload's own flush to report a second time. A candidate holds only what
/// `applyLoadOutcomes` writes into it; what the live table holds is folded in
/// by `installStatuses` under the exclusive lock.
fn mergeStatuses(
    table: []SourceStatus,
    rows: []const sources_repo.SourceRow,
    previous: []const SourceStatus,
) void {
    for (table, rows) |*status, row| {
        status.* = .{ .id = row.id };
        for (previous) |prior| {
            if (prior.id != row.id) continue;
            status.* = prior;
            break;
        }
        status.pass_outcome = false;
        status.pass_failures = 0;
        // After the carry-over: a url edited on the row wins over the one the
        // prior entry recorded.
        status.setUrl(row.url);
    }
}

/// What one `reload` found for one source. The texts are static, so an outcome
/// borrows nothing and stays valid until the swap that commits it.
const LoadOutcome = union(enum) {
    /// The source is switched off. Not in the snapshot, whatever it was before.
    disabled,
    /// Its compiled files are in the snapshot.
    loaded: matcher.Snapshot.Compiled,
    /// It is not in the snapshot, for this reason.
    failed: struct { state: State, text: []const u8 },
};

fn loadFailure(row: sources_repo.SourceRow, file_name: []const u8, err: anyerror) LoadOutcome {
    log.warn("blocklist {f}: reading {s} failed: {s}", .{ SourceLabel.of(row), file_name, @errorName(err) });
    return .{ .failed = .{ .state = .load_failed, .text = @errorName(err) } };
}

/// Writes one reload's findings into the status table. The caller holds the
/// exclusive lock: this runs inside the swap so the table and the published
/// snapshot describe the same thing.
///
/// `rows` and `outcomes` are parallel. A source with no entry is one the table
/// was not synced for, which cannot happen from `reloadLocked` and is skipped
/// rather than asserted, because the table is rebuilt by row id.
fn applyLoadOutcomes(
    statuses: []SourceStatus,
    rows: []const sources_repo.SourceRow,
    outcomes: []const LoadOutcome,
) void {
    for (rows, outcomes) |row, outcome| {
        const entry = entryFor(statuses, row.id) orelse continue;
        switch (outcome) {
            // A source disabled since it last loaded is no longer filtering,
            // and its recorded state describes files nothing reads.
            .disabled => entry.loaded = false,
            .loaded => {
                entry.loaded = true;
                // Two states survive a successful load. `.ok`, because a
                // refresh in this process already filled the counters the
                // compile produced and the five database columns are a subset
                // of them. And any refresh failure, because the files that just
                // loaded are exactly the ones the failed refresh could not
                // replace, so the operator must still see why.
                if (entry.state == .ok or entry.state.isRefreshFailure()) continue;
                entry.succeed(row.last_updated orelse 0, .{
                    .domains = countOf(row.domain_count),
                    .wildcards = countOf(row.wildcard_count),
                    .exceptions = countOf(row.exception_count),
                    .skipped_regex = countOf(row.skipped_regex_count),
                    .skipped_unsupported = countOf(row.skipped_unsupported_count),
                });
            },
            .failed => |reason| {
                entry.loaded = false;
                // The state follows only when nothing more informative is
                // there: a refresh failure already says why the files are
                // missing or stale, and `loaded` already says they are not
                // filtering.
                if (entry.state.isRefreshFailure()) continue;
                entry.fail(reason.state, reason.text);
            },
        }
    }
}

fn entryFor(statuses: []SourceStatus, id: i64) ?*SourceStatus {
    for (statuses) |*entry| {
        if (entry.id == id) return entry;
    }
    return null;
}

const Outcome = union(enum) {
    fetch: fetcher.Error!fetcher.Result,
    expiry: std.Io.Cancelable!void,
};

fn expire(io: std.Io, duration: std.Io.Clock.Duration) std.Io.Cancelable!void {
    return duration.sleep(io);
}

/// A counter column read back from the database. It is `NOT NULL DEFAULT 0` and
/// only this file writes it, so a value outside `u32` means the row was edited
/// behind nxdns's back; the status reports 0 rather than trapping.
fn countOf(value: i64) u32 {
    return std.math.cast(u32, value) orelse 0;
}

fn destroySnapshot(gpa: Allocator, snapshot: *matcher.Snapshot) void {
    snapshot.deinit();
    gpa.destroy(snapshot);
}

/// Copies the lines `parsers.detectFormat` would count — neither blank nor a
/// comment — until it has `parsers.sample_lines` of them, and writes them to
/// `w` newline-separated. Sampling by line rather than by a byte window is what
/// keeps a run of long comment lines from deciding the format: `detectFormat`
/// reads exactly these lines and ignores everything this drops.
///
/// A line over `compiler.max_line_len` is skipped, as the compiler skips it.
fn collectSample(r: *std.Io.Reader, w: *std.Io.Writer) error{ ReadFailed, WriteFailed }!void {
    var considered: usize = 0;
    while (considered < parsers.sample_lines) {
        const event = (try parsers.nextBoundedLine(r, compiler.max_line_len)) orelse return;
        const raw = switch (event) {
            .long_line => continue,
            .line => |line| line,
        };

        const line = std.mem.trim(u8, raw, &std.ascii.whitespace);
        if (line.len == 0) continue;
        if (parsers.isComment(line)) continue;

        considered += 1;
        try w.writeAll(line);
        try w.writeByte('\n');
    }
}

/// Whether three compiled files carry the bodies `expected` was taken over.
fn compiledBodiesMatch(
    list_bytes: []const u8,
    wild_bytes: []const u8,
    allow_bytes: []const u8,
    expected: []const u8,
) bool {
    return std.mem.eql(u8, expected, &bodyChecksum(
        stripHeader(list_bytes),
        stripHeader(wild_bytes),
        stripHeader(allow_bytes),
    ));
}

/// A compile that produced no entry at all while rejecting lines is an error
/// page, a compressed body or a format the sniff got wrong — not a blocklist.
/// Publishing it would replace a working list with nothing and report `ok`. An
/// input that rejected nothing is an empty list, which is legal.
///
/// A list of nothing but exceptions is loadable: an allow-only list published
/// beside a blocking one is a shape operators use, and it produces entries.
fn rejectedWithoutEntries(counts: compiler.Counts) bool {
    if (counts.domains != 0 or counts.wildcards != 0 or counts.exceptions != 0) return false;
    return counts.invalid != 0 or counts.skipped_unsupported != 0 or counts.long_lines != 0;
}

/// The digest the `.list`, `.wild` and `.allow` bodies share, in that order,
/// each followed by `compiler.body_separator`.
///
/// Must stay byte-for-byte what `compiler.compile` produces, separators
/// included: this is the other half of the same digest, and the two are
/// compared against each other on every refresh.
fn bodyChecksum(list_body: []const u8, wild_body: []const u8, allow_body: []const u8) [64]u8 {
    var hasher = Sha256.init(.{});
    hasher.update(list_body);
    hasher.update(compiler.body_separator);
    hasher.update(wild_body);
    hasher.update(compiler.body_separator);
    hasher.update(allow_body);
    hasher.update(compiler.body_separator);
    var digest: [Sha256.digest_length]u8 = undefined;
    hasher.final(&digest);
    return std.fmt.bytesToHex(digest, .lower);
}

fn compiledName(buf: *[name_buf_len]u8, id: i64, suffix: []const u8) []const u8 {
    // An `i64` prints in at most 20 characters and the longest suffix is ten,
    // so `name_buf_len` cannot be exceeded.
    return std.fmt.bufPrint(buf, "{d}{s}", .{ id, suffix }) catch unreachable;
}

/// Every name `compiledName` can produce, longest suffix first so `.list.tmp`
/// is never read as `.list`.
const source_file_suffixes = [_][]const u8{
    ".allow.tmp", ".list.tmp", ".wild.tmp", ".raw.tmp", ".allow", ".list", ".wild",
};

/// The source id a file under the blocklist directory belongs to, or null when
/// the name is not one of ours.
///
/// The four temporaries count. A refresh that dies between writing one and
/// renaming it leaves a file no later refresh reuses and no `defer` reaches, so
/// excluding them from the sweep means nothing ever removes them. Matching them
/// is safe because `pruneOrphans` holds `refresh_lock` for its whole body:
/// every path that creates a temporary runs under that same lock, so no refresh
/// is in flight while the sweep reads the directory, and a temporary the sweep
/// sees belonging to a source that still has a row is kept regardless.
fn sourceFileId(file_name: []const u8) ?i64 {
    for (source_file_suffixes) |suffix| {
        if (!std.mem.endsWith(u8, file_name, suffix)) continue;
        const stem = file_name[0 .. file_name.len - suffix.len];
        return std.fmt.parseInt(i64, stem, 10) catch null;
    }
    return null;
}

fn containsId(rows: []const sources_repo.SourceRow, id: i64) bool {
    for (rows) |row| {
        if (row.id == id) return true;
    }
    return false;
}

// ---------------------------------------------------------------------------
// tests
// ---------------------------------------------------------------------------
//
// Everything here runs against a `:memory:` database. Only the cancellation
// cases below reach a file, and they reach a `testing.tmpDir` — real HTTP and
// real swaps under load are the integration suite's (S9).

const testing = std.testing;
const events_fixture = @import("../storage/events_fixture.zig");
const migrations = @import("../storage/migrations.zig");

fn openMigrated() !db.Db {
    var database = try db.Db.open(":memory:", .{ .mode = .memory });
    errdefer database.close();
    try db.applyPragmas(&database, .{});
    _ = try migrations.migrate(&database);
    return database;
}

fn testManager(database: *db.Db, fetcher_ptr: *fetcher.Fetcher) !Manager {
    return Manager.init(
        testing.allocator,
        database,
        // `acquire` answers before any directory is touched and the header
        // helpers are pure, so this directory is never opened unless a test
        // replaces it with one of its own.
        .{ .dir = std.Io.Dir.cwd() },
        fetcher_ptr,
        .{},
        .{ .raw = .fromSeconds(30), .clock = .awake },
    );
}

test "init leaves the manager with no snapshot and no statuses" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

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

    var f: fetcher.Fetcher = undefined;
    var manager = try testManager(&database, &f);
    defer manager.deinit(io);

    try testing.expectEqual(@as(u64, 0), manager.generation);
    try testing.expectEqual(@as(usize, 0), manager.statuses.len);
    try testing.expect(manager.current == null);
}

test "acquire before any reload returns null and holds no lock" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

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

    var f: fetcher.Fetcher = undefined;
    var manager = try testManager(&database, &f);
    defer manager.deinit(io);

    try testing.expect(manager.acquire(io) == null);
    // A retained shared lock would make this exclusive lock block forever.
    try testing.expect(manager.lock.tryLock(io));
    manager.lock.unlock(io);
}

test "needsRefresh treats a last_updated in the future as due" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

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

    var f: fetcher.Fetcher = undefined;
    var manager = try testManager(&database, &f);
    defer manager.deinit(io);

    // `.ok` is the only state that consults the clock at all; every other one
    // is already due, so the arithmetic below would be unreachable without it.
    var statuses = [_]SourceStatus{.{ .id = 1, .state = .ok }};
    manager.statuses = &statuses;
    defer manager.statuses = &.{};

    const row = testRow(1, true);
    const stamp = row.last_updated.?;
    const interval = model.updateIntervalSeconds(manager.update);

    // The ordinary cases still hold: fresh is not due, stale is.
    try testing.expect(!manager.needsRefresh(io, row, stamp + 1));
    try testing.expect(manager.needsRefresh(io, row, stamp + interval));

    // The Pi has no RTC. A fetch stamped while the clock ran ahead of real
    // time leaves `now - last` negative, which reads as "fetched moments ago"
    // and suspends every refresh until real time catches the poison stamp —
    // for a whole day here, and for as long as the clock was wrong in general.
    try testing.expect(manager.needsRefresh(io, row, stamp - 1));
    try testing.expect(manager.needsRefresh(io, row, stamp - 86_400));
}

test "the disk gate skips a scheduled refresh only while writes are critical" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

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

    var f: fetcher.Fetcher = undefined;
    var manager = try testManager(&database, &f);
    defer manager.deinit(io);

    // No monitor: every pass runs, which is what the tests and `check` rely on.
    try testing.expect(!manager.refreshGated());
    try testing.expectEqual(@as(u64, 0), manager.refreshesGated());

    var monitor: disk_monitor.Monitor = .init(.{}, std.Io.Dir.cwd(), ".", null);
    manager.monitor = &monitor;

    // `.ok` and `.warn` both allow writes: only `critical` stops them.
    try testing.expect(!manager.refreshGated());
    monitor.state_raw.store(@intFromEnum(disk_monitor.State.warn), .monotonic);
    try testing.expect(!manager.refreshGated());
    try testing.expectEqual(@as(u64, 0), manager.refreshesGated());

    monitor.state_raw.store(@intFromEnum(disk_monitor.State.critical), .monotonic);
    try testing.expect(manager.refreshGated());
    try testing.expect(manager.refreshGated());
    try testing.expectEqual(@as(u64, 2), manager.refreshesGated());

    // Free space recovers and the schedule resumes; the counter keeps its total.
    monitor.state_raw.store(@intFromEnum(disk_monitor.State.ok), .monotonic);
    try testing.expect(!manager.refreshGated());
    try testing.expectEqual(@as(u64, 2), manager.refreshesGated());
}

test "statusSnapshot on an empty manager copies nothing" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

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

    var f: fetcher.Fetcher = undefined;
    var manager = try testManager(&database, &f);
    defer manager.deinit(io);

    var out: [4]SourceStatus = undefined;
    try testing.expectEqual(@as(usize, 0), manager.statusSnapshot(io, &out));
}

// ---------------------------------------------------------------------------
// a canceled read of a compiled file
// ---------------------------------------------------------------------------

/// The one file open a `cancelingIo` turns into `error.Canceled`, and the io it
/// hands every other open to.
///
/// A `std.Io` carries its implementation's `userdata`, so a patched vtable entry
/// cannot smuggle a receiver of its own through it and has to read its
/// configuration from here. The test runner runs the tests of one binary in
/// sequence, so one instance is enough.
var canceling_read: struct {
    inner: std.Io = undefined,
    /// The file-name suffix whose open is canceled.
    suffix: []const u8 = "",
} = .{};

/// `inner` with the open of every file whose name ends in `suffix` replaced by
/// `error.Canceled`. `vtable` is the caller's storage for the patched copy and
/// must outlive the returned io.
fn cancelingIo(inner: std.Io, suffix: []const u8, vtable: *std.Io.VTable) std.Io {
    canceling_read = .{ .inner = inner, .suffix = suffix };
    vtable.* = inner.vtable.*;
    vtable.dirOpenFile = cancelingOpenFile;
    return .{ .userdata = inner.userdata, .vtable = vtable };
}

fn cancelingOpenFile(
    userdata: ?*anyopaque,
    dir: std.Io.Dir,
    sub_path: []const u8,
    options: std.Io.Dir.OpenFileOptions,
) std.Io.File.OpenError!std.Io.File {
    if (std.mem.endsWith(u8, sub_path, canceling_read.suffix)) return error.Canceled;
    return canceling_read.inner.vtable.dirOpenFile(userdata, dir, sub_path, options);
}

test "a canceled compiled-file read cancels the reload instead of recording it" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();

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

    var f: fetcher.Fetcher = undefined;
    var mgr = try testManager(&database, &f);
    defer mgr.deinit(io);
    mgr.paths = .{ .dir = tmp.dir };

    const url = "https://lists.example/hosts.txt";
    try sources_repo.insertBlocklistSource(&database, .{ .url = url, .name = "example" }, .{});
    var rows = try sources_repo.listSourceRows(&database, testing.allocator);
    defer rows.deinit(testing.allocator);
    defer sources_repo.freeSourceRows(testing.allocator, rows.items);
    const id = rows.items[0].id;

    const list_body = "aaa.example.com\n";
    const wild_body = "";
    const allow_body = "";
    var dir = try tmp.dir.createDirPathOpen(io, "blocklists", .{});
    defer dir.close(io);
    var list_buf: [name_buf_len]u8 = undefined;
    var wild_buf: [name_buf_len]u8 = undefined;
    var allow_buf: [name_buf_len]u8 = undefined;
    try dir.writeFile(io, .{ .sub_path = compiledName(&list_buf, id, ".list"), .data = list_body });
    try dir.writeFile(io, .{ .sub_path = compiledName(&wild_buf, id, ".wild"), .data = wild_body });
    // Present rather than absent, so the third read is a real one: `loadSource`
    // treats a missing `.allow` as an empty body and would never open it.
    try dir.writeFile(io, .{ .sub_path = compiledName(&allow_buf, id, ".allow"), .data = allow_body });
    try sources_repo.updateSourceStats(&database, id, .{
        .last_updated = 1_700_000_000,
        .domain_count = 1,
        .wildcard_count = 0,
        .skipped_regex_count = 0,
        .skipped_unsupported_count = 0,
        .exception_count = 0,
        .checksum = &bodyChecksum(list_body, wild_body, allow_body),
    });

    // The baseline every assertion below is against: one clean reload, one
    // status entry that says so.
    try mgr.reload(io);
    var out: [4]SourceStatus = undefined;
    try testing.expectEqual(@as(usize, 1), mgr.statusSnapshot(io, &out));
    try testing.expectEqual(State.ok, out[0].state);
    try testing.expect(out[0].loaded);
    const published = mgr.generation;

    // Every catch site, in the order `loadSource` reads the three files. A
    // cancellation is consumed by whoever catches it, so folding it into a load
    // failure would spend the shutdown signal and leave a status row reading
    // "Canceled" behind. The `.allow` read is the one that can get this wrong
    // twice over: it also has to keep `FileNotFound` apart from a cancellation.
    for ([_][]const u8{ ".list", ".wild", ".allow" }) |suffix| {
        var vtable: std.Io.VTable = undefined;
        const canceling = cancelingIo(io, suffix, &vtable);
        try testing.expectError(error.Canceled, mgr.reload(canceling));

        try testing.expectEqual(@as(usize, 1), mgr.statusSnapshot(io, &out));
        try testing.expectEqual(State.ok, out[0].state);
        try testing.expectEqual(@as(usize, 0), out[0].errorText().len);
        try testing.expect(out[0].loaded);
        // Nothing was published either: the snapshot the reload never built
        // cannot have replaced the one still serving.
        try testing.expectEqual(published, mgr.generation);
    }
}

test "committing a status for an id the table has no entry for is not silent" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

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

    var f: fetcher.Fetcher = undefined;
    var mgr = try testManager(&database, &f);
    defer mgr.deinit(io);

    const rows = [_]sources_repo.SourceRow{testRow(1, true)};
    try mgr.syncStatuses(io, &rows);

    // The unknown id: a source inserted through the API after this pass built
    // its table, or deleted before the pass reached its commit. The outcome
    // has nowhere to go, and the warning is the only trace it leaves. The log
    // sink cannot be installed under the test runner — it would eat the
    // harness's own output — so what is asserted here is that the miss is
    // survivable and changes nothing.
    var stranger: SourceStatus = .{ .id = 42 };
    stranger.fail(.fetch_failed, "Timeout");
    mgr.commitStatus(io, stranger);

    var out: [4]SourceStatus = undefined;
    try testing.expectEqual(@as(usize, 1), mgr.statusSnapshot(io, &out));
    try testing.expectEqual(@as(i64, 1), out[0].id);
    try testing.expectEqual(State.never_fetched, out[0].state);

    // The same commit against an id the table does know still lands.
    var known: SourceStatus = .{ .id = 1 };
    known.fail(.fetch_failed, "Timeout");
    mgr.commitStatus(io, known);
    _ = mgr.statusSnapshot(io, &out);
    try testing.expectEqual(State.fetch_failed, out[0].state);
    try testing.expectEqualStrings("Timeout", out[0].errorText());
}

test "the header writer produces the documented text" {
    var buf: [512]u8 = undefined;
    var w: std.Io.Writer = .fixed(&buf);
    const header: Header = .{
        .url = "https://lists.example/hosts.txt",
        .format = .hosts,
        .fetched_at = 1_700_000_000,
        .counts = .{
            .domains = 12,
            .wildcards = 3,
            .exceptions = 7,
            .skipped_regex = 2,
            .skipped_unsupported = 1,
            .invalid = 5,
            .long_lines = 9,
            .duplicates = 4,
        },
        .checksum = "0" ** 64,
    };
    try header.write(&w);

    try testing.expectEqualStrings(
        \\# nxdns blocklist
        \\# url https://lists.example/hosts.txt
        \\# format hosts
        \\# fetched_at 1700000000
        \\# domains 12
        \\# wildcards 3
        \\# exceptions 7
        \\# skipped_regex 2
        \\# skipped_unsupported 1
        \\# invalid 5
        \\
    ++ "# sha256 " ++ "0" ** 64 ++ "\n", w.buffered());
}

test "the log label names a source without printing what its url carries" {
    // Every `log.warn` in this file formats its subject through `SourceLabel`,
    // so this is the text of those lines. A `std.log` line is not observable
    // from a unit test under the default runner; the label is.
    var buf: [1024]u8 = undefined;

    const row: sources_repo.SourceRow = .{
        .id = 3,
        .url = "https://lists.example/download/token/hunter2/hosts.txt?apikey=s3cr3t",
        .name = "ads",
        .enabled = true,
        .last_updated = null,
        .domain_count = 0,
        .wildcard_count = 0,
        .skipped_regex_count = 0,
        .skipped_unsupported_count = 0,
        .checksum = null,
    };
    const printed = try std.fmt.bufPrint(&buf, "blocklist {f}: download failed: {s}", .{
        SourceLabel.of(row),
        @errorName(error.ConnectFailed),
    });
    try testing.expectEqualStrings(
        "blocklist source 3 'ads' 'https://lists.example': download failed: ConnectFailed",
        printed,
    );
    try testing.expect(!std.mem.containsAtLeast(u8, printed, 1, "hunter2"));
    try testing.expect(!std.mem.containsAtLeast(u8, printed, 1, "s3cr3t"));

    // The row is database text, and a path that writes it does not have to
    // validate as strictly as the config validator. Neither column may end the
    // line and start one of the operator's choosing.
    var forged = row;
    forged.name = "ads\n2026-01-01 ERROR forged";
    forged.url = "https://lists.example\n2026-01-01 ERROR forged/hosts.txt";
    const escaped = try std.fmt.bufPrint(&buf, "blocklist {f}", .{SourceLabel.of(forged)});
    try testing.expectEqualStrings(
        "blocklist source 3 'ads\\n2026-01-01 ERROR forged'" ++
            " 'https://lists.example\\n2026-01-01 ERROR forged'",
        escaped,
    );
    try testing.expect(!std.mem.containsAtLeast(u8, escaped, 1, "\n"));

    // A name is operator-supplied and reaches the row through the API, so it
    // can close the quote this label puts around it and open a decoy that reads
    // as the url of a second source. The quote it would close is escaped, and
    // the escape is unambiguous because a `\` is escaped too.
    var decoy = row;
    decoy.name = "ads' (https://decoy.example) --";
    decoy.url = "https://lists.example/hosts.txt";
    const quoted = try std.fmt.bufPrint(&buf, "blocklist {f}", .{SourceLabel.of(decoy)});
    try testing.expectEqualStrings(
        "blocklist source 3 'ads\\' (https://decoy.example) --' 'https://lists.example'",
        quoted,
    );
}

test "stripHeader returns the body of a compiled file" {
    const file =
        "# nxdns blocklist\n" ++
        "# url https://lists.example/hosts.txt\n" ++
        "ads.example.com\ntracker.example.net\n";
    try testing.expectEqualStrings("ads.example.com\ntracker.example.net\n", stripHeader(file));
}

test "stripHeader returns everything for a file with no header" {
    try testing.expectEqualStrings("a.example.com\n", stripHeader("a.example.com\n"));
}

test "stripHeader returns an empty body for a header-only file" {
    try testing.expectEqualStrings("", stripHeader("# nxdns blocklist\n# sha256 x\n"));
}

test "stripHeader tolerates an unterminated header line" {
    try testing.expectEqualStrings("", stripHeader("# nxdns blocklist"));
}

test "SourceStatus truncates a long error at max_error_len" {
    var status: SourceStatus = .{ .id = 1 };
    const long = "E" ** (max_error_len + 40);
    status.fail(.fetch_failed, long);

    try testing.expectEqual(State.fetch_failed, status.state);
    try testing.expectEqual(@as(u8, max_error_len), status.last_error_len);
    try testing.expectEqualStrings("E" ** max_error_len, status.errorText());
}

test "a success clears the recorded error" {
    var status: SourceStatus = .{ .id = 7 };
    status.fail(.compile_failed, "TooManyDomains");
    status.succeed(1_700_000_000, .{ .domains = 3, .wildcards = 1 });

    try testing.expectEqual(State.ok, status.state);
    try testing.expectEqual(@as(i64, 1_700_000_000), status.last_success);
    try testing.expectEqual(@as(u32, 3), status.counts.domains);
    try testing.expectEqualStrings("", status.errorText());
}

comptime {
    // The two tests below spell every suffix out instead of looping over
    // `source_file_suffixes`: a test that reads the table moves with it, so a
    // name dropped from the table would take the assertion that covers it along.
    // An eighth suffix breaks the build here until both are extended.
    std.debug.assert(source_file_suffixes.len == 7);
}

test "compiledName spells every file name of a source" {
    var buf: [name_buf_len]u8 = undefined;
    try testing.expectEqualStrings("42.list", compiledName(&buf, 42, ".list"));
    try testing.expectEqualStrings("42.wild", compiledName(&buf, 42, ".wild"));
    try testing.expectEqualStrings("42.allow", compiledName(&buf, 42, ".allow"));
    try testing.expectEqualStrings("42.raw.tmp", compiledName(&buf, 42, ".raw.tmp"));
    try testing.expectEqualStrings("42.list.tmp", compiledName(&buf, 42, ".list.tmp"));
    try testing.expectEqualStrings("42.wild.tmp", compiledName(&buf, 42, ".wild.tmp"));
    try testing.expectEqualStrings("42.allow.tmp", compiledName(&buf, 42, ".allow.tmp"));
}

test "sourceFileId matches every name a refresh writes, including the temporaries" {
    try testing.expectEqual(@as(?i64, 7), sourceFileId("7.list"));
    try testing.expectEqual(@as(?i64, 7), sourceFileId("7.wild"));
    try testing.expectEqual(@as(?i64, 7), sourceFileId("7.allow"));

    // A temporary left by a refresh that died belongs to its source id, so the
    // sweep can tell whether that source still has a row.
    try testing.expectEqual(@as(?i64, 7), sourceFileId("7.raw.tmp"));
    try testing.expectEqual(@as(?i64, 7), sourceFileId("7.list.tmp"));
    try testing.expectEqual(@as(?i64, 7), sourceFileId("7.wild.tmp"));
    try testing.expectEqual(@as(?i64, 7), sourceFileId("7.allow.tmp"));

    try testing.expectEqual(@as(?i64, null), sourceFileId("notes.list"));
    try testing.expectEqual(@as(?i64, null), sourceFileId("notes.allow"));
    try testing.expectEqual(@as(?i64, null), sourceFileId("notes.raw.tmp"));
    try testing.expectEqual(@as(?i64, null), sourceFileId("notes.allow.tmp"));
    try testing.expectEqual(@as(?i64, null), sourceFileId("7.tmp"));
    try testing.expectEqual(@as(?i64, null), sourceFileId("7.raw"));
    try testing.expectEqual(@as(?i64, null), sourceFileId("7.allowed"));
    try testing.expectEqual(@as(?i64, null), sourceFileId("README"));
}

test "every name compiledName writes is a name the sweep can attribute" {
    // A round-trip over the table, not a coverage check: this loop reads the
    // same array the code reads, so it cannot notice a missing entry. The two
    // tests above are what pins the set.
    var buf: [name_buf_len]u8 = undefined;
    for (source_file_suffixes) |suffix| {
        try testing.expectEqual(@as(?i64, 42), sourceFileId(compiledName(&buf, 42, suffix)));
    }
}

test "a failed refresh keeps the fields of the compiled files still serving" {
    var status: SourceStatus = .{ .id = 3 };
    status.succeed(1_700_000_000, .{ .domains = 5, .wildcards = 2 });

    // What `refreshSourceLocked` starts from, and what a download failure does
    // to it.
    var next = status;
    next.last_attempt = 1_700_003_600;
    next.fail(.fetch_failed, "Timeout");

    try testing.expectEqual(State.fetch_failed, next.state);
    try testing.expectEqualStrings("Timeout", next.errorText());
    try testing.expectEqual(@as(i64, 1_700_000_000), next.last_success);
    try testing.expectEqual(@as(i64, 1_700_003_600), next.last_attempt);
    try testing.expectEqual(@as(u32, 5), next.counts.domains);
    try testing.expectEqual(@as(u32, 2), next.counts.wildcards);
}

test "a load outcome never overwrites a refresh failure" {
    try testing.expect(State.fetch_failed.isRefreshFailure());
    try testing.expect(State.compile_failed.isRefreshFailure());
    try testing.expect(State.no_valid_entries.isRefreshFailure());

    // The three a load produces. `applyLoadOutcomes` may write over these,
    // because nothing more informative is there.
    try testing.expect(!State.ok.isRefreshFailure());
    try testing.expect(!State.never_fetched.isRefreshFailure());
    try testing.expect(!State.load_failed.isRefreshFailure());
}

fn testRow(id: i64, enabled: bool) sources_repo.SourceRow {
    return .{
        .id = id,
        .url = "https://lists.example/hosts.txt",
        .name = "example",
        .enabled = enabled,
        .last_updated = 1_700_000_000,
        .domain_count = 9,
        .wildcard_count = 4,
        .exception_count = 2,
        .skipped_regex_count = 1,
        .skipped_unsupported_count = 5,
        .checksum = "0" ** 64,
    };
}

test "a candidate table carries prior entries over and leaves the published one alone" {
    var published = [_]SourceStatus{ .{ .id = 1 }, .{ .id = 2 } };
    published[0].setUrl("https://lists.example/one.txt");
    published[0].fail(.fetch_failed, "HttpStatus");
    published[0].loaded = true;
    published[1].setUrl("https://lists.example/two.txt");
    published[1].succeed(1_700_000_000, .{ .domains = 4 });

    // Source 2 was deleted and source 3 added; source 1 kept its id and got a
    // new url.
    const rows = [_]sources_repo.SourceRow{
        blk: {
            var row = testRow(1, true);
            row.url = "https://lists.example/moved.txt";
            break :blk row;
        },
        testRow(3, true),
    };

    var candidate: [2]SourceStatus = undefined;
    mergeStatuses(&candidate, &rows, &published);

    try testing.expectEqual(@as(i64, 1), candidate[0].id);
    try testing.expectEqual(State.fetch_failed, candidate[0].state);
    try testing.expectEqualStrings("HttpStatus", candidate[0].errorText());
    try testing.expect(candidate[0].loaded);
    try testing.expectEqualStrings("https://lists.example/moved.txt", candidate[0].urlText());

    try testing.expectEqual(@as(i64, 3), candidate[1].id);
    try testing.expectEqual(State.never_fetched, candidate[1].state);
    try testing.expect(!candidate[1].loaded);

    // The one thing a candidate does not carry. `published[0]` is holding a
    // failure no flush has drained yet; copying its accounting here would leave
    // the same outcome in two tables, and the flush of each would report it.
    try testing.expect(published[0].pass_outcome);
    try testing.expect(!candidate[0].pass_outcome);
    try testing.expectEqual(@as(u16, 0), candidate[0].pass_failures);

    // The published table is untouched, so a reload that fails before the swap
    // leaves it describing the snapshot that is still serving — including the
    // entry of the deleted source, which that snapshot still enforces.
    try testing.expectEqual(@as(usize, 2), published.len);
    try testing.expectEqual(@as(i64, 2), published[1].id);
    try testing.expectEqual(State.ok, published[1].state);
    try testing.expectEqualStrings("https://lists.example/one.txt", published[0].urlText());
}

test "installing a table folds the live pass accounting in by id" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

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

    var f: fetcher.Fetcher = undefined;
    var mgr = try testManager(&database, &f);
    defer mgr.deinit(io);

    // Source 1 recorded a failure the flush has not drained; source 2 was
    // drained already; source 3 recorded one this reload knows nothing about.
    var live: std.heap.ArenaAllocator = .init(testing.allocator);
    const live_items = try live.allocator().alloc(SourceStatus, 3);
    live_items[0] = .{ .id = 1, .pass_outcome = true, .pass_failures = 2 };
    live_items[1] = .{ .id = 2 };
    live_items[2] = .{ .id = 3, .pass_outcome = true, .pass_failures = 1 };
    {
        mgr.lock.lockUncancelable(io);
        defer mgr.lock.unlock(io);
        mgr.installStatuses(.{ .arena = live, .items = live_items });
    }

    // What a reload built beside it, carrying only its own load outcomes.
    var incoming: std.heap.ArenaAllocator = .init(testing.allocator);
    const incoming_items = try incoming.allocator().alloc(SourceStatus, 3);
    incoming_items[0] = .{ .id = 1, .pass_outcome = true, .pass_failures = 1 };
    incoming_items[1] = .{ .id = 2, .pass_outcome = true, .pass_failures = 4 };
    incoming_items[2] = .{ .id = 3 };
    {
        mgr.lock.lockUncancelable(io);
        defer mgr.lock.unlock(io);
        mgr.installStatuses(.{ .arena = incoming, .items = incoming_items });
    }

    try testing.expectEqual(@as(u16, 3), mgr.statuses[0].pass_failures);
    try testing.expect(mgr.statuses[0].pass_outcome);

    // A drained entry adds nothing: what the swap publishes is the reload's own
    // accounting and no resurrection of what was already reported.
    try testing.expectEqual(@as(u16, 4), mgr.statuses[1].pass_failures);
    try testing.expect(mgr.statuses[1].pass_outcome);

    // The half the swap used to lose: an outcome the live table held and the
    // candidate never saw.
    try testing.expectEqual(@as(u16, 1), mgr.statuses[2].pass_failures);
    try testing.expect(mgr.statuses[2].pass_outcome);
}

test "the flush claims every entry that carries pass accounting, once" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

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

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

    var f: fetcher.Fetcher = undefined;
    var mgr = try testManager(&database, &f);
    defer mgr.deinit(io);
    mgr.diagnostics = &fx.store;

    var arena: std.heap.ArenaAllocator = .init(testing.allocator);
    const items = try arena.allocator().alloc(SourceStatus, 3);
    items[0] = .{ .id = 1 };
    items[0].setUrl("https://lists.example/one.txt");
    items[0].fail(.fetch_failed, "HttpStatus");
    // No url, so no episode to report under. The drain has to claim it anyway:
    // an entry left with `pass_outcome` set is the one every later scan finds
    // first, and the entry behind it would never be reached.
    items[1] = .{ .id = 2, .pass_outcome = true, .pass_failures = 1 };
    items[2] = .{ .id = 3 };
    items[2].setUrl("https://lists.example/three.txt");
    items[2].succeed(1_700_000_000, .{ .domains = 3 });
    {
        mgr.lock.lockUncancelable(io);
        defer mgr.lock.unlock(io);
        mgr.installStatuses(.{ .arena = arena, .items = items });
    }

    mgr.flushDiagnostics(io);

    try testing.expectEqual(@as(i64, 1), try fx.count(
        "SELECT count(*) FROM operational_events WHERE code = 'blocklist.refresh'",
    ));
    try testing.expectEqual(@as(i64, 1), try fx.count(
        "SELECT occurrences FROM operational_events WHERE code = 'blocklist.refresh'",
    ));
    for (mgr.statuses) |entry| {
        try testing.expect(!entry.pass_outcome);
        try testing.expectEqual(@as(u16, 0), entry.pass_failures);
    }

    // Drained: flushing the same table again reports nothing a second time.
    mgr.flushDiagnostics(io);
    try testing.expectEqual(@as(i64, 1), try fx.count(
        "SELECT occurrences FROM operational_events WHERE code = 'blocklist.refresh'",
    ));
}

test "the flush closes the episode of a source that is no longer in the table" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

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

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

    var f: fetcher.Fetcher = undefined;
    var mgr = try testManager(&database, &f);
    defer mgr.deinit(io);
    mgr.diagnostics = &fx.store;

    // What the operator deleted while it was failing. Nothing will ever record
    // a success for it, so nothing but the sweep can close this.
    fx.store.report(
        io,
        1_700_000_000,
        .blocklist_refresh,
        "https://lists.example/deleted.txt",
        "lists.example/deleted.txt",
        .warning,
        "fetch_failed: HttpStatus (1 this pass)",
    );

    var arena: std.heap.ArenaAllocator = .init(testing.allocator);
    const items = try arena.allocator().alloc(SourceStatus, 1);
    items[0] = .{ .id = 1 };
    items[0].setUrl("https://lists.example/one.txt");
    items[0].fail(.fetch_failed, "HttpStatus");
    {
        mgr.lock.lockUncancelable(io);
        defer mgr.lock.unlock(io);
        mgr.installStatuses(.{ .arena = arena, .items = items });
    }

    // No snapshot published yet, so the table is not known to describe the
    // source set and the sweep must not run: the drain reports the failing
    // source and the deleted one's episode is left alone.
    mgr.flushDiagnostics(io);
    try testing.expectEqual(@as(i64, 2), try fx.count(
        "SELECT count(*) FROM operational_events WHERE code = 'blocklist.refresh' AND resolved_at IS NULL",
    ));

    mgr.generation = 1;
    mgr.flushDiagnostics(io);

    // One left active, and it is the source that still exists.
    try testing.expectEqual(@as(i64, 1), try fx.count(
        "SELECT count(*) FROM operational_events WHERE code = 'blocklist.refresh' AND resolved_at IS NULL",
    ));
    try testing.expectEqualStrings(
        "https://lists.example/one.txt",
        try fx.text(
            "SELECT subject_key FROM operational_events WHERE code = 'blocklist.refresh' AND resolved_at IS NULL",
        ),
    );
    try testing.expectEqualStrings(
        "https://lists.example/deleted.txt",
        try fx.text(
            "SELECT subject_key FROM operational_events WHERE code = 'blocklist.refresh' AND resolved_at IS NOT NULL",
        ),
    );
}

test "a disabled source stops being loaded" {
    var statuses = [_]SourceStatus{.{ .id = 1 }};
    statuses[0].succeed(1_700_000_000, .{ .domains = 9 });
    statuses[0].loaded = true;

    const rows = [_]sources_repo.SourceRow{testRow(1, false)};
    applyLoadOutcomes(&statuses, &rows, &.{.disabled});

    // Nothing enforces it any more, and the state that described the files it
    // used to serve is left as the record of how it last stood.
    try testing.expect(!statuses[0].loaded);
    try testing.expectEqual(State.ok, statuses[0].state);
}

test "a source that failed to refresh keeps its failure while its old files serve" {
    // What `refreshSourceLocked` records, then what the `reload` that follows
    // it in `refreshAll` finds: the previous files still load.
    var statuses = [_]SourceStatus{.{ .id = 1 }};
    statuses[0].succeed(1_700_000_000, .{ .domains = 9 });
    statuses[0].fail(.fetch_failed, "HttpStatus");

    const rows = [_]sources_repo.SourceRow{testRow(1, true)};
    const body: matcher.Snapshot.Compiled = .{ .list_body = "", .wild_body = "" };
    applyLoadOutcomes(&statuses, &rows, &.{.{ .loaded = body }});

    try testing.expect(statuses[0].loaded);
    try testing.expectEqual(State.fetch_failed, statuses[0].state);
    try testing.expectEqualStrings("HttpStatus", statuses[0].errorText());
    try testing.expectEqual(@as(u32, 9), statuses[0].counts.domains);
}

test "a load failure is recorded when no refresh failure explains it" {
    var statuses = [_]SourceStatus{ .{ .id = 1 }, .{ .id = 2 } };
    statuses[0].succeed(1_700_000_000, .{ .domains = 9 });
    statuses[0].loaded = true;
    statuses[1].fail(.compile_failed, "TooManyDomains");
    statuses[1].loaded = true;

    const rows = [_]sources_repo.SourceRow{ testRow(1, true), testRow(2, true) };
    const reason: LoadOutcome = .{ .failed = .{ .state = .load_failed, .text = "ChecksumMismatch" } };
    applyLoadOutcomes(&statuses, &rows, &.{ reason, reason });

    try testing.expect(!statuses[0].loaded);
    try testing.expectEqual(State.load_failed, statuses[0].state);
    try testing.expectEqualStrings("ChecksumMismatch", statuses[0].errorText());

    // The compile failure is why the files are unusable; it outranks the
    // symptom the loader saw.
    try testing.expect(!statuses[1].loaded);
    try testing.expectEqual(State.compile_failed, statuses[1].state);
    try testing.expectEqualStrings("TooManyDomains", statuses[1].errorText());
}

test "a load of a source this process never refreshed takes the row counters" {
    var statuses = [_]SourceStatus{.{ .id = 1 }};
    const rows = [_]sources_repo.SourceRow{testRow(1, true)};
    const body: matcher.Snapshot.Compiled = .{ .list_body = "", .wild_body = "" };
    applyLoadOutcomes(&statuses, &rows, &.{.{ .loaded = body }});

    try testing.expect(statuses[0].loaded);
    try testing.expectEqual(State.ok, statuses[0].state);
    try testing.expectEqual(@as(i64, 1_700_000_000), statuses[0].last_success);
    try testing.expectEqual(@as(u32, 9), statuses[0].counts.domains);
    try testing.expectEqual(@as(u32, 4), statuses[0].counts.wildcards);
    try testing.expectEqual(@as(u32, 1), statuses[0].counts.skipped_regex);
    // Rehydration: a restart reads this from the row and nowhere else, because
    // no path reparses a compiled file's header.
    try testing.expectEqual(@as(u32, 5), statuses[0].counts.skipped_unsupported);
}

test "a status borrows nothing, so a copy outlives the table it came from" {
    var status: SourceStatus = .{ .id = 5 };
    status.setUrl("https://lists.example/hosts.txt");
    status.fail(.load_failed, "ChecksumMismatch");

    const copy = status;
    // The source of the original is overwritten, as a reload overwrites the
    // table: a copy that borrowed would read the new bytes or freed memory.
    status.setUrl("https://other.example/other.txt");
    status.fail(.fetch_failed, "Timeout");

    try testing.expectEqualStrings("https://lists.example/hosts.txt", copy.urlText());
    try testing.expectEqualStrings("ChecksumMismatch", copy.errorText());
    try testing.expectEqual(State.load_failed, copy.state);
}

test "SourceStatus truncates a long url at max_url_len" {
    var status: SourceStatus = .{ .id = 6 };
    status.setUrl("https://lists.example/" ++ "p" ** max_url_len);

    try testing.expectEqual(@as(u8, max_url_len), status.url_len);
    try testing.expectEqualStrings(
        ("https://lists.example/" ++ "p" ** max_url_len)[0..max_url_len],
        status.urlText(),
    );

    // A shorter url must not leave the tail of the longer one behind it.
    status.setUrl("https://a.example/x");
    try testing.expectEqualStrings("https://a.example/x", status.urlText());
}

test "compiledBodiesMatch verifies the bodies, not the presence of the files" {
    const list_body = "a.example.com\nb.example.com\n";
    const wild_body = "c.example.com\n";
    const allow_body = "d.example.com\n";
    const expected = bodyChecksum(list_body, wild_body, allow_body);

    const header =
        "# nxdns blocklist\n" ++
        "# url https://lists.example/hosts.txt\n";
    try testing.expect(compiledBodiesMatch(
        header ++ list_body,
        header ++ wild_body,
        header ++ allow_body,
        &expected,
    ));

    // The corruption a reload reports as `ChecksumMismatch`: the file is there,
    // its body is not what the checksum was taken over. An allow body that lost
    // its entry counts, because a dropped exception silently restores a block.
    try testing.expect(!compiledBodiesMatch(
        header ++ "a.example.com\nb.exa",
        header ++ wild_body,
        header ++ allow_body,
        &expected,
    ));
    try testing.expect(!compiledBodiesMatch(header ++ list_body, header ++ wild_body, "", &expected));
    try testing.expect(!compiledBodiesMatch("", "", "", &expected));
}

test "bodyChecksum separates the three bodies" {
    const list_body = "a.example.com\nb.example.com\n";
    const wild_body = "c.example.com\n";

    var hasher = Sha256.init(.{});
    hasher.update(list_body);
    hasher.update(compiler.body_separator);
    hasher.update(wild_body);
    hasher.update(compiler.body_separator);
    hasher.update(compiler.body_separator);
    var digest: [Sha256.digest_length]u8 = undefined;
    hasher.final(&digest);
    const expected = std.fmt.bytesToHex(digest, .lower);

    try testing.expectEqualStrings(&expected, &bodyChecksum(list_body, wild_body, ""));
    // No `.allow` file: what `loadSource` and `diskBodiesMatch` pass for one.
    // It is an empty body, and an empty body still gets its separator.
    try testing.expect(compiledBodiesMatch(list_body, wild_body, "", &expected));

    // The framing itself: the same bytes in a different body is a different
    // digest. Unframed these two are equal, and a stale `.list` survives an
    // upstream that switched the name to a wildcard.
    try testing.expect(!std.mem.eql(
        u8,
        &bodyChecksum("a.example\n", "", ""),
        &bodyChecksum("", "a.example\n", ""),
    ));
}

test "rejectedWithoutEntries fails a compile that produced nothing usable" {
    // An html error page: every line is rejected, nothing is written.
    try testing.expect(rejectedWithoutEntries(.{ .invalid = 12, .skipped_unsupported = 3 }));
    // A compressed body: one long binary run with no newline in it.
    try testing.expect(rejectedWithoutEntries(.{ .long_lines = 1 }));

    // An empty list rejects nothing and is legal.
    try testing.expect(!rejectedWithoutEntries(.{}));
    // A real list rejects lines and still produces entries.
    try testing.expect(!rejectedWithoutEntries(.{ .domains = 1000, .invalid = 40 }));
    try testing.expect(!rejectedWithoutEntries(.{ .wildcards = 7, .skipped_unsupported = 90 }));
}

fn sampleOf(input: []const u8, out: []u8) ![]const u8 {
    var r: std.Io.Reader = .fixed(input);
    var w: std.Io.Writer = .fixed(out);
    try collectSample(&r, &w);
    return w.buffered();
}

test "collectSample skips comments instead of spending the sample on them" {
    const gpa = testing.allocator;
    const long_comment = "# " ++ "c" ** (compiler.max_line_len - 2) ++ "\n";

    var input: std.ArrayList(u8) = .empty;
    defer input.deinit(gpa);
    // Sixteen of these fill a 64 KiB window on their own.
    for (0..20) |_| try input.appendSlice(gpa, long_comment);
    try input.appendSlice(gpa, "0.0.0.0 ads.example.com\n0.0.0.0 tracker.example.net\n");

    const out = try gpa.alloc(u8, sample_buf_len);
    defer gpa.free(out);
    const sample = try sampleOf(input.items, out);

    try testing.expectEqualStrings(
        "0.0.0.0 ads.example.com\n0.0.0.0 tracker.example.net\n",
        sample,
    );
    try testing.expectEqual(parsers.Format.hosts, parsers.detectFormat(sample));
}

test "collectSample stops at sample_lines counted lines" {
    const gpa = testing.allocator;

    var input: std.ArrayList(u8) = .empty;
    defer input.deinit(gpa);
    var line_buf: [64]u8 = undefined;
    for (0..parsers.sample_lines + 10) |i| {
        try input.appendSlice(gpa, try std.fmt.bufPrint(&line_buf, "0.0.0.0 host{d}.example.com\n", .{i}));
    }

    const out = try gpa.alloc(u8, sample_buf_len);
    defer gpa.free(out);
    const sample = try sampleOf(input.items, out);

    var lines = std.mem.tokenizeScalar(u8, sample, '\n');
    var count: usize = 0;
    while (lines.next()) |_| count += 1;
    try testing.expectEqual(parsers.sample_lines, count);
}

test "collectSample keeps the abp marker a long comment run would have hidden" {
    const gpa = testing.allocator;
    const long_comment = "! " ++ "c" ** (compiler.max_line_len - 2) ++ "\n";

    var input: std.ArrayList(u8) = .empty;
    defer input.deinit(gpa);
    for (0..20) |_| try input.appendSlice(gpa, long_comment);
    try input.appendSlice(gpa, "||ads.example.com^\n");

    const out = try gpa.alloc(u8, sample_buf_len);
    defer gpa.free(out);
    const sample = try sampleOf(input.items, out);

    try testing.expectEqualStrings("||ads.example.com^\n", sample);
    try testing.expectEqual(parsers.Format.abp, parsers.detectFormat(sample));
}

test "collectSample skips a line over max_line_len" {
    const gpa = testing.allocator;

    var input: std.ArrayList(u8) = .empty;
    defer input.deinit(gpa);
    try input.appendNTimes(gpa, 'x', 8 * compiler.max_line_len);
    try input.appendSlice(gpa, "\nads.example.com\n");

    const out = try gpa.alloc(u8, sample_buf_len);
    defer gpa.free(out);

    // A `Reader.fixed` holds the whole input, so the over-long line comes back
    // rather than being refused. The compiler would skip it, so the sniff does.
    const sample = try sampleOf(input.items, out);
    try testing.expectEqualStrings("ads.example.com\n", sample);
}

test "collectSample steps over a line that does not fit the reader buffer" {
    const gpa = testing.allocator;

    var input: std.ArrayList(u8) = .empty;
    defer input.deinit(gpa);
    try input.appendNTimes(gpa, 'x', 4 * compiler.max_line_len);
    try input.appendSlice(gpa, "\nads.example.com\n");

    // A reader buffer smaller than the long line makes `takeDelimiter` report
    // `error.StreamTooLong` and leave the stream where it was, which is the
    // path that loops forever without the discard.
    var backing: std.Io.Reader = .fixed(input.items);
    var reader_buf: [compiler.max_line_len]u8 = undefined;
    var limited = backing.limited(.unlimited, &reader_buf);

    const out = try gpa.alloc(u8, sample_buf_len);
    defer gpa.free(out);
    var w: std.Io.Writer = .fixed(out);
    try collectSample(&limited.interface, &w);

    try testing.expectEqualStrings("ads.example.com\n", w.buffered());
}

test "bodyChecksum covers the list body, then the wild body, then the allow body" {
    const all = bodyChecksum("a.example.com\n", "b.example.com\n", "c.example.com\n");
    var hasher = Sha256.init(.{});
    hasher.update("a.example.com\n");
    hasher.update(compiler.body_separator);
    hasher.update("b.example.com\n");
    hasher.update(compiler.body_separator);
    hasher.update("c.example.com\n");
    hasher.update(compiler.body_separator);
    var digest: [Sha256.digest_length]u8 = undefined;
    hasher.final(&digest);
    try testing.expectEqualStrings(&std.fmt.bytesToHex(digest, .lower), &all);

    // Order matters: the three parts are not interchangeable.
    try testing.expect(!std.mem.eql(
        u8,
        &all,
        &bodyChecksum("b.example.com\n", "a.example.com\n", "c.example.com\n"),
    ));
    try testing.expect(!std.mem.eql(
        u8,
        &all,
        &bodyChecksum("a.example.com\n", "c.example.com\n", "b.example.com\n"),
    ));
}

// ---------------------------------------------------------------------------
// wakeable scheduler (milestone-34 S3.6)
// ---------------------------------------------------------------------------

/// A `ScheduleClock` that never sleeps. Each park is recorded, then the clock
/// jumps straight to the deadline so the loop runs the next pass at once; a
/// budget of parks ends the run with the `error.Canceled` shutdown is the only
/// other source of. A park may also fire a `setSchedule`, which is what a
/// settings PUT landing while the scheduler waits looks like.
const StepClock = struct {
    const max_parks = 16;

    mutex: std.Io.Mutex = .init,
    manager: *Manager,
    now_s: i64 = 0,
    /// Deadline of each park in order; null means "parked with no deadline",
    /// which is what a disabled schedule does.
    parks: [max_parks]?i64 = @splat(null),
    park_count: usize = 0,
    budget: usize = 2,
    /// Fired from inside the park at this index, before the wait returns.
    change_at_park: ?usize = null,
    change_enabled: bool = true,
    change_hours: u16 = 1,

    fn clock(self: *StepClock) ScheduleClock {
        return .{ .ctx = self, .nowFn = now, .waitFn = wait };
    }

    fn now(ctx: ?*anyopaque, io: std.Io) i64 {
        const self: *StepClock = @ptrCast(@alignCast(ctx.?));
        self.mutex.lockUncancelable(io);
        defer self.mutex.unlock(io);
        return self.now_s;
    }

    fn wait(ctx: ?*anyopaque, io: std.Io, _: *std.Io.Event, deadline_s: ?i64) std.Io.Cancelable!void {
        const self: *StepClock = @ptrCast(@alignCast(ctx.?));

        self.mutex.lockUncancelable(io);
        const index = self.park_count;
        if (index < max_parks) self.parks[index] = deadline_s;
        self.park_count = index + 1;
        const fire_change = self.change_at_park == index;
        const over_budget = self.park_count >= self.budget;
        if (deadline_s) |d| self.now_s = d;
        self.mutex.unlock(io);

        // Taken outside this clock's own mutex: `setSchedule` takes the
        // manager's, and the loop reads this clock under neither.
        if (fire_change) {
            self.manager.setSchedule(io, self.change_enabled, self.change_hours);
            return;
        }
        if (over_budget or deadline_s == null) return error.Canceled;
    }

    fn parked(self: *StepClock) []const ?i64 {
        return self.parks[0..@min(self.park_count, max_parks)];
    }
};

const hour = 3_600;

test "the scheduler parks one interval past the anchor and again past each pass" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

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

    var f: fetcher.Fetcher = undefined;
    var manager = try testManager(&database, &f);
    defer manager.deinit(io);
    manager.update = .{ .enabled = true, .interval_hours = 2 };

    var step: StepClock = .{ .manager = &manager, .budget = 3 };
    manager.schedule_clock = step.clock();

    try testing.expectError(error.Canceled, manager.runScheduler(io));

    // The startup pass anchored at 0, so the first park is due at 2 h and each
    // completed pass re-anchors: 2 h, 4 h, 6 h.
    try testing.expectEqualSlices(?i64, &.{ 2 * hour, 4 * hour, 6 * hour }, step.parked());
}

test "a shortened interval moves the next refresh onto the new cadence" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

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

    var f: fetcher.Fetcher = undefined;
    var manager = try testManager(&database, &f);
    defer manager.deinit(io);
    manager.update = .{ .enabled = true, .interval_hours = 24 };

    // The PUT lands while the loop waits out the 24-hour deadline.
    var step: StepClock = .{
        .manager = &manager,
        .budget = 4,
        .change_at_park = 0,
        .change_enabled = true,
        .change_hours = 1,
    };
    manager.schedule_clock = step.clock();

    try testing.expectError(error.Canceled, manager.runScheduler(io));

    // Park 0 was the old 24-hour deadline; the change woke it, and every park
    // after it is one hour past the anchor the previous pass set.
    const parks = step.parked();
    try testing.expectEqual(@as(usize, 4), parks.len);
    try testing.expectEqual(@as(?i64, 24 * hour), parks[0]);
    try testing.expectEqual(@as(?i64, 24 * hour + hour), parks[1]);
    try testing.expectEqual(@as(?i64, 25 * hour + hour), parks[2]);
}

test "a disabled schedule parks with no deadline and the startup pass still runs" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

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

    var f: fetcher.Fetcher = undefined;
    var manager = try testManager(&database, &f);
    defer manager.deinit(io);
    manager.update = .{ .enabled = false, .interval_hours = 1 };

    var step: StepClock = .{ .manager = &manager, .budget = 8 };
    manager.schedule_clock = step.clock();

    try testing.expectError(error.Canceled, manager.runScheduler(io));

    // The startup pass ran — it published a snapshot even with updates off —
    // and then the loop parked once, on nothing.
    try testing.expect(manager.generation > 0);
    try testing.expectEqualSlices(?i64, &.{null}, step.parked());
}

test "re-enabling anchors the first interval on the last pass that ran" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

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

    var f: fetcher.Fetcher = undefined;
    var manager = try testManager(&database, &f);
    defer manager.deinit(io);
    manager.update = .{ .enabled = false, .interval_hours = 1 };

    var step: StepClock = .{
        .manager = &manager,
        .budget = 3,
        .change_at_park = 0,
        .change_enabled = true,
        .change_hours = 3,
    };
    manager.schedule_clock = step.clock();

    try testing.expectError(error.Canceled, manager.runScheduler(io));

    const parks = step.parked();
    // Park 0 is the disabled park; the enable wakes it, and the first deadline
    // is three hours past the STARTUP pass's anchor rather than past the
    // moment the operator flipped the switch.
    try testing.expectEqual(@as(?i64, null), parks[0]);
    try testing.expectEqual(@as(?i64, 3 * hour), parks[1]);
}

test "an interval already elapsed at enable time refreshes immediately" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

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

    var f: fetcher.Fetcher = undefined;
    var manager = try testManager(&database, &f);
    defer manager.deinit(io);
    manager.update = .{ .enabled = true, .interval_hours = 2 };

    var step: StepClock = .{ .manager = &manager, .budget = 2 };
    manager.schedule_clock = step.clock();
    // The anchor is four hours in the past, so two hours past it is already
    // gone and the loop must not wait at all before its first pass.
    manager.schedule_anchor_s = -4 * hour;
    step.now_s = 0;

    try testing.expectError(error.Canceled, manager.runScheduler(io));

    // The startup pass re-anchors at 0, so this proves nothing on its own
    // unless the anchor survives it; assert on the parks instead: the first
    // park is one interval past the startup anchor, never a wait for a
    // deadline already behind us.
    const parks = step.parked();
    try testing.expectEqual(@as(?i64, 2 * hour), parks[0]);
}

test "a gate-skipped pass advances the anchor rather than retrying early" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

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

    var f: fetcher.Fetcher = undefined;
    var manager = try testManager(&database, &f);
    defer manager.deinit(io);
    manager.update = .{ .enabled = true, .interval_hours = 2 };

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

    var step: StepClock = .{ .manager = &manager, .budget = 3 };
    manager.schedule_clock = step.clock();

    try testing.expectError(error.Canceled, manager.runScheduler(io));

    // Every scheduled pass was refused by the gate, and each one still spent
    // its slot: the deadlines march one interval at a time instead of
    // collapsing onto the same anchor.
    try testing.expectEqualSlices(?i64, &.{ 2 * hour, 4 * hour, 6 * hour }, step.parked());
    // The startup pass is gated too, so three refusals: one startup and the
    // two scheduled passes the parks above bracket.
    try testing.expectEqual(@as(u64, 3), manager.refreshesGated());
}

test "setSchedule is what the live schedule readers see" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

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

    var f: fetcher.Fetcher = undefined;
    var manager = try testManager(&database, &f);
    defer manager.deinit(io);

    manager.setSchedule(io, false, 6);
    const live = manager.schedule(io);
    try testing.expect(!live.enabled);
    try testing.expectEqual(@as(u16, 6), live.interval_hours);
    try testing.expect(manager.schedule_event.isSet());
}