//! Per-minute upstream outcome history (milestone-26 rulings 1, 3, 4).
//!
//! Outcomes are aggregated into their wall-clock UTC minute at the moment the
//! pool records them, and the aggregates are flushed to `querylog.db` once a
//! minute. **Nothing samples a lifetime counter and subtracts.** That is what
//! makes the stored numbers additive facts: a restart inside a minute adds to
//! the same row, a crash loses at most the cells that had not flushed yet, and
//! no path anywhere can produce a negative delta.
//!
//! The query path may not touch SQLite, so recording is memory-only under this
//! module's own mutex and the writing happens on a task of its own.
//!
//! Two failure modes, deliberately kept apart:
//!
//!   * **Overflow.** More live `(url, minute)` pairs than `max_pending`. The
//!     oldest minute is dropped, `rows_dropped` counts it and
//!     `last_drop_minute` remembers how new the newest lost minute was, so a
//!     window that starts after it can still be reported as complete.
//!   * **A failed flush.** Nothing is dropped: the rows go back into the
//!     accumulator and the next pass writes them again.
//!
//! The wall clock, not `.awake`: history participates in wall-clock periods, so
//! a minute here is the same minute the dashboard's period picker means.
//! Routing state (`health.zig`) stays on `.awake` and is untouched by this
//! file.

const std = @import("std");

const db = @import("../storage/db.zig");
const health = @import("health.zig");
const upstream_history_repo = @import("../storage/repositories/upstream_history_repo.zig");

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

/// Live `(url, minute)` cells. Not a knob (m26 anti-requirements). A household
/// pool is a handful of upstreams, so the bound is only reachable when flushing
/// has been failing for hours.
pub const max_pending = 4096;

/// One flush pass per minute, so an unflushed cell is at most about a minute
/// old. `.boot` rather than `.awake`, for `retention.zig`'s reason: a box that
/// suspends must still see its interval elapse.
pub const flush_interval_s = 60;

/// The UTC minute `wall_s` falls in, as its start in seconds. `@divFloor`, not
/// `@divTrunc`: a negative second belongs to the minute before it.
pub fn minuteOf(wall_s: i64) i64 {
    return @divFloor(wall_s, 60) * 60;
}

/// The write seam. Production passes `upstream_history_repo.flush`; a test
/// passes a stub that fails, or one that records what it was handed.
pub const WriteFn = *const fn (*db.Db, []const upstream_history_repo.FlushRow) db.Error!void;

