diff --git a/src/db.zig b/src/db.zig index 57ba93e..1ca867d 100644 --- a/src/db.zig +++ b/src/db.zig @@ -1612,14 +1612,20 @@ pub const Engine = struct { const catalog_magic: u32 = 0x4D464354; // "MFCT" const catalog_version: u32 = 1; - fn write_catalog(self: *Engine, out: *std.ArrayListUnmanaged(u8)) !void { + /// Serialize the catalog and return the live-byte total it observed. + /// + /// The engine's own total is by definition the sum over collections, and + /// `read_catalog` rebuilds it that way, so a divergence means some path + /// published or evicted bytes at one level and not the other -- with a + /// compaction trigger that fires never or always as the visible symptom. + /// The check is worth making and this is where every collection is walked + /// anyway, but it cannot be made *here*: the sum is accumulated across + /// collections over time while the engine's total moves under it, so a + /// writer landing mid-walk would trip it on a database that is perfectly + /// consistent. The caller asserts it after the `seq` check has established + /// that no writer landed at all. + fn write_catalog(self: *Engine, out: *std.ArrayListUnmanaged(u8)) !u64 { const gpa = self.gpa; - // The engine's live-byte total is by definition the sum over - // collections, and `read_catalog` rebuilds it that way. Check it here, - // where every collection is being walked regardless: a divergence means - // some path published or evicted bytes at one level and not the other, - // and the visible symptom would be a compaction trigger that fires - // never or always. var live_sum: u64 = 0; try put_u32(gpa, out, catalog_magic); try put_u32(gpa, out, catalog_version); @@ -1653,11 +1659,8 @@ pub const Engine = struct { for (coll.indexes.items) |ix| try write_index_catalog(gpa, out, ix); } } - assert_msg( - live_sum == self.live_bytes, - "the engine's live-byte total must equal the sum over collections", - ); try put_u64(gpa, out, std.hash.XxHash3.hash(0, out.items)); + return live_sum; } fn write_index_catalog( @@ -1883,7 +1886,8 @@ pub const Engine = struct { buf.clearRetainingCapacity(); try self.catalog_lock.lockShared(self.io); const snapshot_seq = self.seq; - self.write_catalog(&buf) catch |err| { + const live_before = self.live_bytes; + const live_sum = self.write_catalog(&buf) catch |err| { self.catalog_lock.unlockShared(self.io); return err; }; @@ -1896,9 +1900,42 @@ pub const Engine = struct { self.log_lock.unlock(self.io); continue; } - assert_msg( - snapshot_seq <= self.committed_seq, - "checkpoint watermark past the durable log tail", + if (snapshot_seq > self.committed_seq) { + // A writer appended before the snapshot and its commit has not + // landed yet -- it is between `insert` and `commit`, or inside + // one, waiting on the leader's fsync. The seq check above does + // not catch this: nothing appended *during* the walk, the + // append was already there when it started. + // + // Publishing here would claim durability for a record that is + // still in the log's buffer, and the truncation that follows a + // checkpoint would then throw it away. That is the one thing + // the whole watermark ordering exists to prevent (PLAN D6), and + // it used to be an assertion -- so the failure mode was a + // server abort under exactly the load that makes checkpoints + // frequent. Reproduced in seconds by four writers against a + // checkpoint loop, and the window is as wide as an fsync. + // + // Seal it and take the snapshot again rather than spinning: + // `commit` covers every append made so far, so one more round + // is enough. Outside `log_lock`, which `commit` takes itself. + self.log_lock.unlock(self.io); + try self.commit(); + continue; + } + // Only when the walk was quiet. An unchanged `seq` is not enough on + // its own: a writer bumps it when it appends the log record and + // updates the byte counters afterwards, so it can be past the seq + // the snapshot captured and still be about to move `live_bytes` + // under a collection the walk has already been through. Requiring + // the engine total to be unmoved across the whole walk closes that, + // at the cost of skipping the check under sustained writes -- which + // is the right trade, because what it guards against is a code path + // that updates one level and not the other, and that is + // deterministic wherever it exists. + if (self.live_bytes == live_before) assert_msg( + live_sum == live_before, + "the engine's live-byte total must equal the sum over collections", ); const pages: u32 = @intCast((buf.items.len + pgr.page_size - 1) / pgr.page_size); const first = self.pager.alloc_pages(pages) catch |err| { @@ -1906,6 +1943,12 @@ pub const Engine = struct { return err; }; @memcpy(self.pager.bytes_mut(@as(u64, first) << pgr.page_shift, buf.items.len), buf.items); + // The invariant, on the line that would break it: `log_lock` has + // been held since the check above and `committed_seq` only grows. + assert_msg( + snapshot_seq <= self.committed_seq, + "checkpoint watermark past the durable log tail", + ); self.pager.publish(.{ .seq = snapshot_seq, .catalog_page = first, @@ -2889,6 +2932,85 @@ test "compaction rewrites log and keeps data" { engine3.unlock(); } +test "a checkpoint runs alongside writers on several collections" { + // `write_catalog` reads each collection's slab extents, indexes and byte + // counters while holding only the *shared catalog* lock -- and a writer + // holds that same lock shared, taking the collection's lock exclusively. + // So the snapshot walked structures its owner was free to mutate, and + // `slab_extents` is an ArrayList a new extent appends to: a reallocation + // mid-walk leaves the serializer reading freed memory. + // + // Several collections rather than one, because the interesting overlap is a + // writer on collection B while the catalog is serializing collection A. + // + // Mutation check: drop the `lockShared` from `write_catalog`'s collection + // loop. Not reliably red -- a data race never is -- but it runs under + // ReleaseSafe, where the reads it makes are bounds-checked. + const gpa = testing.allocator; + var threaded: std.Io.Threaded = .init(gpa, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var tmp = try TmpLog.init(gpa); + defer tmp.deinit(gpa); + var engine = try Engine.open(gpa, io, tmp.path); + defer engine.deinit(); + + const colls = [_][]const u8{ "a", "b", "c", "d" }; + const per_coll: i32 = 150; + var done = std.atomic.Value(usize).init(colls.len); + + const Worker = struct { + fn writer( + e: *Engine, + name: []const u8, + left: *std.atomic.Value(usize), + alloc: std.mem.Allocator, + ) error{Canceled}!void { + defer _ = left.fetchSub(1, .release); + for (1..per_coll + 1) |i| { + var doc = make_doc(alloc, @intCast(i), "user") catch return error.Canceled; + defer doc.deinit(); + { + e.lock() catch return error.Canceled; + defer e.unlock(); + e.insert("app", name, &doc, undefined) catch return error.Canceled; + } + // As the dispatch epilogue does (commands.zig): the append bumps + // `seq`, the commit is what makes it durable, and a checkpoint + // may only describe what is durable. + e.commit() catch return error.Canceled; + } + } + + fn checkpointer(e: *Engine, left: *std.atomic.Value(usize)) error{Canceled}!void { + while (left.load(.acquire) > 0) { + // Errors are the point of the retry loop inside `checkpoint`, + // not a failure of this test; a checkpoint that gives up under + // sustained writes has still not corrupted anything. + e.checkpoint() catch {}; + } + } + }; + + var group: std.Io.Group = .init; + defer group.cancel(io); + for (colls) |name| group.async(io, Worker.writer, .{ &engine, name, &done, gpa }); + group.async(io, Worker.checkpointer, .{ &engine, &done }); + try group.await(io); + + // Every write is still there, and the catalog the checkpoints wrote agrees + // with the engine -- the second half is what `write_catalog`'s own assertion + // checks on the way through. + try engine.checkpoint(); + try engine.lock_read(); + defer engine.unlock_read(); + for (colls) |name| { + const coll = engine.get_collection("app", name) orelse return error.TestUnexpectedResult; + try testing.expectEqual(@as(usize, @intCast(per_coll)), coll.id_index.count()); + } +} + test "concurrent readers and writers on a threaded Io" { // Real worker threads: writers hold the exclusive lock, readers the // shared lock. Proves the RwLock split keeps committed writes visible