diff --git a/src/db.zig b/src/db.zig index 6cd38a3..238b4b1 100644 --- a/src/db.zig +++ b/src/db.zig @@ -765,6 +765,27 @@ pub const Engine = struct { /// A leaf: nothing else is taken while it is held, and it is never held /// across an append, an fsync, or an allocation. counter_lock: std.Io.Mutex = .init, + /// One checkpoint at a time. + /// + /// Two can be in flight without it -- a writer's epilogue claims the + /// pending flag while another writer's epilogue is inside `compact`, which + /// checkpoints of its own. The publish itself was always safe, because it + /// runs under `log_lock`; what is not is the phase in front of it, which is + /// new. `reclaim_slabs` frees pages under a collection's lock, and the + /// other checkpoint's `write_catalog` may already have serialized that + /// collection's runs. It then publishes a catalog claiming pages that are + /// on the free list, and two generations later they are handed out and + /// written over -- so the crash that falls back to that generation reads a + /// document that is no longer there. The `seq` retry cannot see it, because + /// neither a reclamation nor a rebuild appends a log record. + /// + /// The argument is by construction; no test reproduces the interleaving. + /// What would catch it is the ownership assertion in `write_catalog`, which + /// is armed in test and Debug builds and is proven to fire. + /// + /// Taken before `catalog_lock`, so the order is checkpoint -> catalog -> + /// collection, the same descent everything else makes. + checkpoint_lock: std.Io.Mutex = .init, /// The checkpoint's own page promise, for the catalog and free-list pages it /// writes. Separate from any collection's for the same reason those are /// separate from each other. @@ -2335,6 +2356,17 @@ pub const Engine = struct { // no second read path to keep working. try put_u32(gpa, out, @intCast(coll.slab_runs.items.len)); for (coll.slab_runs.items) |r| { + // A catalog that claims a page already on the free list is + // the one failure this whole design can produce silently: + // the page is handed out two generations later, written + // over, and the crash that falls back to this generation + // reads a document that is no longer there. Nothing else + // notices. Checked where every run is walked anyway, in the + // builds that can afford three list scans. + if (builtin.is_test or builtin.mode == .Debug) assert_msg( + !self.pager.owns_freed(r.first, r.pages), + "the catalog claims a slab run that is already on the free list", + ); try put_u32(gpa, out, r.first); try put_u32(gpa, out, r.pages); } @@ -2629,6 +2661,8 @@ pub const Engine = struct { /// compaction has always used. That is the crash-recovery invariant (PLAN /// D6) reduced to an ordering. pub fn checkpoint(self: *Engine) !void { + try self.checkpoint_lock.lock(self.io); + defer self.checkpoint_lock.unlock(self.io); try self.commit(); self.reclaim_slabs(); @@ -4584,6 +4618,83 @@ test "a checkpoint runs alongside writers on several collections" { try testing.expectEqual(docs_sum, engine.live_docs); } +test "checkpoints reclaim under concurrent writers without losing a page" { + // Reclamation is now a checkpoint phase, so it runs concurrently with + // writers and with a second checkpoint -- two are reachable without + // contrivance, since a writer's epilogue can claim the pending flag while + // another is inside `compact`, which checkpoints of its own. + // + // The ownership assertion in `write_catalog` is armed here: no collection + // may claim a run the pager has already been given. That is the one failure + // this design can produce silently, and it is proven to fire -- have + // `reclaim_windows` free the pages and keep the old run list, and this test + // panics on it. + // + // What it does *not* do is reproduce the interleaving `checkpoint_lock` + // exists for; removing that lock leaves this green. The lock is there by + // argument, and this is the harness that would catch the argument being + // wrong. Said plainly rather than left to be assumed. + 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(); + engine.compact_threshold = std.math.maxInt(u64); // no rebuild may intervene + + var done = std.atomic.Value(usize).init(1); + const Worker = struct { + fn writer(e: *Engine, left: *std.atomic.Value(usize), alloc: std.mem.Allocator) error{Canceled}!void { + defer _ = left.fetchSub(1, .release); + for (0..300) |i| { + var doc = make_padded(alloc, @intCast(i), 4000) catch return error.Canceled; + defer doc.deinit(); + { + e.lock_catalog(false) catch return error.Canceled; + defer e.unlock_catalog(false); + const coll = (e.lock_collection("app", "c", true, true) catch + return error.Canceled) orelse return error.Canceled; + defer e.unlock_collection(coll, true); + e.insert("app", "c", &doc, undefined) catch return error.Canceled; + if (i % 3 == 2) _ = e.remove_by_id("app", "c", .{ .int32 = @intCast(i - 1) }) catch + return error.Canceled; + } + e.commit() catch return error.Canceled; + } + } + + fn checkpointer(e: *Engine, left: *std.atomic.Value(usize)) error{Canceled}!void { + while (left.load(.acquire) > 0) e.checkpoint() catch {}; + } + }; + + var group: std.Io.Group = .init; + defer group.cancel(io); + group.async(io, Worker.writer, .{ &engine, &done, gpa }); + group.async(io, Worker.checkpointer, .{ &engine, &done }); + group.async(io, Worker.checkpointer, .{ &engine, &done }); + try group.await(io); + + // Checkpoints actually happened, and the last one published cleanly: the + // watermark the pager loaded is the generation it just wrote. + const generation = engine.pager.generation; + try testing.expect(generation > 1); + try engine.checkpoint(); + try testing.expectEqual(generation + 1, engine.pager.generation); + try testing.expectEqual(generation + 1, engine.pager.loaded.generation); + + // And the accounting the two of them were racing on still adds up. + const coll = engine.get_collection("app", "c").?; + try testing.expectEqual(coll.slab_used - coll.live_bytes, engine.dead_bytes); + try testing.expectEqual( + coll.slab_used - coll.live_bytes, + coll.dead_located() + coll.dead_unlocated, + ); +} + 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 diff --git a/src/pager.zig b/src/pager.zig index aa960c6..d9e96fb 100644 --- a/src/pager.zig +++ b/src/pager.zig @@ -1216,6 +1216,29 @@ pub const Pager = struct { try self.free_pending.append(self.gpa, .{ .first = first, .pages = pages }); } + /// Whether any page of `[first, first+pages)` has been handed to the free + /// list. A consumer that still claims one is claiming a page the pager is + /// about to give to somebody else, and the symptom is a document quietly + /// overwritten rather than anything failing -- so this is the detector the + /// enlarged free list deserves, and it is why the free lists are readable + /// from outside at all. + /// + /// Walks three lists, so it is for assertions in test and Debug builds. + pub fn owns_freed(self: *Pager, first: u32, pages: u32) bool { + self.alloc_lock.lockUncancelable(self.io); + defer self.alloc_lock.unlock(self.io); + for ([_][]const Extent{ + self.free_pending.items, + self.free_hold.items, + self.free_ready.items, + }) |list| { + for (list) |e| { + if (first < e.first + e.pages and e.first < first + pages) return true; + } + } + return false; + } + /// Pages available for immediate reuse. pub fn free_ready_pages(self: *const Pager) u32 { var n: u32 = 0;