pub const Accumulator = struct {
    pub const Cell = struct {
        /// Borrowed from the pool entry's endpoint, which lives as long as the
        /// process. Nothing here copies it, and nothing here may outlive it.
        url: []const u8,
        minute_ts: i64,
        /// Both counters saturate instead of wrapping: every write site uses
        /// `+|=` — `recordSuccess`, `recordFailure`, and `mergeBack`, which
        /// sums a failed flush's copy back into the live cell. The saturation
        /// is deliberate and unreachable: a cell counts one upstream's
        /// outcomes inside a single wall-clock minute, so filling a `u32`
        /// would take about 72 million exchanges per second with that one
        /// upstream. Nothing reports it, by design — `last_drop_minute` and
        /// the `complete` flag it feeds describe capacity drops, and a
        /// saturated counter is not a drop.
        successes: u32,
        failures: u32,
        last_failure_ts: ?i64,
        last_error_buf: [health.error_name_capacity]u8,
        last_error_len: u8,

        fn lastError(self: *const Cell) []const u8 {
            return self.last_error_buf[0..self.last_error_len];
        }
    };

    /// A consistent copy for `/metrics`, `/api/health` and the API layer.
    pub const Stats = struct {
        flushes: u64 = 0,
        flush_failures: u64 = 0,
        rows_dropped: u64 = 0,
        pending: u32 = 0,
        /// The newest minute capacity has ever cost this process, or null when
        /// nothing was ever dropped. A window that starts after it is complete
        /// again, so one historical overflow does not mark every later answer.
        last_drop_minute: ?i64 = null,
        /// Current state, not a count: set by a failed flush and cleared by the
        /// next successful one. Feeds the `/api/health` rollup.
        last_flush_failed: bool = false,
    };

    /// Atomic for the reason `retention.zig`'s are: the flush task writes them
    /// and the web task reads them, on different threads. They are bumped
    /// outside the mutex, so the lock is not what orders them.
    const Counters = struct {
        flushes: std.atomic.Value(u64) = .init(0),
        flush_failures: std.atomic.Value(u64) = .init(0),
        rows_dropped: std.atomic.Value(u64) = .init(0),
    };

    mutex: std.Io.Mutex = .init,
    cells: [max_pending]Cell,
    count: u32,
    last_drop_minute: ?i64,
    last_flush_failed: bool,
    counters: Counters,
    /// Owned by whichever task runs `flushOnce`, which is one task. It is a
    /// field rather than a local so that the megabyte it costs lives wherever
    /// the accumulator was placed instead of on a task's stack.
    flush_cells: [max_pending]Cell,
    flush_rows: [max_pending]upstream_history_repo.FlushRow,
    flush_count: u32,

    /// `cells` and `flush_cells` are `undefined`: a cell is always written
    /// before it is read, and `count` is what says which ones exist.
    pub const init: Accumulator = .{
        .mutex = .init,
        .cells = undefined,
        .count = 0,
        .last_drop_minute = null,
        .last_flush_failed = false,
        .counters = .{},
        .flush_cells = undefined,
        .flush_rows = undefined,
        .flush_count = 0,
    };

    pub fn recordSuccess(self: *Accumulator, io: std.Io, url: []const u8, wall_s: i64) void {
        // Uncancelable for the reason the pool's health sections are: this
        // takes no Io and never blocks on a peer, and losing the record of a
        // completed exchange to a cancellation would undercount for good.
        self.mutex.lockUncancelable(io);
        defer self.mutex.unlock(io);
        const cell = self.cellFor(url, minuteOf(wall_s));
        cell.successes +|= 1;
    }

    pub fn recordFailure(
        self: *Accumulator,
        io: std.Io,
        url: []const u8,
        wall_s: i64,
        error_name: []const u8,
    ) void {
        self.mutex.lockUncancelable(io);
        defer self.mutex.unlock(io);
        const cell = self.cellFor(url, minuteOf(wall_s));
        cell.failures +|= 1;
        noteFailure(cell, wall_s, error_name);
    }

    /// The only read surface. Every field above is private to this module, so
    /// no consumer can read one of them without the mutex.
    pub fn snapshotStats(self: *Accumulator, io: std.Io) Stats {
        self.mutex.lockUncancelable(io);
        defer self.mutex.unlock(io);
        return .{
            .flushes = self.counters.flushes.load(.monotonic),
            .flush_failures = self.counters.flush_failures.load(.monotonic),
            .rows_dropped = self.counters.rows_dropped.load(.monotonic),
            .pending = self.count,
            .last_drop_minute = self.last_drop_minute,
            .last_flush_failed = self.last_flush_failed,
        };
    }

    /// The cell for `(url, minute_ts)`, created if it is not there yet. The
    /// caller holds the mutex.
    ///
    /// Linear: the live set is one cell per upstream per unflushed minute,
    /// which at household scale is a handful. A miss at capacity evicts the
    /// oldest minute — see `evictOldest`.
    fn cellFor(self: *Accumulator, url: []const u8, minute_ts: i64) *Cell {
        for (self.cells[0..self.count]) |*cell| {
            if (cell.minute_ts != minute_ts) continue;
            // Pointer equality first: every call from the pool passes the same
            // `Entry.endpoint.url`, so the byte compare is the cold path.
            if (cell.url.ptr == url.ptr and cell.url.len == url.len) return cell;
            if (std.mem.eql(u8, cell.url, url)) return cell;
        }

        const slot = if (self.count < max_pending) fresh: {
            const index = self.count;
            self.count += 1;
            break :fresh &self.cells[index];
        } else self.evictOldest();

        slot.* = .{
            .url = url,
            .minute_ts = minute_ts,
            .successes = 0,
            .failures = 0,
            .last_failure_ts = null,
            .last_error_buf = @splat(0),
            .last_error_len = 0,
        };
        return slot;
    }

    /// Frees the cell holding the oldest minute and accounts for what it cost.
    ///
    /// `last_drop_minute` moves through `@max` and never through assignment: a
    /// merge-back after a failed flush can evict a cell older than one already
    /// dropped, and a watermark that moved backwards would report a window as
    /// complete when outcomes inside it are gone.
    fn evictOldest(self: *Accumulator) *Cell {
        var oldest: usize = 0;
        for (self.cells[1..self.count], 1..) |*cell, i| {
            if (cell.minute_ts < self.cells[oldest].minute_ts) oldest = i;
        }
        const evicted = self.cells[oldest].minute_ts;
        self.last_drop_minute = @max(self.last_drop_minute orelse evicted, evicted);
        _ = self.counters.rows_dropped.fetchAdd(1, .monotonic);
        return &self.cells[oldest];
    }

    /// One flush pass: swap the dirty cells out, write them, and on failure put
    /// them back.
    ///
    /// **A swap, never a subtraction.** SQLite runs outside the mutex, so while
    /// it does, a full accumulator can evict a cell that was copied out and
    /// then recreate the same `(url, minute_ts)`. A post-flush subtract would
    /// then destroy outcomes recorded during the write. Moving the cells out
    /// makes the flush own them: what is recorded beside it is new data, and a
    /// failure merges the two additively.
    ///
    /// Every failure is counted and warned about once; nothing here returns an
    /// error, because there is no caller that could do anything the next pass
    /// will not do anyway.
    pub fn flushOnce(self: *Accumulator, io: std.Io, database: *db.Db, write: WriteFn) void {
        {
            self.mutex.lockUncancelable(io);
            defer self.mutex.unlock(io);
            if (self.count == 0) return;
            @memcpy(self.flush_cells[0..self.count], self.cells[0..self.count]);
            self.flush_count = self.count;
            self.count = 0;
        }

        const rows = self.flush_rows[0..self.flush_count];
        for (self.flush_cells[0..self.flush_count], rows) |*cell, *row| {
            row.* = .{
                .url = cell.url,
                .minute_ts = cell.minute_ts,
                .successes = cell.successes,
                .failures = cell.failures,
                .last_failure_ts = cell.last_failure_ts,
                .last_error = cell.lastError(),
            };
        }

        if (write(database, rows)) {
            _ = self.counters.flushes.fetchAdd(1, .monotonic);
            self.mutex.lockUncancelable(io);
            defer self.mutex.unlock(io);
            self.last_flush_failed = false;
            return;
        } else |err| {
            _ = self.counters.flush_failures.fetchAdd(1, .monotonic);
            log.warn("flushing {d} upstream history rows failed: {s}", .{ rows.len, @errorName(err) });
            self.mergeBack(io);
        }
    }

    /// Puts a failed pass's cells back through the rules recording uses: a cell
    /// recorded during the write keeps its outcomes and the merge sums into it,
    /// and a merge that overflows follows the ordinary drop policy.
    fn mergeBack(self: *Accumulator, io: std.Io) void {
        self.mutex.lockUncancelable(io);
        defer self.mutex.unlock(io);
        self.last_flush_failed = true;

        for (self.flush_cells[0..self.flush_count]) |*saved| {
            const cell = self.cellFor(saved.url, saved.minute_ts);
            cell.successes +|= saved.successes;
            cell.failures +|= saved.failures;
            if (saved.last_failure_ts) |at| noteFailure(cell, at, saved.lastError());
        }
    }

    /// Daily-loop shape (`retention.zig`): flush first, then sleep, so a
    /// process that is about to be canceled has already written once.
    pub fn run(self: *Accumulator, io: std.Io, database: *db.Db) std.Io.Cancelable!void {
        const interval: std.Io.Clock.Duration = .{
            .raw = .fromSeconds(flush_interval_s),
            .clock = .boot,
        };
        while (true) {
            self.flushOnce(io, database, upstream_history_repo.flush);
            try interval.sleep(io);
        }
    }
};

