query log batching: one transaction per flush interval, not per query

src/storage/logger.zig

@@ -13,9 +13,19 @@
 //! order; `QuerySink` calls them separately so both of its consumers see the
 //! one transformed entry.
 //!
+//! One transaction per `logging.query_log_flush_interval_s`, not one per query.
+//! At household rates a per-query commit costs orders of magnitude more disk
+//! writes than the rows are worth — a minute of batching is what keeps an SD
+//! card alive. The window is also what a crash costs: while the writer is
+//! healthy and the disk gate is open, a process that dies loses about one
+//! interval of query history. It is not a ceiling — a batch the gate is
+//! holding, or one waiting on a write lock, is older than that — and nothing
+//! here promises one, because query history is the cheapest data on the box.
+//!
 //! Log rows are expendable. A full queue drops the oldest unflushed entry, a
 //! failed batch is dropped whole, and a disk that crossed the critical
-//! threshold holds batches back indefinitely. Each of the three has a counter.
+//! threshold holds batches back until it recovers — or, if the process is
+//! already stopping, drops what it holds. Each of the three has a counter.
 //! A writer that cannot prepare its statements closes the queue and marks
 //! `writer_failed`, so the loss is visible rather than silent.
 
@@ -31,10 +41,10 @@ const queries_repo = @import("repositories/queries_repo.zig");
 /// and the two names collide inside the struct.
 const scope = std.log.scoped(.query_logger);
 
-/// Flush tuning is comptime: §12.1 defines no configuration keys for it and a
-/// household deployment has no reason to tune it.
+/// Rows per transaction. Comptime: the window, not the batch size, is what an
+/// operator has a reason to move, and a batch this size already amortizes the
+/// commit at any household rate.
 pub const flush_batch = 100;
-pub const flush_interval_ms = 100;
 
 /// What `hide_domains` and `hide_client_ips` store instead of the real value.
 pub const hidden_marker = "hidden";
