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

pub const Monitor = struct {
    cfg: model.Disk,
    data_dir: std.Io.Dir,
    data_path: [:0]const u8,
    log_dir_path: ?[:0]const u8,

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

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

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

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

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

    /// One pass: free space from `statvfs`, then the two size gauges. A failed
    /// `statvfs` leaves the state untouched — an unreadable filesystem is not
    /// evidence that the disk filled — and a failed size scan leaves that one
    /// gauge at its previous reading. Every failure increments
    /// `sample_failures` and logs one line at `warn`.
    pub fn sample(self: *Monitor, io: std.Io, 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);
        }

        if (self.log_dir_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.cfg), free);
    }

    /// Sample first, then sleep: a process that starts on a full disk must not
    /// serve a whole interval believing the state is `.ok`. `.boot` so a
    /// suspended box still sees the interval elapse.
    pub fn run(self: *Monitor, io: std.Io, 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.cfg = .{ .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.cfg = .{ .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;
    monitor.log_dir_path = try std.fmt.bufPrintZ(&path_buf, ".zig-cache/tmp/{s}/logs", .{tmp.sub_path});
    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"),
    );
}

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