/// Max-wins, matching the SQL upsert exactly: the newest failure in the minute
/// is the one whose name the cell keeps.
fn noteFailure(cell: *Accumulator.Cell, at: i64, error_name: []const u8) void {
    if (cell.last_failure_ts) |existing| {
        if (at < existing) return;
    }
    cell.last_failure_ts = at;
    const copied = @min(error_name.len, cell.last_error_buf.len);
    @memcpy(cell.last_error_buf[0..copied], error_name[0..copied]);
    cell.last_error_len = @intCast(copied);
}

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

const querylog_schema = @import("../storage/querylog_schema.zig");

const testing = std.testing;

fn openLog() !db.Db {
    var database = try db.Db.open(":memory:", .{ .mode = .memory });
    errdefer database.close();
    try db.applyPragmas(&database, .{});
    try database.exec(querylog_schema.ddl);
    return database;
}

/// The accumulator is about a megabyte, which is more than a test frame should
/// carry.
fn newAccumulator() !*Accumulator {
    const acc = try testing.allocator.create(Accumulator);
    acc.* = .init;
    return acc;
}

fn failingWrite(_: *db.Db, _: []const upstream_history_repo.FlushRow) db.Error!void {
    return error.Busy;
}

/// What the last stubbed flush was handed, copied out so an assertion can read
/// it after the pass returned.
var recorded: [8]upstream_history_repo.FlushRow = undefined;
var recorded_len: usize = 0;

