logger tests: join every writer future before its database closes, fix the gate-open race

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));