nxdns/src/storage/disk_monitor.zig

//! Free-space monitor (PLAN §11.6). Samples the filesystem holding the data
//! directory every 60 seconds, classifies the result against the configured
//! thresholds, and publishes the state plus three size gauges through atomics.
//!
//! The state is the gate other components read before a non-essential write:
//! the query logger holds its batches while `writesAllowed` is false, and
//! the blocklist scheduler gates its refresh passes the same way
//! (`filter/manager.zig`'s `refreshGated`). Nothing here edits a
//! milestone-5 file; the gate is pulled, not pushed.

const std = @import("std");
const events = @import("events.zig");
const model = @import("../config/model.zig");
const statfs = @import("../platform/statfs.zig");

const log = std.log.scoped(.disk_monitor);

pub const sample_interval_s = 60;

pub const State = enum(u8) { ok, warn, critical };

pub const Gauges = struct {
    free_bytes: u64,
    db_bytes: u64,
    log_bytes: u64,
};

/// `min_free_mb` is checked first, so a configuration whose warn threshold sits
/// below its critical threshold still reports the more severe of the two.
pub fn classify(free_bytes: u64, cfg: model.Disk) State {
    if (free_bytes < model.minFreeBytes(cfg)) return .critical;
    if (free_bytes < model.warnFreeBytes(cfg)) return .warn;
    return .ok;
}

/// `min_free_mb` and `warn_free_mb` are one invariant pair — `classify` reads
/// both and reports the more severe verdict — so they live in one atomic word
/// and a reader unpacks a single load. Two atomics would let a sample land
/// between the two stores and classify against half of one configuration and
/// half of another.
fn packThresholds(d: model.Disk) u64 {
    return (@as(u64, d.min_free_mb) << 32) | d.warn_free_mb;
}

fn unpackThresholds(bits: u64) model.Disk {
    return .{
        .min_free_mb = @truncate(bits >> 32),
        .warn_free_mb = @truncate(bits),
    };
}

/// Where the measured log directory comes from. `sample` borrows the path
/// across a directory scan, so the path cannot simply be replaced under it: a
/// reader pins a generation for the whole borrow and `setLogDir` retires the
/// old one, which is freed by whichever of the two — the last reader or the
/// setter — finds it retired with no refs.
pub const LogDirSource = union(enum) {
    /// The path `init` was given, borrowed from the config. It outlives the
    /// process, so a reader still holding it after a swap is safe and it needs
    /// no pin. Null means logs do not go to a file.
    boot: ?[:0]const u8,
    /// Every generation `setLogDir` installs. Null means the same as above.
    installed: ?*LogDir,
};

/// A reader's hold on the log directory for the length of one scan. `pinned`
/// is null for the boot source, which nothing frees.
pub const LogDirBorrow = struct {
    path: ?[:0]const u8,
    pinned: ?*LogDir,
};

pub const LogDir = struct {
    path: [:0]const u8,
    refs: u32 = 0,
    retired: bool = false,
    /// Non-null exactly for heap generations, and the allocator that frees
    /// them.
    gpa: ?std.mem.Allocator = null,

    fn destroy(self: *LogDir) void {
        const gpa = self.gpa orelse return;
        gpa.free(self.path);
        gpa.destroy(self);
    }
};