fn recordingWrite(_: *db.Db, rows: []const upstream_history_repo.FlushRow) db.Error!void {
    recorded_len = @min(rows.len, recorded.len);
    @memcpy(recorded[0..recorded_len], rows[0..recorded_len]);
}

/// The interleaving of ruling 4: a flush is in flight, and the recording side
/// recreates a swapped-out cell and then fills the accumulator to overflow.
var interleaved: ?*Accumulator = null;
var interleave_io: ?std.Io = null;
var interleave_urls: [max_pending][8]u8 = undefined;

fn interleavingWrite(_: *db.Db, _: []const upstream_history_repo.FlushRow) db.Error!void {
    const acc = interleaved.?;
    const io = interleave_io.?;
    // The very `(url, minute_ts)` the flush is holding, recorded again while
    // the write runs.
    acc.recordSuccess(io, "https://a.example", 60);
    // And then enough distinct minutes to fill the accumulator and evict.
    for (&interleave_urls, 0..) |*name, i| {
        const url = std.fmt.bufPrint(name, "u{d:0>6}", .{i}) catch unreachable;
        acc.recordSuccess(io, url, @as(i64, @intCast(i)) * 600 + 6000);
    }
    return error.Busy;
}

test "minuteOf floors to the minute, including before the epoch" {
    try testing.expectEqual(@as(i64, 0), minuteOf(0));
    try testing.expectEqual(@as(i64, 0), minuteOf(59));
    try testing.expectEqual(@as(i64, 60), minuteOf(60));
    try testing.expectEqual(@as(i64, 120), minuteOf(179));
    try testing.expectEqual(@as(i64, -60), minuteOf(-1));
}

test "outcomes land in the cell of their own minute and upstream" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    const acc = try newAccumulator();
    defer testing.allocator.destroy(acc);

    acc.recordSuccess(io, "https://a.example", 65);
    acc.recordSuccess(io, "https://a.example", 119);
    acc.recordFailure(io, "https://a.example", 100, "Timeout");
    acc.recordSuccess(io, "https://a.example", 130);
    acc.recordSuccess(io, "https://b.example", 70);

    // Three cells: a/60, a/120 and b/60.
    try testing.expectEqual(@as(u32, 3), acc.snapshotStats(io).pending);

    const first = acc.cellFor("https://a.example", 60);
    try testing.expectEqual(@as(u32, 2), first.successes);
    try testing.expectEqual(@as(u32, 1), first.failures);
    try testing.expectEqual(@as(?i64, 100), first.last_failure_ts);
    try testing.expectEqualStrings("Timeout", first.lastError());

    const second = acc.cellFor("https://a.example", 120);
    try testing.expectEqual(@as(u32, 1), second.successes);
    try testing.expectEqual(@as(u32, 0), second.failures);
    try testing.expectEqual(@as(?i64, null), second.last_failure_ts);
}

test "a cell keeps the newest failure's error and ignores an older one" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    const acc = try newAccumulator();
    defer testing.allocator.destroy(acc);

    acc.recordFailure(io, "https://a.example", 100, "Timeout");
    acc.recordFailure(io, "https://a.example", 80, "ConnectFailed");
    const cell = acc.cellFor("https://a.example", 60);
    try testing.expectEqual(@as(u32, 2), cell.failures);
    try testing.expectEqual(@as(?i64, 100), cell.last_failure_ts);
    try testing.expectEqualStrings("Timeout", cell.lastError());

    acc.recordFailure(io, "https://a.example", 110, "BadResponse");
    try testing.expectEqual(@as(?i64, 110), cell.last_failure_ts);
    try testing.expectEqualStrings("BadResponse", cell.lastError());

    const long = "A" ** 200;
    acc.recordFailure(io, "https://a.example", 115, long);
    try testing.expectEqual(@as(usize, health.error_name_capacity), cell.lastError().len);
}

