metrics: peak gauges for in-flight queries, stream connections and the query log queue

src/storage/logger.zig

@@ -320,7 +320,68 @@ fn emptyAsNull(value: []const u8) ?[]const u8 {
     return if (value.len == 0) null else value;
 }
 
-const EntryQueue = std.Io.Queue(Entry);
+/// `std.Io.Queue` plus the occupancy high-water that sizes
+/// `logging.query_log_buffer_max` (specs/memory-budget.md item 0).
+///
+/// The queue keeps its length behind its own mutex with no accessor, so the
+/// count lives beside it. Raised before an insert and lowered after a removal,
+/// it never reads below the true occupancy; it overstates it by at most the
+/// producers inside `tryPut` plus the entries a getter has taken and not yet
+/// subtracted. An upper bound is the safe side for sizing a buffer.
+const EntryQueue = struct {
+    inner: std.Io.Queue(Entry),
+    held: std.atomic.Value(u64) = .init(0),
+    peak: std.atomic.Value(u64) = .init(0),
+
+    pub fn init(buffer: []Entry) EntryQueue {
+        return .{ .inner = .init(buffer) };
+    }
+
+    pub fn capacity(q: *const EntryQueue) usize {
+        return q.inner.capacity();
+    }
+
+    pub fn close(q: *EntryQueue, io: std.Io) void {
+        q.inner.close(io);
+    }
+
+    pub fn peakOccupancy(q: *const EntryQueue) u64 {
+        return q.peak.load(.monotonic);
+    }
+
+    /// Takes `entry` if there is room or a waiting getter, without blocking.
+    pub fn tryPut(q: *EntryQueue, io: std.Io, entry: Entry) (std.Io.QueueClosedError || std.Io.Cancelable)!bool {
+        const held = q.held.fetchAdd(1, .monotonic) + 1;
+        const put = q.inner.put(io, &.{entry}, 0) catch |err| {
+            _ = q.held.fetchSub(1, .monotonic);
+            return err;
+        };
+        if (put == 0) {
+            _ = q.held.fetchSub(1, .monotonic);
+            return false;
+        }
+        _ = q.peak.fetchMax(held, .monotonic);
+        return true;
+    }
+
+    pub fn get(q: *EntryQueue, io: std.Io, buffer: []Entry, min: usize) (std.Io.QueueClosedError || std.Io.Cancelable)!usize {
+        const n = try q.inner.get(io, buffer, min);
+        _ = q.held.fetchSub(n, .monotonic);
+        return n;
+    }
+
+    pub fn getUncancelable(q: *EntryQueue, io: std.Io, buffer: []Entry, min: usize) std.Io.QueueClosedError!usize {
+        const n = try q.inner.getUncancelable(io, buffer, min);
+        _ = q.held.fetchSub(n, .monotonic);
+        return n;
+    }
+
+    pub fn getOne(q: *EntryQueue, io: std.Io) (std.Io.QueueClosedError || std.Io.Cancelable)!Entry {
+        const entry = try q.inner.getOne(io);
+        _ = q.held.fetchSub(1, .monotonic);
+        return entry;
+    }
+};
 
 /// A database failure is not in here: a batch the database refuses is dropped,
 /// counted and reported where it happens, and the writer carries on.
@@ -525,8 +586,7 @@ pub const Logger = struct {
         while (true) {
             // A closed queue or a canceled task means shutdown is underway;
             // both leave this entry unwritten, which is what the counter says.
-            const put = self.queue.put(io, &.{entry}, 0) catch break;
-            if (put == 1) return;
+            if (self.queue.tryPut(io, entry) catch break) return;
 
             // A zero-capacity queue holds nothing to drop: the put above was
             // this entry's one chance at a waiting getter.
@@ -1442,6 +1502,34 @@ test "a zero-capacity queue drops every entry exactly once" {
     try testing.expectEqual(@as(u64, 5), logger.queries_dropped.load(.monotonic));
 }
 
+test "the queue peak is the most entries held at once, never the arrivals" {
+    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    var buf: [4]Entry = undefined;
+    var logger: Logger = .init(.{}, &buf);
+    try testing.expectEqual(@as(u64, 0), logger.queue.peakOccupancy());
+
+    logger.log(io, sampleEntry(1, "one.example"));
+    logger.log(io, sampleEntry(2, "two.example"));
+    logger.log(io, sampleEntry(3, "three.example"));
+    _ = try logger.queue.getOne(io);
+    _ = try logger.queue.getOne(io);
+    logger.log(io, sampleEntry(4, "four.example"));
+    try testing.expectEqual(@as(u64, 3), logger.queue.peakOccupancy());
+
+    // A full queue evicts to make room, so the peak stops at the capacity.
+    for (5..10) |i| logger.log(io, sampleEntry(@intCast(i), "more.example"));
+    try testing.expectEqual(@as(u64, 4), logger.queue.peakOccupancy());
+    try testing.expectEqual(@as(u64, 4), logger.queue.held.load(.monotonic));
+
+    var drained: [4]Entry = undefined;
+    try testing.expectEqual(@as(usize, 4), try logger.queue.get(io, &drained, 0));
+    try testing.expectEqual(@as(u64, 0), logger.queue.held.load(.monotonic));
+    try testing.expectEqual(@as(u64, 4), logger.queue.peakOccupancy());
+}
+
 test "log after shutdown drops instead of blocking" {
     var threaded: std.Io.Threaded = .init(testing.allocator, .{});
     defer threaded.deinit();

src/storage/logger_controller.zig

@@ -63,12 +63,14 @@ const Totals = struct {
     rows_written: u64 = 0,
     batches_gated: u64 = 0,
     last_drop_s: i64 = 0,
+    queue_peak: u64 = 0,
 
     fn fold(self: *Totals, lg: *const logger.Logger) void {
         self.queries_dropped += lg.queries_dropped.load(.monotonic);
         self.rows_written += lg.rows_written.load(.monotonic);
         self.batches_gated += lg.batches_gated.load(.monotonic);
         self.last_drop_s = @max(self.last_drop_s, lg.last_drop_s.load(.monotonic));
+        self.queue_peak = @max(self.queue_peak, lg.queue.peakOccupancy());
     }
 };
 
@@ -80,6 +82,8 @@ pub const Sample = struct {
     batches_gated: u64,
     /// Null while nothing has been dropped, exactly as `Logger.lastDropSeconds`.
     last_drop_s: ?i64,
+    /// The most entries any generation's queue has held at once.
+    queue_peak: u64,
     /// The LIVE writer's state. A retired writer's failures are in the counters.
     writer_failed: bool,
     /// The worst episode any outstanding generation is in. A retired writer
@@ -376,6 +380,7 @@ pub const Controller = struct {
             .rows_written = totals.rows_written,
             .batches_gated = totals.batches_gated,
             .last_drop_s = if (totals.last_drop_s == 0) null else totals.last_drop_s,
+            .queue_peak = totals.queue_peak,
             .writer_failed = self.live.logger.writer_failed.load(.acquire),
             .gate_episode = episode,
         };