pub const Monitor = struct {
    thresholds_packed: std.atomic.Value(u64),
    data_dir: std.Io.Dir,
    data_path: [:0]const u8,
    /// Guards `log_dir` and every generation's `refs`/`retired`.
    log_dir_mutex: std.Io.Mutex,
    log_dir: LogDirSource,

    state_raw: std.atomic.Value(u8),
    free_bytes: std.atomic.Value(u64),
    db_bytes: std.atomic.Value(u64),
    log_bytes: std.atomic.Value(u64),
    sample_failures: std.atomic.Value(u64),

    /// `data_dir` must be open with `.iterate = true`; sizing the databases
    /// scans it. `data_path` names the filesystem to measure and `data_dir` the
    /// directory to size — normally the same place, but `statvfs` takes a path
    /// and the scan takes a handle. `log_dir_path` resolves against the process
    /// working directory and is null when logs do not go to a file.
    pub fn init(
        cfg: model.Disk,
        data_dir: std.Io.Dir,
        data_path: [:0]const u8,
        log_dir_path: ?[:0]const u8,
    ) Monitor {
        return .{
            .thresholds_packed = .init(packThresholds(cfg)),
            .data_dir = data_dir,
            .data_path = data_path,
            .log_dir_mutex = .init,
            .log_dir = .{ .boot = log_dir_path },
            .state_raw = .init(@intFromEnum(State.ok)),
            .free_bytes = .init(0),
            .db_bytes = .init(0),
            .log_bytes = .init(0),
            .sample_failures = .init(0),
        };
    }

    /// The live threshold pair, from one load: `warn >= min` holds for every
    /// value this ever returns, whatever a concurrent `setThresholds` does.
    pub fn thresholds(self: *const Monitor) model.Disk {
        return unpackThresholds(self.thresholds_packed.load(.monotonic));
    }

    pub fn setThresholds(self: *Monitor, d: model.Disk) void {
        self.thresholds_packed.store(packThresholds(d), .monotonic);
    }

    /// Pins the log directory for one scan. Every borrow is matched by a
    /// `releaseLogDir`, which is what lets `setLogDir` free a generation the
    /// moment no scan is reading its path.
    pub fn acquireLogDir(self: *Monitor, io: std.Io) LogDirBorrow {
        self.log_dir_mutex.lockUncancelable(io);
        defer self.log_dir_mutex.unlock(io);

        switch (self.log_dir) {
            .boot => |path| return .{ .path = path, .pinned = null },
            .installed => |maybe| {
                const gen = maybe orelse return .{ .path = null, .pinned = null };
                gen.refs += 1;
                return .{ .path = gen.path, .pinned = gen };
            },
        }
    }

    pub fn releaseLogDir(self: *Monitor, io: std.Io, borrow: LogDirBorrow) void {
        const gen = borrow.pinned orelse return;

        self.log_dir_mutex.lockUncancelable(io);
        std.debug.assert(gen.refs > 0);
        gen.refs -= 1;
        const free_it = gen.retired and gen.refs == 0;
        self.log_dir_mutex.unlock(io);

        if (free_it) gen.destroy();
    }

    /// Prepare half of a log-directory change: allocates the owned path and
    /// its generation node before any commit, so publish cannot fail. `path`
    /// null means logs no longer go to a file and nothing is measured.
    pub fn prepareLogDir(
        gpa: std.mem.Allocator,
        path: ?[]const u8,
    ) std.mem.Allocator.Error!?*LogDir {
        const p = path orelse return null;
        const owned = try gpa.dupeZ(u8, p);
        errdefer gpa.free(owned);
        const gen = try gpa.create(LogDir);
        gen.* = .{ .path = owned, .gpa = gpa };
        return gen;
    }

    /// Discards a generation `prepareLogDir` built that will not be published.
    pub fn destroyPreparedLogDir(prepared: ?*LogDir) void {
        if (prepared) |gen| gen.destroy();
    }

    /// Publish half: infallible and I/O-free. The old generation is retired
    /// and freed here when no scan holds it, or by the last release otherwise.
    pub fn setLogDir(self: *Monitor, io: std.Io, prepared: ?*LogDir) void {
        self.log_dir_mutex.lockUncancelable(io);
        const old: ?*LogDir = switch (self.log_dir) {
            .boot => null,
            .installed => |maybe| maybe,
        };
        self.log_dir = .{ .installed = prepared };
        var free_old = false;
        if (old) |gen| {
            gen.retired = true;
            free_old = gen.refs == 0;
        }
        self.log_dir_mutex.unlock(io);

        if (free_old) old.?.destroy();
    }

    /// Frees any installed log-directory generation. Every scan must have
    /// released first, which shutdown ordering guarantees.
    pub fn deinit(self: *Monitor, io: std.Io) void {
        self.setLogDir(io, null);
    }

    pub fn state(self: *const Monitor) State {
        return @enumFromInt(self.state_raw.load(.monotonic));
    }

    pub fn writesAllowed(self: *const Monitor) bool {
        return self.state() != .critical;
    }

    pub fn gauges(self: *const Monitor) Gauges {
        return .{
            .free_bytes = self.free_bytes.load(.monotonic),
            .db_bytes = self.db_bytes.load(.monotonic),
            .log_bytes = self.log_bytes.load(.monotonic),
        };
    }

    /// One pass: free space from `statvfs`, then the two size gauges. A failed
    /// `statvfs` leaves the state untouched — an unreadable filesystem is not
    /// evidence that the disk filled — and a failed size scan leaves that one
    /// gauge at its previous reading. Every failure increments
    /// `sample_failures` and logs one line at `warn`.
    pub fn sample(self: *Monitor, io: std.Io, store: ?*events.Store, now_s: i64) void {
        const free = statfs.freeBytes(self.data_path) catch |err| {
            self.countFailure();
            log.warn("statvfs on {s} failed", .{self.data_path});
            probeFailed(store, io, now_s, "statvfs", "statvfs on the data path failed", err);
            return;
        };
        self.free_bytes.store(free, .monotonic);
        if (store) |s| s.resolve(io, now_s, .disk_probe, "statvfs");

        if (sumDir(io, self.data_dir, isDatabaseFile)) |bytes| {
            self.db_bytes.store(bytes, .monotonic);
            if (store) |s| s.resolve(io, now_s, .disk_probe, "data_dir");
        } else |err| {
            self.countFailure();
            log.warn("sizing the data directory failed: {s}", .{@errorName(err)});
            probeFailed(store, io, now_s, "data_dir", "sizing the data directory failed", err);
        }

        const borrow = self.acquireLogDir(io);
        defer self.releaseLogDir(io, borrow);
        if (borrow.path) |path| {
            if (self.sumLogDir(io, path)) |bytes| {
                self.log_bytes.store(bytes, .monotonic);
                if (store) |s| s.resolve(io, now_s, .disk_probe, "log_dir");
            } else |err| {
                self.countFailure();
                log.warn("sizing {s} failed: {s}", .{ path, @errorName(err) });
                probeFailed(store, io, now_s, "log_dir", "sizing the log directory failed", err);
            }
        }

        self.publish(io, store, now_s, classify(free, self.thresholds()), free);
    }

    /// Sample first, then sleep: a process that starts on a full disk must not
    /// serve a whole interval believing the state is `.ok`. `.boot` so a
    /// suspended box still sees the interval elapse.
    pub fn run(self: *Monitor, io: std.Io, store: ?*events.Store) std.Io.Cancelable!void {
        const interval: std.Io.Clock.Duration = .{
            .raw = .fromSeconds(sample_interval_s),
            .clock = .boot,
        };
        while (true) {
            self.sample(io, store, std.Io.Clock.real.now(io).toSeconds());
            try interval.sleep(io);
        }
    }

    fn countFailure(self: *Monitor) void {
        _ = self.sample_failures.fetchAdd(1, .monotonic);
    }

    /// Logs on transitions only. A disk that sits at `.warn` for a week
    /// produces one line, not ten thousand.
    fn publish(
        self: *Monitor,
        io: std.Io,
        store: ?*events.Store,
        now_s: i64,
        next: State,
        free: u64,
    ) void {
        const previous: State = @enumFromInt(self.state_raw.swap(@intFromEnum(next), .monotonic));
        if (previous == next) return;
        log.warn("disk state {t} -> {t}: {d} bytes free on {s}", .{
            previous,
            next,
            free,
            self.data_path,
        });

        const s = store orelse return;
        if (next == .ok) {
            s.resolve(io, now_s, .disk_space, disk_space_key);
            return;
        }
        var buf: [events.Store.max_detail_len]u8 = undefined;
        const detail = std.fmt.bufPrint(&buf, "disk state {t} -> {t}: {d} bytes free on {s}", .{
            previous,
            next,
            free,
            self.data_path,
        }) catch buf[0..];
        s.report(io, now_s, .disk_space, disk_space_key, "data directory", switch (next) {
            .warn => .warning,
            .critical => .@"error",
            .ok => unreachable,
        }, detail);
    }

    fn sumLogDir(self: *Monitor, io: std.Io, path: [:0]const u8) !u64 {
        _ = self;
        var dir = try std.Io.Dir.cwd().openDir(io, path, .{ .iterate = true });
        defer dir.close(io);
        return sumDir(io, dir, everyFile);
    }
};