test "at capacity the oldest minute is dropped, counted, and the watermark only moves forward" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    const acc = try newAccumulator();
    defer testing.allocator.destroy(acc);

    var names: [max_pending][8]u8 = undefined;
    for (&names, 0..) |*name, i| {
        const url = std.fmt.bufPrint(name, "u{d:0>6}", .{i}) catch unreachable;
        // Minute 600 is the oldest; every later cell is newer.
        acc.recordSuccess(io, url, 600 + @as(i64, @intCast(i)) * 60);
    }
    try testing.expectEqual(@as(u32, max_pending), acc.snapshotStats(io).pending);
    try testing.expectEqual(@as(u64, 0), acc.snapshotStats(io).rows_dropped);
    try testing.expectEqual(@as(?i64, null), acc.snapshotStats(io).last_drop_minute);

    // One more cell evicts the oldest minute and nothing else.
    acc.recordSuccess(io, "https://new.example", 10_000_000);
    const after = acc.snapshotStats(io);
    try testing.expectEqual(@as(u32, max_pending), after.pending);
    try testing.expectEqual(@as(u64, 1), after.rows_dropped);
    try testing.expectEqual(@as(?i64, 600), after.last_drop_minute);

    // A later eviction of an *older* minute must not move the watermark back.
    acc.recordSuccess(io, "https://older.example", 120);
    const back = acc.snapshotStats(io);
    try testing.expectEqual(@as(u64, 2), back.rows_dropped);
    try testing.expectEqual(@as(?i64, 660), back.last_drop_minute);

    acc.recordSuccess(io, "https://newer.example", 20_000_000);
    const forward = acc.snapshotStats(io);
    try testing.expectEqual(@as(u64, 3), forward.rows_dropped);
    // The minute just evicted is 120, older than the 660 already recorded.
    try testing.expectEqual(@as(?i64, 660), forward.last_drop_minute);
}

test "a successful flush hands over every cell, empties the accumulator and counts once" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

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

    const acc = try newAccumulator();
    defer testing.allocator.destroy(acc);

    acc.recordSuccess(io, "https://a.example", 65);
    acc.recordFailure(io, "https://a.example", 100, "Timeout");
    acc.flushOnce(io, &database, recordingWrite);

    try testing.expectEqual(@as(usize, 1), recorded_len);
    try testing.expectEqualStrings("https://a.example", recorded[0].url);
    try testing.expectEqual(@as(i64, 60), recorded[0].minute_ts);
    try testing.expectEqual(@as(u32, 1), recorded[0].successes);
    try testing.expectEqual(@as(u32, 1), recorded[0].failures);
    try testing.expectEqual(@as(?i64, 100), recorded[0].last_failure_ts);
    try testing.expectEqualStrings("Timeout", recorded[0].last_error);

    const stats = acc.snapshotStats(io);
    try testing.expectEqual(@as(u32, 0), stats.pending);
    try testing.expectEqual(@as(u64, 1), stats.flushes);
    try testing.expectEqual(@as(u64, 0), stats.flush_failures);
    try testing.expect(!stats.last_flush_failed);
}

test "a pass with nothing pending counts neither a flush nor a failure" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

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

    const acc = try newAccumulator();
    defer testing.allocator.destroy(acc);

    acc.flushOnce(io, &database, failingWrite);
    acc.flushOnce(io, &database, recordingWrite);

    const stats = acc.snapshotStats(io);
    try testing.expectEqual(@as(u64, 0), stats.flushes);
    try testing.expectEqual(@as(u64, 0), stats.flush_failures);
    try testing.expect(!stats.last_flush_failed);
}

