diff --git a/src/storage/logger.zig b/src/storage/logger.zig index 3173cf2..66b305d 100644 --- a/src/storage/logger.zig +++ b/src/storage/logger.zig @@ -1334,6 +1334,13 @@ test "shutdown writes the batch the writer holds and the rest of the queue" { &database, @as(?*disk_monitor.Monitor, null), }); + // Declared after the `database.close` defer so LIFO stops the writer first: + // an early return anywhere below would otherwise close the handle under a + // live writer and leave `Threaded.deinit` joining a task nothing ends. + defer { + logger.shutdown(io); + future.await(io) catch {}; + } var names: [250][32]u8 = undefined; for (&names, 0..) |*name, i| { @@ -1420,6 +1427,12 @@ test "the writer holds an entry for the length of the flush interval" { &database, @as(?*disk_monitor.Monitor, null), }); + // See the note in "shutdown writes the batch the writer holds": this must + // run before the deferred `database.close`. + defer { + logger.shutdown(io); + future.await(io) catch {}; + } logger.log(io, sampleEntry(1, "only.example")); @@ -1466,6 +1479,12 @@ test "a full batch flushes without waiting for the interval" { &database, @as(?*disk_monitor.Monitor, null), }); + // See the note in "shutdown writes the batch the writer holds": this must + // run before the deferred `database.close`. + defer { + logger.shutdown(io); + future.await(io) catch {}; + } for (0..150) |i| logger.log(io, sampleEntry(@intCast(i), "burst.example")); @@ -1513,6 +1532,12 @@ test "a gated flush holds the batch until the disk recovers" { @as([]const Entry, &entries), @as(?*disk_monitor.Monitor, &monitor), }); + // This task is `flush`, not `runWriter`: no queue shutdown can release it, + // so only cancellation ends it on an early return. Declared after the + // `writer.deinit`/`database.close` defers so LIFO runs it first. + defer { + _ = future.cancel(io) catch {}; + } const poll: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(5), .clock = .awake }; var waited: usize = 0; @@ -1634,6 +1659,12 @@ test "the gating episode opens on the gate, turns losing on a drop, and clears o &database, @as(?*disk_monitor.Monitor, &monitor), }); + // See the note in "shutdown writes the batch the writer holds": this must + // run before the deferred `database.close`. + defer { + logger.shutdown(io); + future.await(io) catch {}; + } const poll: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(5), .clock = .awake }; var waited: usize = 0; @@ -1657,11 +1688,14 @@ test "the gating episode opens on the gate, turns losing on a drop, and clears o // 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) { + // The write is the completion condition, not the episode: `flush` reopens + // the gate before it calls `writeBatch`, so a poll on the episode alone + // returns while the row is still in flight and `rows_written` is still 0. + while (logger.rows_written.load(.monotonic) == 0) : (waited += 1) { try testing.expect(waited < 400); try poll.sleep(io); } - try testing.expect(logger.rows_written.load(.monotonic) > 0); + try testing.expectEqual(GateEpisode.open, logger.gateEpisode()); // The count keeps the history the state does not. try testing.expect(logger.queries_dropped.load(.monotonic) > 0); @@ -1736,6 +1770,13 @@ test "a gate that closes mid-enqueue still gets the discard that follows it" { // 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 }); + // The producer parks inside `enqueue` and only `release` frees it: an early + // return below would otherwise leave `Threaded.deinit` joining it forever. + defer { + discard_stall.armed = false; + discard_stall.release.set(io); + producer.await(io); + } discard_stall.parked.waitUncancelable(io); // The producer is parked with the row it will evict still on the queue, so @@ -1799,6 +1840,12 @@ test "a canceled writer counts the batch it was holding" { &database, @as(?*disk_monitor.Monitor, &monitor), }); + // Cancellation is this case's subject, so the guard is the same operation + // the body performs; declared after the `database.close` defer so LIFO ends + // the writer before the handle goes away. + defer { + _ = future.cancel(io) catch {}; + } const poll: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(5), .clock = .awake }; var waited: usize = 0; @@ -1842,6 +1889,12 @@ test "a disk-gated writer drops what it holds at shutdown instead of hanging" { &database, @as(?*disk_monitor.Monitor, &monitor), }); + // See the note in "shutdown writes the batch the writer holds": this must + // run before the deferred `fx.deinit` and `database.close`. + defer { + logger.shutdown(io); + future.await(io) catch {}; + } const poll: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(5), .clock = .awake }; var waited: usize = 0; diff --git a/src/storage/phase6_integration_test.zig b/src/storage/phase6_integration_test.zig index f031a12..fddf032 100644 --- a/src/storage/phase6_integration_test.zig +++ b/src/storage/phase6_integration_test.zig @@ -211,6 +211,13 @@ test "S8 case 1: the logger writes a real querylog.db end to end" { log_db.database(), @as(?*disk_monitor.Monitor, null), }); + // Declared after the `log_db.deinit` defer so LIFO stops the writer first: + // an early return below would otherwise close the database under a live + // writer and leave the never-closed queue parking it forever. + defer { + query_log.shutdown(io); + future.await(io) catch {}; + } var name_buf: [32]u8 = undefined; for (0..250) |i| { @@ -252,6 +259,13 @@ test "S8 case 2: a single entry reaches the file once the flush interval passes" log_db.database(), @as(?*disk_monitor.Monitor, null), }); + // Declared after the `log_db.deinit` defer so LIFO stops the writer first: + // an early return below would otherwise close the database under a live + // writer and leave the never-closed queue parking it forever. + defer { + query_log.shutdown(io); + future.await(io) catch {}; + } query_log.log(io, entryAt(1, "only.example")); @@ -300,6 +314,13 @@ test "S8 case 3: a full queue drops the oldest entries and the newest survive" { log_db.database(), @as(?*disk_monitor.Monitor, &monitor), }); + // Declared after the `log_db.deinit` defer so LIFO stops the writer first: + // an early return below would otherwise close the database under a live + // writer and leave the never-closed queue parking it forever. + defer { + query_log.shutdown(io); + future.await(io) catch {}; + } try awaitCount(&query_log.batches_gated, 1, 200); try testing.expectEqual(@as(i64, 0), try queries_repo.countRows(log_db.database())); @@ -341,6 +362,13 @@ test "S8 case 4: the privacy transforms reach the stored rows" { log_db.database(), @as(?*disk_monitor.Monitor, null), }); + // Declared after the `log_db.deinit` defer so LIFO stops the writer first: + // an early return below would otherwise close the database under a live + // writer and leave the never-closed queue parking it forever. + defer { + query_log.shutdown(io); + future.await(io) catch {}; + } var name_buf: [32]u8 = undefined; for (0..20) |i| { @@ -433,6 +461,13 @@ test "S8 case 6: a critical disk gates the flushes and recovery releases them" { log_db.database(), @as(?*disk_monitor.Monitor, &monitor), }); + // Declared after the `log_db.deinit` defer so LIFO stops the writer first: + // an early return below would otherwise close the database under a live + // writer and leave the never-closed queue parking it forever. + defer { + query_log.shutdown(io); + future.await(io) catch {}; + } try awaitCount(&query_log.batches_gated, 1, 200); try testing.expectEqual(@as(u64, 0), query_log.rows_written.load(.monotonic));