/// The one subject `disk.space` ever has: this box has exactly one data
/// directory, and its filesystem is what the thresholds classify.
const disk_space_key = "data";

/// Every probe failure is a warning, not an error: an unreadable filesystem is
/// a gap in what the monitor can see, and the state it published last stands.
fn probeFailed(
    store: ?*events.Store,
    io: std.Io,
    now_s: i64,
    operation: []const u8,
    message: []const u8,
    err: anyerror,
) void {
    const s = store orelse return;
    var buf: [events.Store.max_detail_len]u8 = undefined;
    const detail = std.fmt.bufPrint(&buf, "{s}: {s}", .{ message, @errorName(err) }) catch buf[0..];
    s.report(io, now_s, .disk_probe, operation, operation, .warning, detail);
}

fn everyFile(_: []const u8) bool {
    return true;
}

/// The sqlite trio: `x.db`, its write-ahead log and its shared-memory index.
/// All three live on the watched filesystem and all three grow.
fn isDatabaseFile(name: []const u8) bool {
    return std.mem.endsWith(u8, name, ".db") or
        std.mem.endsWith(u8, name, ".db-wal") or
        std.mem.endsWith(u8, name, ".db-shm");
}

fn sumDir(io: std.Io, dir: std.Io.Dir, accept: *const fn ([]const u8) bool) !u64 {
    var total: u64 = 0;
    var it = dir.iterate();
    while (try it.next(io)) |entry| {
        if (entry.kind != .file) continue;
        if (!accept(entry.name)) continue;
        // A file that vanishes between `iterate` and `statFile` is normal:
        // rotation and database recreate both delete under a running scan. Any
        // other stat failure makes the whole scan fail, because a partial total
        // published as a gauge reads as a shrinking database.
        const st = dir.statFile(io, entry.name, .{}) catch |err| switch (err) {
            error.FileNotFound => continue,
            else => return err,
        };
        total += st.size;
    }
    return total;
}

