milestone 30: overview as a dashboard, explicit health contract, period aggregations

src/storage/logger.zig

@@ -30,6 +30,7 @@
 //! `writer_failed`, so the loss is visible rather than silent.
 
 const std = @import("std");
+const builtin = @import("builtin");
 
 const db = @import("db.zig");
 const disk_monitor = @import("disk_monitor.zig");
@@ -335,12 +336,85 @@ const Outcome = union(enum) {
     expiry: std.Io.Cancelable!void,
 };
 
+/// Where the disk gate stands right now, as a state rather than a tally.
+///
+/// `open` is the steady state. `gated` says the gate is holding writes back but
+/// nothing has been lost to it yet, and `losing` says this episode has already
+/// cost rows. A cumulative drop counter cannot say any of that: it only ever
+/// grows, so a rollup computed from it would latch on the first overflow.
+pub const GateEpisode = enum(u8) { open, gated, losing };
+
+/// The episode state and the identity of the episode it belongs to, in one
+/// word so a compare-and-swap can test both at once.
+///
+/// The identity is what the state alone cannot carry. A producer decides a row
+/// is lost, stalls, and wakes after the gate has closed, reopened and closed
+/// again; a bare `gated -> losing` swap would then mark an episode that has
+/// cost nothing. `generation` rises every time an episode opens, so that swap
+/// fails against the newer episode and the stale loss is discarded.
+const Gate = packed struct(u64) {
+    episode: GateEpisode,
+    generation: u56,
+
+    const initial: Gate = .{ .episode = .open, .generation = 0 };
+
+    fn bits(self: Gate) u64 {
+        return @bitCast(self);
+    }
+
+    fn of(bits_value: u64) Gate {
+        return @bitCast(bits_value);
+    }
+};
+
+/// Parks a producer immediately before it evicts an entry, so a test can move
+/// the disk gate while that producer is still inside `enqueue` and the row it
+/// is about to lose is still queued.
+///
+/// Nothing else reaches that point. `enqueue` never suspends, so from outside a
+/// gate read before its loop and one read at the eviction give the same answer,
+/// and no test could tell the two apart — which is exactly the difference that
+/// decides whether a discard can join an episode that opened after the producer
+/// started.
+///
+/// Before the eviction and not after it: parking after the row is already gone
+/// would let a test open an episode and then file an earlier loss under it,
+/// which is the misattribution the sampling rule exists to prevent. The loss
+/// has to happen inside the episode for the episode to own it. The storage
+/// exists in a test build only, and `park` reduces to nothing everywhere else —
+/// the settings hash seam's shape (`web/handlers/settings.zig`).
+const discard_stall = if (builtin.is_test) struct {
+    var armed: bool = false;
+    var parked: std.Io.Event = .unset;
+    var release: std.Io.Event = .unset;
+
+    fn park(io: std.Io) void {
+        if (!armed) return;
+        parked.set(io);
+        release.waitUncancelable(io);
+    }
+} else struct {
+    fn park(io: std.Io) void {
+        _ = io;
+    }
+};
+
 pub const Logger = struct {
     cfg: model.Logging,
     queue: EntryQueue,
     queries_dropped: std.atomic.Value(u64),
+    /// When the newest drop happened, in unix seconds; 0 means none yet. Read
+    /// through `lastDropSeconds`, which is what turns the sentinel into a null.
+    last_drop_s: std.atomic.Value(i64),
     rows_written: std.atomic.Value(u64),
     batches_gated: std.atomic.Value(u64),
+    /// The gating episode `/api/health` reports as `query_history.losing`, as
+    /// `Gate` bits. Moved only by `gateHolds`/`gateReopened`, which the writer
+    /// calls as it observes the monitor, and raised to `losing` by a drop that
+    /// carries the identity of the episode still holding.
+    ///
+    /// Read it through `gateEpisode` or `sampleGate`, never as a raw integer.
+    gate: std.atomic.Value(u64),
     /// Set when `runWriter` gives up before it consumed anything. The queue is
     /// closed and every entry counts as dropped from that point, so a caller
     /// that sees this must not expect rows.
@@ -363,8 +437,10 @@ pub const Logger = struct {
             .cfg = cfg,
             .queue = .init(queue_buf),
             .queries_dropped = .init(0),
+            .last_drop_s = .init(0),
             .rows_written = .init(0),
             .batches_gated = .init(0),
+            .gate = .init(Gate.initial.bits()),
             .writer_failed = .init(false),
             .draining = .init(false),
         };
@@ -422,10 +498,17 @@ pub const Logger = struct {
             if (self.queue.capacity() == 0) break;
 
             var oldest: [1]Entry = undefined;
+            discard_stall.park(io);
             const got = self.queue.get(io, &oldest, 0) catch break;
-            if (got == 1) self.countDropped(1);
+            // Sampled after the eviction and not before the loop: the row is
+            // lost on the line above, and a sample taken while the gate was
+            // still open would let an episode that opened during `put` escape
+            // being marked for a row it really cost. Reading it here cannot
+            // misattribute in the other direction either — a sample the gate
+            // outruns fails `countDropped`'s generation check.
+            if (got == 1) self.countDropped(io, 1, self.sampleGate());
         }
-        self.countDropped(1);
+        self.countDropped(io, 1, self.sampleGate());
     }
 
     /// The writer task: owns `database` and its prepared statements for its