@@ -161,6 +171,14 @@ fn emptyAsNull(value: []const u8) ?[]const u8 {
 
 const EntryQueue = std.Io.Queue(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.
+/// `GatedAtShutdown` is the one condition `flush` cannot settle by itself —
+/// the disk gate is shut and the process is stopping, so the batch in hand is
+/// lost and so is everything still queued behind it. Only `runWriter` can see
+/// both, so it does the counting and files the single episode.
+pub const FlushError = std.Io.Cancelable || error{GatedAtShutdown};
+
 /// What the flush interval race can produce. `Select` demands that each field
 /// type match its task's return type exactly.
 const Outcome = union(enum) {
@@ -178,6 +196,10 @@ pub const Logger = struct {
     /// closed and every entry counts as dropped from that point, so a caller
     /// that sees this must not expect rows.
     writer_failed: std.atomic.Value(bool),
+    /// Set by `shutdown`, read by the disk gate. The gate holds a batch for as
+    /// long as the disk stays critical, which is right while the process is
+    /// running and a hang once it is stopping — nothing will ever release it.
+    draining: std.atomic.Value(bool),
     /// Wired by the composition root after `init`, following the
     /// `gate: ?*disk_monitor.Monitor` idiom. Null in every unit test here.
     diagnostics: ?*events.Store = null,
@@ -195,6 +217,7 @@ pub const Logger = struct {
             .rows_written = .init(0),
             .batches_gated = .init(0),
             .writer_failed = .init(false),
+            .draining = .init(false),
         };
     }
 
@@ -246,6 +269,11 @@ pub const Logger = struct {
     /// whole life. Returns when `shutdown` closes the queue and the last batch
     /// is flushed, or when the task is canceled.
     ///
+    /// It must outlive the producers rather than share their lifetime: a
+    /// cancellation that races the close decides at random whether the batch in
+    /// hand is written or counted as dropped. `app.zig` spawns this outside the
+    /// group it cancels for exactly that reason.
+    ///
     /// `monitor` is the §11.6 gate. Null disables gating.
     pub fn runWriter(
         self: *Logger,
@@ -262,7 +290,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);
             return;
         };
         defer writer.deinit();
@@ -275,10 +303,7 @@ pub const Logger = struct {
                 error.Closed => return,
                 error.Canceled => |e| return e,
             };
-            const deadline: std.Io.Clock.Timestamp = .fromNow(io, .{
-                .raw = .fromMilliseconds(flush_interval_ms),
-                .clock = .awake,
-            });
+            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;
@@ -286,17 +311,55 @@ pub const Logger = struct {
                 self.countDropped(n);
                 return err;
             };
-            self.flush(io, &writer, batch[0..n], monitor) catch |err| {
-                self.countDropped(n);
-                return err;
+            self.flush(io, &writer, batch[0..n], monitor) catch |err| switch (err) {
+                error.Canceled => |e| {
+                    self.countDropped(n);
+                    return e;
+                },
+                // Nothing will open the gate now. Everything still queued is
+                // lost with the batch in hand, so it is counted here and
+                // 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);
+                    scope.warn(
+                        "query log: {d} rows dropped at shutdown, the disk gate was closed",
+                        .{lost},
+                    );
+                    self.reportWrite(
+                        io,
+                        "batch",
+                        "query log rows were dropped at shutdown",
+                        "DiskCritical",
+                        lost,
+                    );
+                    return;
+                },
             };
         }
     }
 
-    /// Counts every entry left in a closed queue as dropped. 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) void {
+    /// When the batch that starts now must be committed. `.boot` and not
+    /// `.awake`: a suspended box would otherwise stretch the window by however
+    /// long it slept, and the rows are already in memory waiting.
+    ///
+    /// An interval of 0 yields a deadline that has already passed, which is
+    /// exactly the documented sentinel — `fill` then takes what is queued and
+    /// returns without waiting for anything.
+    fn flushDeadline(self: *const Logger, io: std.Io) std.Io.Clock.Timestamp {
+        return .fromNow(io, .{
+            .raw = .fromSeconds(self.cfg.query_log_flush_interval_s),
+            .clock = .boot,
+        });
+    }
+
+    /// Counts every entry left in a closed queue as dropped, and returns how
+    /// 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 {
+        var total: usize = 0;
         var leftover: [flush_batch]Entry = undefined;
         while (true) {
             const n = self.queue.getUncancelable(io, &leftover, 0) catch |err| switch (err) {
@@ -304,18 +367,24 @@ pub const Logger = struct {
             };
             if (n == 0) break;
             self.countDropped(n);
+            total += n;
         }
+        return total;
     }
 
-    /// Closes the queue. `log` drops from here on and `runWriter` returns once
-    /// it has flushed what was left.
+    /// Closes the queue and tells the writer to stop waiting on anything that
+    /// may never arrive. `log` drops from here on, and `runWriter` returns once
+    /// it has flushed what was left — a blocked `getOne` on a closed queue
+    /// returns immediately, so the interval is never waited out at shutdown.
     ///
-    /// A writer held by the disk gate keeps holding: it flushes when the disk
-    /// recovers, and the `group.cancel` that follows this call in `app.zig`
-    /// stops a writer that will not wait. A canceled
-    /// writer counts the batch it holds under `queries_dropped`.
+    /// The close comes first: a writer that sees `draining` set must be able to
+    /// drain the queue to the end, and only a closed queue reports its end.
+    ///
+    /// Every producer must be stopped and joined before this is called
+    /// (`app.zig`): an entry enqueued after the close is a dropped entry.
     pub fn shutdown(self: *Logger, io: std.Io) void {
         self.queue.close(io);
+        self.draining.store(true, .release);
     }
 
     /// Fills `batch` behind the entry already in slot 0, until it is full or
@@ -387,7 +456,7 @@ pub const Logger = struct {
         writer: *queries_repo.BatchWriter,
         entries: []const Entry,
         monitor: ?*disk_monitor.Monitor,
-    ) std.Io.Cancelable!void {
+    ) FlushError!void {
         if (entries.len == 0) return;
 
         if (monitor) |m| {
@@ -396,6 +465,13 @@ pub const Logger = struct {
                 .clock = .awake,
             };
             while (!m.writesAllowed()) {
+                // 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
+                // the only choice left is between losing it counted and
+                // hanging the exit. The counting and the one report belong to
+                // `runWriter`, which knows the rest of the queue is lost too.
+                if (self.draining.load(.acquire)) return error.GatedAtShutdown;
                 _ = self.batches_gated.fetchAdd(1, .monotonic);
                 try pause.sleep(io);
             }
@@ -691,7 +767,7 @@ test "log after shutdown drops instead of blocking" {
     try testing.expectEqual(@as(u64, 1), logger.queries_dropped.load(.monotonic));
 }
 
-test "the writer drains every entry and shutdown ends it" {
+test "shutdown writes the batch the writer holds and the rest of the queue" {
     var threaded: std.Io.Threaded = .init(testing.allocator, .{});
     defer threaded.deinit();
     const io = threaded.io();
@@ -700,7 +776,9 @@ test "the writer drains every entry and shutdown ends it" {
     defer database.close();
 
     var buf: [512]Entry = undefined;
-    var logger: Logger = .init(.{}, &buf);
+    // An hour: nothing here can be explained by the window expiring. Every row
+    // that lands does so because the close released it.
+    var logger: Logger = .init(.{ .query_log_flush_interval_s = 3600 }, &buf);
 
     var future = try io.concurrent(Logger.runWriter, .{
         &logger,
@@ -714,6 +792,10 @@ test "the writer drains every entry and shutdown ends it" {
         const written = try std.fmt.bufPrint(name, "d{d}.example", .{i % 10});
         logger.log(io, sampleEntry(@intCast(i), written));
     }
+
+    // The app.zig order: the only producer is this task and it is done, so the
+    // close cannot lose an entry, and the writer is awaited rather than
+    // canceled.
     logger.shutdown(io);
     try future.await(io);
 
@@ -723,7 +805,57 @@ test "the writer drains every entry and shutdown ends it" {
     try testing.expectEqual(@as(i64, 10), try queries_repo.countDomains(&database));
 }
 
-test "the writer flushes an entry once the interval passes" {
+test "entries that arrive inside one window reach the database in one batch" {
+    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    var database = try openLog();
+    defer database.close();
+    var writer = try queries_repo.BatchWriter.init(&database);
+    defer writer.deinit();
+
+    var buf: [16]Entry = undefined;
+    var logger: Logger = .init(.{ .query_log_flush_interval_s = 3600 }, &buf);
+
+    for (0..6) |i| logger.log(io, sampleEntry(@intCast(i), "batched.example"));
+    // No further producer exists, so the close is what ends the fill — the same
+    // thing that ends it at shutdown.
+    logger.shutdown(io);
+
+    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 testing.expectEqual(@as(usize, 6), n);
+
+    // One `writeBatch` call, which is one transaction (`queries_repo.zig`).
+    try logger.flush(io, &writer, batch[0..n], null);
+    try testing.expectEqual(@as(u64, 6), logger.rows_written.load(.monotonic));
+    try testing.expectEqual(@as(i64, 6), try queries_repo.countRows(&database));
+}
+
+test "a zero interval takes what is queued and waits for nothing" {
+    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    var buf: [8]Entry = undefined;
+    var logger: Logger = .init(.{ .query_log_flush_interval_s = 0 }, &buf);
+
+    logger.log(io, sampleEntry(1, "now.example"));
+    logger.log(io, sampleEntry(2, "now.example"));
+
+    var batch: [flush_batch]Entry = undefined;
+    batch[0] = try logger.queue.getOne(io);
+    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 testing.expectEqual(@as(usize, 2), n);
+}
+
+test "the writer holds an entry for the length of the flush interval" {
     var threaded: std.Io.Threaded = .init(testing.allocator, .{});
     defer threaded.deinit();
     const io = threaded.io();
@@ -732,7 +864,7 @@ test "the writer flushes an entry once the interval passes" {
     defer database.close();
 
     var buf: [8]Entry = undefined;
-    var logger: Logger = .init(.{}, &buf);
+    var logger: Logger = .init(.{ .query_log_flush_interval_s = 1 }, &buf);
 
     var future = try io.concurrent(Logger.runWriter, .{
         &logger,
@@ -743,20 +875,71 @@ test "the writer flushes an entry once the interval passes" {
 
     logger.log(io, sampleEntry(1, "only.example"));
 
+    // Sampled, not asserted, while the writer runs: an assertion that fails
+    // here would return before the writer is stopped, and `Threaded.deinit`
+    // then waits on a task nothing will ever end.
+    const quarter: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(250), .clock = .awake };
+    try quarter.sleep(io);
+    const written_at_a_quarter = logger.rows_written.load(.monotonic);
+
+    // Three times the interval. A flush that has not happened by then is a
+    // failure, not slowness.
     const poll: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(5), .clock = .awake };
     var waited: usize = 0;
-    while (logger.rows_written.load(.monotonic) == 0) : (waited += 1) {
-        // Ten times the interval; a flush that has not happened by then is a
-        // failure, not slowness.
-        try testing.expect(waited < 200);
+    while (logger.rows_written.load(.monotonic) == 0 and waited < 600) : (waited += 1) {
         try poll.sleep(io);
     }
 
     logger.shutdown(io);
     try future.await(io);
+
+    // A writer that commits per query has already written at a quarter of the
+    // window; this one has not.
+    try testing.expectEqual(@as(u64, 0), written_at_a_quarter);
+    try testing.expect(waited < 600);
+    try testing.expectEqual(@as(u64, 1), logger.rows_written.load(.monotonic));
     try testing.expectEqual(@as(i64, 1), try queries_repo.countRows(&database));
 }
 
+test "a full batch flushes without waiting for the interval" {
+    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    var database = try openLog();
+    defer database.close();
+
+    var buf: [256]Entry = undefined;
+    var logger: Logger = .init(.{ .query_log_flush_interval_s = 3600 }, &buf);
+
+    var future = try io.concurrent(Logger.runWriter, .{
+        &logger,
+        io,
+        &database,
+        @as(?*disk_monitor.Monitor, null),
+    });
+
+    for (0..150) |i| logger.log(io, sampleEntry(@intCast(i), "burst.example"));
+
+    // The window is an hour away, so only a full batch can release a flush.
+    const poll: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(5), .clock = .awake };
+    var waited: usize = 0;
+    while (logger.rows_written.load(.monotonic) < flush_batch and waited < 400) : (waited += 1) {
+        try poll.sleep(io);
+    }
+    // Sampled with the writer still running, asserted once it has stopped.
+    const written_before_shutdown = logger.rows_written.load(.monotonic);
+
+    logger.shutdown(io);
+    try future.await(io);
+
+    // Exactly one batch went out early, and the 50 behind it waited for the
+    // close rather than for the hour.
+    try testing.expect(waited < 400);
+    try testing.expectEqual(@as(u64, flush_batch), written_before_shutdown);
+    try testing.expectEqual(@as(i64, 150), try queries_repo.countRows(&database));
+}
+
 test "a gated flush holds the batch until the disk recovers" {
     var threaded: std.Io.Threaded = .init(testing.allocator, .{});
     defer threaded.deinit();
@@ -867,7 +1050,9 @@ test "a canceled writer counts the batch it was holding" {
     defer database.close();
 
     var buf: [8]Entry = undefined;
-    var logger: Logger = .init(.{}, &buf);
+    // Zero interval: the writer reaches the gate with what is queued instead of
+    // waiting a minute for a producer that no longer exists.
+    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);
@@ -898,6 +1083,67 @@ test "a canceled writer counts the batch it was holding" {
     try testing.expectEqual(@as(i64, 0), try queries_repo.countRows(&database));
 }
 
+test "a disk-gated writer drops what it holds at shutdown instead of hanging" {
+    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    var database = try openLog();
+    defer database.close();
+
+    var fx: events_fixture.Fixture = .{};
+    try fx.init(io, 1000);
+    defer fx.deinit();
+
+    var buf: [512]Entry = undefined;
+    var logger: Logger = .init(.{ .query_log_flush_interval_s = 0 }, &buf);
+    logger.diagnostics = &fx.store;
+
+    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"));
+    logger.log(io, sampleEntry(2, "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 and waited < 400) : (waited += 1) {
+        try poll.sleep(io);
+    }
+
+    // Two and a half chunks wait behind the batch the gate is holding, so a
+    // drain that reported per chunk would file three episodes' worth of writes
+    // to the disk that is out of space.
+    const queued = 250;
+    for (0..queued) |i| logger.log(io, sampleEntry(@intCast(i + 3), "queued.example"));
+
+    // The disk never recovers. The wait still ends, and every entry is counted.
+    logger.shutdown(io);
+    try future.await(io);
+
+    try testing.expect(waited < 400);
+    try testing.expectEqual(@as(u64, queued + 2), logger.queries_dropped.load(.monotonic));
+    try testing.expectEqual(@as(u64, 0), logger.rows_written.load(.monotonic));
+    try testing.expectEqual(@as(i64, 0), try queries_repo.countRows(&database));
+
+    // One episode, reported once, and its detail carries the whole loss rather
+    // than the size of whichever chunk was last.
+    try testing.expectEqual(@as(i64, 1), try fx.count("SELECT count(*) FROM operational_events"));
+    try testing.expectEqual(@as(i64, 1), try fx.count("SELECT occurrences FROM operational_events"));
+    try testing.expectEqualStrings("batch", try fx.text("SELECT subject_key FROM operational_events"));
+    try testing.expectEqualStrings(
+        "query log rows were dropped at shutdown: DiskCritical (252 rows)",
+        try fx.text("SELECT detail FROM operational_events"),
+    );
+}
+
 test "an empty batch touches neither the database nor the counters" {
     var threaded: std.Io.Threaded = .init(testing.allocator, .{});
     defer threaded.deinit();