// ---------------------------------------------------------------------------
// tests
// ---------------------------------------------------------------------------

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

const mb = 1024 * 1024;

/// Set by the vanished-file test only: `acceptGoneDeleted` needs a handle and an
/// io, and the `accept` signature carries neither.
var vanish_dir: ?std.Io.Dir = null;
var vanish_io: ?std.Io = null;

/// Deletes `gone.db` between `iterate` and `statFile`, which is the race the
/// scan must tolerate.
fn acceptGoneDeleted(name: []const u8) bool {
    if (!isDatabaseFile(name)) return false;
    if (std.mem.eql(u8, name, "gone.db")) {
        vanish_dir.?.deleteFile(vanish_io.?, name) catch {};
    }
    return true;
}

test "classify below the critical threshold" {
    const cfg: model.Disk = .{ .min_free_mb = 200, .warn_free_mb = 500 };
    try testing.expectEqual(State.critical, classify(0, cfg));
    try testing.expectEqual(State.critical, classify(199 * mb, cfg));
    try testing.expectEqual(State.critical, classify(200 * mb - 1, cfg));
}

test "classify at and above the critical threshold" {
    const cfg: model.Disk = .{ .min_free_mb = 200, .warn_free_mb = 500 };
    try testing.expectEqual(State.warn, classify(200 * mb, cfg));
    try testing.expectEqual(State.warn, classify(350 * mb, cfg));
    try testing.expectEqual(State.warn, classify(500 * mb - 1, cfg));
}

test "classify at and above the warn threshold" {
    const cfg: model.Disk = .{ .min_free_mb = 200, .warn_free_mb = 500 };
    try testing.expectEqual(State.ok, classify(500 * mb, cfg));
    try testing.expectEqual(State.ok, classify(64 * 1024 * mb, cfg));
}