@@ -453,7 +536,7 @@ pub const Logger = struct {
             // life of the process, so the row stays active, which is the truth.
             self.reportWrite(io, "writer", "preparing the batch statements failed", @errorName(err), 0);
             self.queue.close(io);
-            _ = self.dropRemaining(io);
+            _ = self.dropRemaining(io, self.sampleGate());
             return;
         };
         defer writer.deinit();
@@ -466,17 +549,24 @@ pub const Logger = struct {
                 error.Closed => return,
                 error.Canceled => |e| return e,
             };
+            // Before `fill`, not only inside `flush`: `fill` spends the whole
+            // flush interval taking entries off the queue, and a producer that
+            // overflows the queue during that wait is losing rows to the gate
+            // just as surely as the held batch is. Observing here is what makes
+            // the episode start cover those drops instead of misfiling them as
+            // ordinary overflow.
+            const at = self.observeGate(monitor);
             const deadline = self.flushDeadline(io);
             // `n` is live across both calls: entries already taken off the
             // queue are lost if either one is canceled, so they must count.
             var n: usize = 1;
-            self.fill(io, &batch, deadline, &n) catch |err| {
-                self.countDropped(n);
+            self.fill(io, &batch, deadline, &n, at) catch |err| {
+                self.countDropped(io, n, at);
                 return err;
             };
             self.flush(io, &writer, batch[0..n], monitor) catch |err| switch (err) {
                 error.Canceled => |e| {
-                    self.countDropped(n);
+                    self.countDropped(io, n, at);
                     return e;
                 },
                 // Nothing will open the gate now. Everything still queued is
@@ -484,8 +574,12 @@ pub const Logger = struct {
                 // announced once — a per-chunk report would write to the very
                 // disk that is out of space, dozens of times, on the way out.
                 error.GatedAtShutdown => {
-                    self.countDropped(n);
-                    const lost = n + self.dropRemaining(io);
+                    // Re-sampled rather than reusing `at`: `flush` observed the
+                    // gate again on its way to this error, so the episode that
+                    // is costing these rows is the one holding now.
+                    const gated_at = self.sampleGate();
+                    self.countDropped(io, n, gated_at);
+                    const lost = n + self.dropRemaining(io, gated_at);
                     scope.warn(
                         "query log: {d} rows dropped at shutdown, the disk gate was closed",
                         .{lost},
@@ -521,7 +615,7 @@ pub const Logger = struct {
     /// many. The drain is uncancelable: a cancellation racing the writer's own
     /// failure would otherwise abandon the buffered entries without counting
     /// them.
-    fn dropRemaining(self: *Logger, io: std.Io) usize {
+    fn dropRemaining(self: *Logger, io: std.Io, at: Gate) usize {
         var total: usize = 0;
         var leftover: [flush_batch]Entry = undefined;
         while (true) {
@@ -529,7 +623,7 @@ pub const Logger = struct {
                 error.Closed => break,
             };
             if (n == 0) break;
-            self.countDropped(n);
+            self.countDropped(io, n, at);
             total += n;
         }
         return total;
@@ -559,13 +653,14 @@ pub const Logger = struct {
         batch: *[flush_batch]Entry,
         deadline: std.Io.Clock.Timestamp,
         n: *usize,
+        at: Gate,
     ) std.Io.Cancelable!void {
         n.* += self.drainAvailable(io, batch[n.*..]);
 
         while (n.* < batch.len) {
             const remaining = deadline.durationFromNow(io);
             if (remaining.raw.nanoseconds <= 0) break;
-            const entry = try self.getWithin(io, remaining) orelse break;
+            const entry = try self.getWithin(io, remaining, at) orelse break;
             batch[n.*] = entry;
             n.* += 1;
             n.* += self.drainAvailable(io, batch[n.*..]);
@@ -588,6 +683,7 @@ pub const Logger = struct {
         self: *Logger,
         io: std.Io,
         budget: std.Io.Clock.Duration,
+        at: Gate,
     ) std.Io.Cancelable!?Entry {
         var outcomes: [2]Outcome = undefined;
         var race: std.Io.Select(Outcome) = .init(io, &outcomes);
@@ -603,7 +699,7 @@ pub const Logger = struct {
 
         const first = race.await() catch |err| {
             // Teardown: the entry the getter already took has nowhere to go.
-            if (drainRace(&race)) |_| self.countDropped(1);
+            if (drainRace(&race)) |_| self.countDropped(io, 1, at);
             return err;
         };
         const late = drainRace(&race);
@@ -628,6 +724,7 @@ pub const Logger = struct {
                 .clock = .awake,
             };
             while (!m.writesAllowed()) {
+                self.gateHolds();
                 // Waiting for the disk to recover is right while the process
                 // runs and wrong once it is stopping: nothing is going to free
                 // space during shutdown, so the batch is lost either way and
@@ -638,6 +735,7 @@ pub const Logger = struct {
                 _ = self.batches_gated.fetchAdd(1, .monotonic);
                 try pause.sleep(io);
             }
+            self.gateReopened();
         }
 
         var rows: [flush_batch]queries_repo.Row = undefined;
@@ -645,7 +743,10 @@ pub const Logger = struct {
 
         writer.writeBatch(rows[0..entries.len]) catch |err| {
             scope.warn("query log batch of {d} rows dropped: {s}", .{ entries.len, @errorName(err) });
-            self.countDropped(entries.len);
+            // Sampled here, so a batch the gate already let through is counted
+            // against whatever episode is open now — a database failure is not
+            // a gating loss.
+            self.countDropped(io, entries.len, self.sampleGate());
             self.reportWrite(io, "batch", "a query log batch was dropped", @errorName(err), entries.len);
             return;
         };
@@ -683,8 +784,92 @@ pub const Logger = struct {
         );
     }
 
-    fn countDropped(self: *Logger, n: usize) void {
+    /// Counts `n` lost rows, stamps the loss, and raises the episode `at` to
+    /// `losing` if that episode is still the current one.
+    ///
+    /// `at` is sampled where the rows were lost, not here: see `Gate`. A caller
+    /// that lost rows outside any episode passes what `sampleGate` gave it and
+    /// the raise is simply a no-op.
+    ///
+    /// The stamp is a second atomic rather than a field beside the count, so a
+    /// reader can briefly see the new total against the previous timestamp. The
+    /// health contract says so: `dropped_total` above zero with a null
+    /// `last_drop_s` is a legal, momentary answer.
+    fn countDropped(self: *Logger, io: std.Io, n: usize, at: Gate) void {
         _ = self.queries_dropped.fetchAdd(n, .monotonic);
+        self.stampDrop(std.Io.Clock.real.now(io).toSeconds());
+        if (at.episode != .gated) return;
+        const losing: Gate = .{ .episode = .losing, .generation = at.generation };
+        _ = self.gate.cmpxchgStrong(at.bits(), losing.bits(), .acq_rel, .monotonic);
+    }
+
+    /// Moves the stamp forward only. Two producers can reach `countDropped` out
+    /// of order, and a plain store would let the older one publish its
+    /// timestamp over the newer drop's — a `last_drop_s` that walks backwards
+    /// while drops are still arriving.
+    fn stampDrop(self: *Logger, at_s: i64) void {
+        var seen = self.last_drop_s.load(.monotonic);
+        while (at_s > seen) {
+            seen = self.last_drop_s.cmpxchgWeak(seen, at_s, .monotonic, .monotonic) orelse return;
+        }
+    }
+
+    /// When the newest drop happened, or null while nothing has been dropped.
+    pub fn lastDropSeconds(self: *const Logger) ?i64 {
+        const stamped = self.last_drop_s.load(.monotonic);
+        return if (stamped == 0) null else stamped;
+    }
+
+    /// Where the gate stands right now, for a reader that only wants the state.
+    pub fn gateEpisode(self: *const Logger) GateEpisode {
+        return Gate.of(self.gate.load(.acquire)).episode;
+    }
+
+    /// The identity a caller must carry with rows it loses.
+    ///
+    /// Take it where the loss actually happens. A producer discarding one entry
+    /// samples at the discard; the writer samples once at `observeGate`,
+    /// because the batch it is holding belongs to the episode that was open
+    /// while it filled. Sampling earlier than the loss hides episodes that
+    /// opened in between; sampling later cannot misattribute, because the
+    /// generation makes an outrun sample fail its swap.
+    fn sampleGate(self: *const Logger) Gate {
+        return Gate.of(self.gate.load(.acquire));
+    }
+
+    /// Opens a gating episode under a fresh generation, or leaves one that is
+    /// already holding alone — `losing` must not fall back to `gated`, and a
+    /// second observation of the same gate must not look like a new episode.
+    fn gateHolds(self: *Logger) void {
+        var current = self.sampleGate();
+        while (current.episode == .open) {
+            const next: Gate = .{ .episode = .gated, .generation = current.generation +% 1 };
+            const raced = self.gate.cmpxchgWeak(current.bits(), next.bits(), .acq_rel, .acquire) orelse
+                return;
+            current = Gate.of(raced);
+        }
+    }
+
+    /// Ends the episode, keeping its generation so the next one gets a number
+    /// no stale producer holds. A drop sampled during the episode that lands
+    /// after this reopen finds its generation gone and is discarded.
+    fn gateReopened(self: *Logger) void {
+        var current = self.sampleGate();
+        while (current.episode != .open) {
+            const next: Gate = .{ .episode = .open, .generation = current.generation };
+            const raced = self.gate.cmpxchgWeak(current.bits(), next.bits(), .acq_rel, .acquire) orelse
+                return;
+            current = Gate.of(raced);
+        }
+    }
+
+    /// Moves the episode to wherever the monitor says the gate is, and returns
+    /// the episode the caller's rows now belong to. A null monitor is no gating
+    /// at all, so the episode stays `open`.
+    fn observeGate(self: *Logger, monitor: ?*disk_monitor.Monitor) Gate {
+        const m = monitor orelse return self.sampleGate();
+        if (m.writesAllowed()) self.gateReopened() else self.gateHolds();
+        return self.sampleGate();
     }
 };
 
@@ -1189,7 +1374,7 @@ test "entries that arrive inside one window reach the database in one batch" {
     var batch: [flush_batch]Entry = undefined;
     batch[0] = try logger.queue.getOne(io);
     var n: usize = 1;
-    try logger.fill(io, &batch, logger.flushDeadline(io), &n);
+    try logger.fill(io, &batch, logger.flushDeadline(io), &n, logger.sampleGate());
     try testing.expectEqual(@as(usize, 6), n);
 
     // One `writeBatch` call, which is one transaction (`queries_repo.zig`).
@@ -1214,7 +1399,7 @@ test "a zero interval takes what is queued and waits for nothing" {
     var n: usize = 1;
     // The queue is open and the batch has room: any non-zero interval blocks
     // here until it expires. Zero returns with what was already queued.
-    try logger.fill(io, &batch, logger.flushDeadline(io), &n);
+    try logger.fill(io, &batch, logger.flushDeadline(io), &n, logger.sampleGate());
     try testing.expectEqual(@as(usize, 2), n);
 }
 
@@ -1404,6 +1589,189 @@ test "a writer that cannot prepare closes the queue and counts every entry" {
     try testing.expectEqual(@as(u64, 4), logger.queries_dropped.load(.monotonic));
 }
 
+test "a queue overflow stamps the drop and leaves the gate episode open" {
+    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    // No writer and no monitor: nothing consumes the queue, so the third entry
+    // has to displace the oldest, and no gate is involved in the loss.
+    var buf: [2]Entry = undefined;
+    var logger: Logger = .init(.{}, &buf);
+
+    try testing.expectEqual(@as(?i64, null), logger.lastDropSeconds());
+
+    for (0..3) |i| logger.log(io, sampleEntry(@intCast(i), "overflow.example"));
+
+    try testing.expectEqual(@as(u64, 1), logger.queries_dropped.load(.monotonic));
+    const stamped = logger.lastDropSeconds() orelse return error.TestUnexpectedResult;
+    try testing.expect(stamped > 1_700_000_000);
+    // Cumulative loss is not a current fault: the episode never opened.
+    try testing.expectEqual(GateEpisode.open, logger.gateEpisode());
+}
+
+test "the gating episode opens on the gate, turns losing on a drop, and clears on recovery" {
+    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    var database = try openLog();
+    defer database.close();
+
+    // Small enough that the entries logged below have to displace each other,
+    // which is the gate-caused overflow this episode is meant to catch.
+    var buf: [4]Entry = undefined;
+    var logger: Logger = .init(.{ .query_log_flush_interval_s = 0 }, &buf);
+
+    var monitor: disk_monitor.Monitor = .init(.{}, std.Io.Dir.cwd(), ".", null);
+    monitor.state_raw.store(@intFromEnum(disk_monitor.State.critical), .monotonic);
+
+    logger.log(io, sampleEntry(1, "held.example"));
+
+    var future = try io.concurrent(Logger.runWriter, .{
+        &logger,
+        io,
+        &database,
+        @as(?*disk_monitor.Monitor, &monitor),
+    });
+
+    const poll: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(5), .clock = .awake };
+    var waited: usize = 0;
+    while (logger.batches_gated.load(.monotonic) == 0) : (waited += 1) {
+        try testing.expect(waited < 400);
+        try poll.sleep(io);
+    }
+
+    // The gate is holding and nothing has been lost yet, which is not a fault:
+    // the batch is still going to be written if the disk recovers.
+    try testing.expectEqual(GateEpisode.gated, logger.gateEpisode());
+    try testing.expectEqual(@as(u64, 0), logger.queries_dropped.load(.monotonic));
+
+    // Now overflow the queue behind the held batch. These drops belong to the
+    // episode, and that is what turns it from held to losing.
+    for (0..32) |i| logger.log(io, sampleEntry(@intCast(i + 2), "queued.example"));
+    try testing.expect(logger.queries_dropped.load(.monotonic) > 0);
+    try testing.expectEqual(GateEpisode.losing, logger.gateEpisode());
+    try testing.expect(logger.lastDropSeconds() != null);
+
+    // The disk recovers: the held batch goes out and the episode ends.
+    monitor.state_raw.store(@intFromEnum(disk_monitor.State.ok), .monotonic);
+    waited = 0;
+    while (logger.gateEpisode() != .open) : (waited += 1) {
+        try testing.expect(waited < 400);
+        try poll.sleep(io);
+    }
+    try testing.expect(logger.rows_written.load(.monotonic) > 0);
+    // The count keeps the history the state does not.
+    try testing.expect(logger.queries_dropped.load(.monotonic) > 0);
+
+    logger.shutdown(io);
+    try future.await(io);
+}
+
+// The two tests below drive the gate primitives directly. A threaded test
+// cannot prove the absence of the race they close — it can only fail to hit it
+// — so they pin the mechanism instead: the generation a drop must match, and
+// the direction the stamp may move.
+
+test "a drop sampled in one episode cannot mark the next one losing" {
+    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    var queue_buf: [2]Entry = undefined;
+    var logger: Logger = .init(.{}, &queue_buf);
+
+    logger.gateHolds();
+    // What a producer holds while it stalls: episode one, still holding.
+    const stalled_in = logger.sampleGate();
+    try testing.expectEqual(GateEpisode.gated, stalled_in.episode);
+
+    // The gate opens and closes again while that producer is descheduled.
+    logger.gateReopened();
+    logger.gateHolds();
+    const episode_two = logger.sampleGate();
+    try testing.expectEqual(GateEpisode.gated, episode_two.episode);
+    try testing.expect(episode_two.generation != stalled_in.generation);
+
+    // The stale drop still counts as a lost row — it was one — but episode two
+    // has cost nothing and must not be told it has.
+    logger.countDropped(io, 1, stalled_in);
+    try testing.expectEqual(@as(u64, 1), logger.queries_dropped.load(.monotonic));
+    try testing.expectEqual(GateEpisode.gated, logger.gateEpisode());
+
+    // A drop that really belongs to episode two does raise it.
+    logger.countDropped(io, 1, episode_two);
+    try testing.expectEqual(GateEpisode.losing, logger.gateEpisode());
+
+    // And a second hold of a gate that never opened is the same episode, not a
+    // new one: `losing` must not fall back to `gated`.
+    logger.gateHolds();
+    try testing.expectEqual(GateEpisode.losing, logger.gateEpisode());
+    try testing.expectEqual(episode_two.generation, logger.sampleGate().generation);
+}
+
+fn logOne(logger: *Logger, io: std.Io) void {
+    logger.log(io, sampleEntry(2, "second.example"));
+}
+
+test "a gate that closes mid-enqueue still gets the discard that follows it" {
+    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    // Capacity one and no writer, so the second entry has to evict the first
+    // and the producer discards inside `enqueue`.
+    var queue_buf: [1]Entry = undefined;
+    var logger: Logger = .init(.{}, &queue_buf);
+
+    logger.log(io, sampleEntry(1, "first.example"));
+    try testing.expectEqual(GateEpisode.open, logger.gateEpisode());
+
+    discard_stall.parked = .unset;
+    discard_stall.release = .unset;
+    discard_stall.armed = true;
+    defer discard_stall.armed = false;
+
+    // The producer enters `enqueue` with the gate open — which is the reading a
+    // sample taken before the loop would keep for the rest of the call.
+    var producer = try io.concurrent(logOne, .{ &logger, io });
+    discard_stall.parked.waitUncancelable(io);
+
+    // The producer is parked with the row it will evict still on the queue, so
+    // the episode opens strictly before the loss rather than after it. Nothing
+    // has been dropped yet, and that is what makes the row this episode's.
+    try testing.expectEqual(@as(u64, 0), logger.queries_dropped.load(.monotonic));
+    logger.gateHolds();
+    discard_stall.armed = false;
+    discard_stall.release.set(io);
+    producer.await(io);
+
+    try testing.expectEqual(@as(u64, 1), logger.queries_dropped.load(.monotonic));
+    // The assertion the pre-loop sample fails: it would still be holding
+    // `open`, `countDropped` would return before its swap, and the episode
+    // would sit at `gated` having silently cost a row.
+    try testing.expectEqual(GateEpisode.losing, logger.gateEpisode());
+}
+
+test "the drop stamp only ever moves forward" {
+    var queue_buf: [2]Entry = undefined;
+    var logger: Logger = .init(.{}, &queue_buf);
+
+    try testing.expectEqual(@as(?i64, null), logger.lastDropSeconds());
+
+    logger.stampDrop(1_700_000_100);
+    try testing.expectEqual(@as(?i64, 1_700_000_100), logger.lastDropSeconds());
+
+    // A producer that read its clock earlier but arrives later: the newer drop
+    // already published its time and must keep it.
+    logger.stampDrop(1_700_000_050);
+    try testing.expectEqual(@as(?i64, 1_700_000_100), logger.lastDropSeconds());
+
+    logger.stampDrop(1_700_000_200);
+    try testing.expectEqual(@as(?i64, 1_700_000_200), logger.lastDropSeconds());
+}
+
 test "a canceled writer counts the batch it was holding" {
     var threaded: std.Io.Threaded = .init(testing.allocator, .{});
     defer threaded.deinit();