test "a failed flush keeps the rows, sets the flag, and the next success clears it" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

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

    const acc = try newAccumulator();
    defer testing.allocator.destroy(acc);

    acc.recordSuccess(io, "https://a.example", 65);
    acc.recordFailure(io, "https://a.example", 100, "Timeout");
    acc.flushOnce(io, &database, failingWrite);

    const failed = acc.snapshotStats(io);
    try testing.expectEqual(@as(u64, 0), failed.flushes);
    try testing.expectEqual(@as(u64, 1), failed.flush_failures);
    try testing.expectEqual(@as(u64, 0), failed.rows_dropped);
    try testing.expect(failed.last_flush_failed);
    // Nothing was lost: the cell is back, whole.
    try testing.expectEqual(@as(u32, 1), failed.pending);
    const cell = acc.cellFor("https://a.example", 60);
    try testing.expectEqual(@as(u32, 1), cell.successes);
    try testing.expectEqual(@as(u32, 1), cell.failures);
    try testing.expectEqualStrings("Timeout", cell.lastError());

    // The retry writes what the failed pass could not.
    acc.recordSuccess(io, "https://a.example", 70);
    acc.flushOnce(io, &database, recordingWrite);
    const cleared = acc.snapshotStats(io);
    try testing.expectEqual(@as(u64, 1), cleared.flushes);
    try testing.expectEqual(@as(u64, 1), cleared.flush_failures);
    try testing.expect(!cleared.last_flush_failed);
    try testing.expectEqual(@as(usize, 1), recorded_len);
    try testing.expectEqual(@as(u32, 2), recorded[0].successes);
    try testing.expectEqual(@as(u32, 1), recorded[0].failures);
}

test "a merge-back sums into what was recorded beside the flush" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

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

    const acc = try newAccumulator();
    defer testing.allocator.destroy(acc);

    // The write recreates the swapped-out cell before it fails, which is the
    // interleaving the swap makes possible.
    interleaved = acc;
    interleave_io = io;
    defer {
        interleaved = null;
        interleave_io = null;
    }

    acc.recordSuccess(io, "https://a.example", 65);
    acc.recordFailure(io, "https://a.example", 100, "Timeout");
    acc.flushOnce(io, &database, interleavingWrite);

    const stats = acc.snapshotStats(io);
    try testing.expect(stats.last_flush_failed);
    try testing.expectEqual(@as(u64, 1), stats.flush_failures);
    // Two drops, and they are different drops: filling to capacity during the
    // write evicted the recreated minute 60, and the merge-back then found no
    // room either and evicted the oldest of what the write had left, 6000.
    // Overflow loss is the specced policy; what matters is that it is counted.
    try testing.expectEqual(@as(u32, max_pending), stats.pending);
    try testing.expectEqual(@as(u64, 2), stats.rows_dropped);
    // Forward only: the second eviction was the newer minute of the two.
    try testing.expectEqual(@as(?i64, 6000), stats.last_drop_minute);
    // The merged cell is back, carrying what the failed flush was holding.
    const merged = acc.cellFor("https://a.example", 60);
    try testing.expectEqual(@as(u32, 1), merged.successes);
    try testing.expectEqual(@as(u32, 1), merged.failures);
    try testing.expectEqualStrings("Timeout", merged.lastError());

    // The flush-owned buffer was not touched by any of the recording that
    // happened beside it: it still holds exactly what was swapped out.
    try testing.expectEqual(@as(u32, 1), acc.flush_count);
    try testing.expectEqual(@as(u32, 1), acc.flush_cells[0].successes);
    try testing.expectEqual(@as(u32, 1), acc.flush_cells[0].failures);
    try testing.expectEqual(@as(i64, 60), acc.flush_cells[0].minute_ts);
}

test "a flush against the real repository writes the minute rows" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

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

    const acc = try newAccumulator();
    defer testing.allocator.destroy(acc);

    acc.recordSuccess(io, "https://a.example", 65);
    acc.recordFailure(io, "https://a.example", 100, "Timeout");
    acc.recordSuccess(io, "https://b.example", 130);
    acc.flushOnce(io, &database, upstream_history_repo.flush);

    // The second pass is the restart shape: the same minute, added to.
    acc.recordSuccess(io, "https://a.example", 90);
    acc.flushOnce(io, &database, upstream_history_repo.flush);

    const stats = try upstream_history_repo.windowStats(&database, "https://a.example", 0, 200);
    try testing.expectEqual(@as(u64, 2), stats.successes);
    try testing.expectEqual(@as(u64, 1), stats.failures);
    try testing.expectEqual(@as(u64, 3), stats.attempts);
    try testing.expectEqual(@as(?i64, 100), stats.last_failure_ts);
    try testing.expectEqualStrings("Timeout", stats.lastFailureError());
    try testing.expectEqual(@as(i64, 2), try upstream_history_repo.countMinutes(&database));
    try testing.expectEqual(@as(u64, 2), acc.snapshotStats(io).flushes);
}