test "classify with both thresholds at zero never leaves ok" {
    const cfg: model.Disk = .{ .min_free_mb = 0, .warn_free_mb = 0 };
    try testing.expectEqual(State.ok, classify(0, cfg));
    try testing.expectEqual(State.ok, classify(1, cfg));
}

test "classify reports the more severe state when warn sits below min" {
    const cfg: model.Disk = .{ .min_free_mb = 500, .warn_free_mb = 200 };
    try testing.expectEqual(State.critical, classify(300 * mb, cfg));
    try testing.expectEqual(State.ok, classify(500 * mb, cfg));
}

test "the database file filter accepts the sqlite trio only" {
    try testing.expect(isDatabaseFile("querylog.db"));
    try testing.expect(isDatabaseFile("querylog.db-wal"));
    try testing.expect(isDatabaseFile("querylog.db-shm"));
    try testing.expect(!isDatabaseFile("querylog.db.bak"));
    try testing.expect(!isDatabaseFile("nxdns.log"));
    try testing.expect(!isDatabaseFile(""));
}

test "init reports ok with zero gauges and allows writes" {
    var monitor: Monitor = .init(.{}, std.Io.Dir.cwd(), ".", null);
    try testing.expectEqual(State.ok, monitor.state());
    try testing.expect(monitor.writesAllowed());
    try testing.expectEqual(Gauges{ .free_bytes = 0, .db_bytes = 0, .log_bytes = 0 }, monitor.gauges());
    try testing.expectEqual(@as(u64, 0), monitor.sample_failures.load(.monotonic));
}

test "writesAllowed is false only at critical" {
    var monitor: Monitor = .init(.{}, std.Io.Dir.cwd(), ".", null);
    monitor.state_raw.store(@intFromEnum(State.warn), .monotonic);
    try testing.expect(monitor.writesAllowed());
    monitor.state_raw.store(@intFromEnum(State.critical), .monotonic);
    try testing.expect(!monitor.writesAllowed());
}

test "a sample sizes the databases and ignores every other file" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

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

    try tmp.dir.writeFile(io, .{ .sub_path = "querylog.db", .data = &[_]u8{'a'} ** 100 });
    try tmp.dir.writeFile(io, .{ .sub_path = "querylog.db-wal", .data = &[_]u8{'b'} ** 50 });
    try tmp.dir.writeFile(io, .{ .sub_path = "querylog.db-shm", .data = &[_]u8{'c'} ** 10 });
    try tmp.dir.writeFile(io, .{ .sub_path = "notes.txt", .data = &[_]u8{'d'} ** 4096 });

    var monitor: Monitor = .init(.{ .min_free_mb = 0, .warn_free_mb = 0 }, tmp.dir, ".", null);
    monitor.sample(io, null, 0);

    const g = monitor.gauges();
    try testing.expectEqual(@as(u64, 160), g.db_bytes);
    try testing.expectEqual(@as(u64, 0), g.log_bytes);
    try testing.expect(g.free_bytes > 0);
    try testing.expectEqual(@as(u64, 0), monitor.sample_failures.load(.monotonic));
    try testing.expectEqual(State.ok, monitor.state());
}

test "a sample sizes every file in the log directory" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var tmp = testing.tmpDir(.{ .iterate = true });
    defer tmp.cleanup();
    try tmp.dir.createDirPath(io, "logs");

    var logs = try tmp.dir.openDir(io, "logs", .{});
    defer logs.close(io);
    try logs.writeFile(io, .{ .sub_path = "nxdns.log", .data = &[_]u8{'a'} ** 300 });
    try logs.writeFile(io, .{ .sub_path = "nxdns.log.1", .data = &[_]u8{'b'} ** 700 });

    var path_buf: [256]u8 = undefined;
    const log_path = try std.fmt.bufPrintZ(&path_buf, ".zig-cache/tmp/{s}/logs", .{tmp.sub_path});

    var monitor: Monitor = .init(.{ .min_free_mb = 0, .warn_free_mb = 0 }, tmp.dir, ".", log_path);
    monitor.sample(io, null, 0);

    try testing.expectEqual(@as(u64, 1000), monitor.gauges().log_bytes);
    try testing.expectEqual(@as(u64, 0), monitor.sample_failures.load(.monotonic));
}

test "a failed statvfs counts and keeps the previous state" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var monitor: Monitor = .init(.{}, std.Io.Dir.cwd(), "./nxdns-no-such-path-7c21", null);
    monitor.state_raw.store(@intFromEnum(State.warn), .monotonic);
    monitor.sample(io, null, 0);

    try testing.expectEqual(State.warn, monitor.state());
    try testing.expectEqual(@as(u64, 1), monitor.sample_failures.load(.monotonic));
    try testing.expectEqual(@as(u64, 0), monitor.gauges().free_bytes);
}

test "an unreadable log directory counts a failure but still publishes a state" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

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

    var monitor: Monitor = .init(
        .{ .min_free_mb = 0, .warn_free_mb = 0 },
        tmp.dir,
        ".",
        "./nxdns-no-such-dir-4f8a",
    );
    monitor.sample(io, null, 0);

    try testing.expectEqual(@as(u64, 1), monitor.sample_failures.load(.monotonic));
    try testing.expectEqual(State.ok, monitor.state());
    try testing.expect(monitor.gauges().free_bytes > 0);
}

test "a file deleted during the scan is skipped and the rest still counts" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

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

    try tmp.dir.writeFile(io, .{ .sub_path = "gone.db", .data = &[_]u8{'a'} ** 100 });
    try tmp.dir.writeFile(io, .{ .sub_path = "stays.db", .data = &[_]u8{'b'} ** 40 });

    vanish_dir = tmp.dir;
    vanish_io = io;
    defer vanish_dir = null;

    try testing.expectEqual(@as(u64, 40), try sumDir(io, tmp.dir, acceptGoneDeleted));
}

test "an unreadable data directory fails the scan and keeps the previous gauge" {
    // Mode bits do not apply to root, so the denial the test needs cannot happen.
    if (std.c.geteuid() == 0) return error.SkipZigTest;

    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

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

    try tmp.dir.writeFile(io, .{ .sub_path = "querylog.db", .data = &[_]u8{'a'} ** 100 });

    var monitor: Monitor = .init(.{ .min_free_mb = 0, .warn_free_mb = 0 }, tmp.dir, ".", null);
    monitor.db_bytes.store(4096, .monotonic);

    // The handle keeps its read permission from open time, so `iterate` still
    // lists the file, but path resolution under the directory now fails.
    try tmp.dir.setPermissions(io, .fromMode(0o600));
    monitor.sample(io, null, 0);
    try tmp.dir.setPermissions(io, .fromMode(0o700));

    try testing.expectEqual(@as(u64, 4096), monitor.gauges().db_bytes);
    try testing.expectEqual(@as(u64, 1), monitor.sample_failures.load(.monotonic));
    try testing.expectEqual(State.ok, monitor.state());
}

test "a threshold above the real free space drives the state to critical" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

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

    const unreachable_mb = std.math.maxInt(u32);
    var monitor: Monitor = .init(
        .{ .min_free_mb = unreachable_mb, .warn_free_mb = unreachable_mb },
        tmp.dir,
        ".",
        null,
    );
    monitor.sample(io, null, 0);
    try testing.expectEqual(State.critical, monitor.state());
    try testing.expect(!monitor.writesAllowed());

    monitor.setThresholds(.{ .min_free_mb = 0, .warn_free_mb = 0 });
    monitor.sample(io, null, 0);
    try testing.expectEqual(State.ok, monitor.state());
    try testing.expect(monitor.writesAllowed());
    try testing.expectEqual(@as(u64, 0), monitor.sample_failures.load(.monotonic));
}

test "a disk transition records an episode per severity and closes it on recovery" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

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

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

    const unreachable_mb = std.math.maxInt(u32);
    var monitor: Monitor = .init(
        .{ .min_free_mb = unreachable_mb, .warn_free_mb = unreachable_mb },
        tmp.dir,
        ".",
        null,
    );
    monitor.sample(io, &fx.store, 1000);
    try testing.expectEqual(State.critical, monitor.state());
    try testing.expectEqualStrings("disk.space", try fx.text(
        "SELECT code FROM operational_events WHERE resolved_at IS NULL",
    ));
    try testing.expectEqualStrings("error", try fx.text(
        "SELECT severity FROM operational_events WHERE resolved_at IS NULL",
    ));
    try testing.expectEqualStrings("data", try fx.text(
        "SELECT subject_key FROM operational_events WHERE resolved_at IS NULL",
    ));

    // A second critical sample is the same episode, not a second row: `publish`
    // only reports on a transition.
    monitor.sample(io, &fx.store, 1060);
    try testing.expectEqual(@as(i64, 1), try fx.count("SELECT count(*) FROM operational_events"));

    monitor.setThresholds(.{ .min_free_mb = 0, .warn_free_mb = 0 });
    monitor.sample(io, &fx.store, 1120);
    try testing.expectEqual(State.ok, monitor.state());
    try testing.expectEqual(
        @as(i64, 0),
        try fx.count("SELECT count(*) FROM operational_events WHERE resolved_at IS NULL"),
    );
    try testing.expectEqual(@as(i64, 1120), try fx.count("SELECT resolved_at FROM operational_events"));
}

test "a failed probe opens an episode the next clean pass closes" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

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

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

    var monitor: Monitor = .init(
        .{ .min_free_mb = 0, .warn_free_mb = 0 },
        tmp.dir,
        ".",
        "./nxdns-no-such-dir-4f8a",
    );
    monitor.sample(io, &fx.store, 1000);

    try testing.expectEqualStrings("disk.probe", try fx.text(
        "SELECT code FROM operational_events WHERE resolved_at IS NULL",
    ));
    try testing.expectEqualStrings("log_dir", try fx.text(
        "SELECT subject_key FROM operational_events WHERE resolved_at IS NULL",
    ));
    try testing.expectEqualStrings("warning", try fx.text(
        "SELECT severity FROM operational_events WHERE resolved_at IS NULL",
    ));

    try tmp.dir.createDirPath(io, "logs");
    var path_buf: [256]u8 = undefined;
    const good_dir = try std.fmt.bufPrint(&path_buf, ".zig-cache/tmp/{s}/logs", .{tmp.sub_path});
    monitor.setLogDir(io, try Monitor.prepareLogDir(testing.allocator, good_dir));
    defer monitor.deinit(io);
    monitor.sample(io, &fx.store, 1100);

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

/// Alternates between two pairs that each satisfy `warn >= min`, so any
/// observed pair violating it can only have been torn out of two stores.
fn storeThresholdPairs(monitor: *Monitor, rounds: usize) void {
    for (0..rounds) |i| {
        monitor.setThresholds(if (i % 2 == 0)
            .{ .min_free_mb = 1, .warn_free_mb = 2 }
        else
            .{ .min_free_mb = 3_000_000, .warn_free_mb = 4_000_000 });
    }
}

test "a threshold reader never observes a pair from two different stores" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var monitor: Monitor = .init(.{ .min_free_mb = 1, .warn_free_mb = 2 }, std.Io.Dir.cwd(), ".", null);

    const rounds = 20_000;
    var writer = try io.concurrent(storeThresholdPairs, .{ &monitor, rounds });
    // Recorded, not asserted, while the writer runs: an assertion that returned
    // here would leave `Threaded.deinit` joining a task nothing ends.
    var torn = false;
    for (0..rounds) |_| {
        const pair = monitor.thresholds();
        if (pair.warn_free_mb < pair.min_free_mb) torn = true;
    }
    writer.await(io);
    try testing.expect(!torn);
}

test "every emit site is inert when the store is absent" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var monitor: Monitor = .init(
        .{ .min_free_mb = std.math.maxInt(u32), .warn_free_mb = std.math.maxInt(u32) },
        std.Io.Dir.cwd(),
        "./nxdns-no-such-path-7c21",
        "./nxdns-no-such-dir-4f8a",
    );
    monitor.sample(io, null, 0);
    try testing.expectEqual(@as(u64, 1), monitor.sample_failures.load(.monotonic));
}

// ---------------------------------------------------------------------------
// setLogDir (milestone-34 S3.5)
// ---------------------------------------------------------------------------

test "setLogDir re-points the measurement and frees the retired generation" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var tmp = testing.tmpDir(.{ .iterate = true });
    defer tmp.cleanup();
    try tmp.dir.createDirPath(io, "first");
    try tmp.dir.createDirPath(io, "second");
    try tmp.dir.writeFile(io, .{ .sub_path = "first/nxdns.log", .data = "aaaa" });
    try tmp.dir.writeFile(io, .{ .sub_path = "second/nxdns.log", .data = "bbbbbbbb" });

    var first_buf: [160]u8 = undefined;
    var second_buf: [160]u8 = undefined;
    const first = try std.fmt.bufPrint(&first_buf, ".zig-cache/tmp/{s}/first", .{tmp.sub_path});
    const second = try std.fmt.bufPrint(&second_buf, ".zig-cache/tmp/{s}/second", .{tmp.sub_path});

    var monitor: Monitor = .init(.{ .min_free_mb = 0, .warn_free_mb = 0 }, tmp.dir, ".", null);
    defer monitor.deinit(io);

    // Boot measures nothing.
    monitor.sample(io, null, 1_000);
    try testing.expectEqual(@as(u64, 0), monitor.gauges().log_bytes);

    monitor.setLogDir(io, try Monitor.prepareLogDir(testing.allocator, first));
    monitor.sample(io, null, 1_100);
    try testing.expectEqual(@as(u64, 4), monitor.gauges().log_bytes);

    // The retired generation is freed here; the testing allocator says so.
    monitor.setLogDir(io, try Monitor.prepareLogDir(testing.allocator, second));
    monitor.sample(io, null, 1_200);
    try testing.expectEqual(@as(u64, 8), monitor.gauges().log_bytes);

    // Output moved away from file: nothing is measured, and the gauge keeps
    // its last reading rather than claiming zero bytes of logs.
    monitor.setLogDir(io, null);
    monitor.sample(io, null, 1_300);
    try testing.expectEqual(@as(u64, 8), monitor.gauges().log_bytes);
}

test "a prepared log directory that is never published is freed by the caller" {
    const prepared = try Monitor.prepareLogDir(testing.allocator, "/var/log/nxdns");
    Monitor.destroyPreparedLogDir(prepared);
    try testing.expectEqual(@as(?*LogDir, null), try Monitor.prepareLogDir(testing.allocator, null));
}

test "a sample borrowing a log directory survives a concurrent setLogDir" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var tmp = testing.tmpDir(.{ .iterate = true });
    defer tmp.cleanup();
    try tmp.dir.createDirPath(io, "logs");
    try tmp.dir.writeFile(io, .{ .sub_path = "logs/nxdns.log", .data = "aaaa" });

    var path_buf: [160]u8 = undefined;
    const logs = try std.fmt.bufPrint(&path_buf, ".zig-cache/tmp/{s}/logs", .{tmp.sub_path});

    var monitor: Monitor = .init(.{ .min_free_mb = 0, .warn_free_mb = 0 }, tmp.dir, ".", null);
    defer monitor.deinit(io);
    monitor.setLogDir(io, try Monitor.prepareLogDir(testing.allocator, logs));

    const Racer = struct {
        fn sample(m: *Monitor, sio: std.Io) void {
            for (0..200) |i| m.sample(sio, null, @intCast(1_000 + i));
        }
        fn repoint(m: *Monitor, sio: std.Io, p: []const u8) void {
            for (0..200) |i| {
                const prepared = Monitor.prepareLogDir(testing.allocator, if (i % 2 == 0) p else null) catch return;
                m.setLogDir(sio, prepared);
            }
        }
    };

    var group: std.Io.Group = .init;
    defer group.cancel(io);
    try group.concurrent(io, Racer.sample, .{ &monitor, io });
    try group.concurrent(io, Racer.repoint, .{ &monitor, io, logs });
    try group.await(io);
}