diff --git a/src/db.zig b/src/db.zig index 0f80a43..1a4015f 100644 --- a/src/db.zig +++ b/src/db.zig @@ -438,9 +438,14 @@ pub const Engine = struct { engine.seq = replay_from; engine.committed_seq = replay_from; engine.live_docs = engine.pager.loaded.live_docs; - // Without this a restart forgets its garbage, and a churned - // database would never compact again. - engine.dead_bytes = engine.pager.loaded.dead_bytes; + // `dead_bytes` is *not* restored from the watermark. It is + // derived, not stored: `read_catalog` has already summed + // `slab_used - live_bytes` over the collections the catalog + // still lists. The watermark's copy is a hint for anything + // inspecting the header without parsing the catalog, and it + // would be wrong here in one specific way -- a collection + // dropped after the last checkpoint takes its garbage with it, + // and the hint would keep charging the engine for it. } } @@ -498,7 +503,28 @@ pub const Engine = struct { self.dead_docs += coll.doc_count; assert_msg(self.live_bytes >= coll.live_bytes, "dropping a collection would underflow the engine's live bytes"); self.live_bytes -= coll.live_bytes; - self.dead_bytes += coll.live_bytes; + // A drop *reclaims*, it does not deaden. The loop below hands every page + // this collection owned back to the pager, so its live bytes are not + // garbage waiting for a rebuild -- they are already gone. Adding them to + // `dead_bytes` armed a compaction for space that had just been returned, + // and a rebuild costs a full copy of every *other* collection. + // + // Its garbage goes the other way, for the same reason: the bytes this + // collection had already lost to eviction were counted in `dead_bytes` + // when they died, and those pages are being freed too. That keeps + // `dead_bytes` exactly the sum of `slab_used - live_bytes` over the + // collections that still exist, which is what `read_catalog` recomputes + // on open and what `write_catalog` asserts. + assert_msg( + coll.slab_used >= coll.live_bytes, + "a collection cannot hold more live bytes than it ever appended", + ); + const coll_dead = coll.slab_used - coll.live_bytes; + assert_msg( + self.dead_bytes >= coll_dead, + "dropping a collection would underflow the engine's dead bytes", + ); + self.dead_bytes -= coll_dead; coll.id_index.deinit(self.gpa); for (coll.indexes.items) |ix| { ix.deinit(self.gpa); @@ -1643,21 +1669,26 @@ pub const Engine = struct { const catalog_magic: u32 = 0x4D464354; // "MFCT" const catalog_version: u32 = 1; - /// Serialize the catalog and return the live-byte total it observed. + /// What a catalog walk observed, for the caller to check the engine's own + /// running totals against. + const CatalogSums = struct { live: u64, dead: u64 }; + + /// Serialize the catalog and return the byte totals 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 + /// The engine's own totals are by definition the sums over collections, and + /// `read_catalog` rebuilds them 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 + /// anyway, but it cannot be made *here*: the sums are accumulated across + /// collections over time while the engine's totals move under them, 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 + /// consistent. The caller asserts them after the `seq` check has established /// that no writer landed at all. - fn write_catalog(self: *Engine, out: *std.ArrayListUnmanaged(u8)) !u64 { + fn write_catalog(self: *Engine, out: *std.ArrayListUnmanaged(u8)) !CatalogSums { const gpa = self.gpa; var live_sum: u64 = 0; + var dead_sum: u64 = 0; try put_u32(gpa, out, catalog_magic); try put_u32(gpa, out, catalog_version); try put_u64(gpa, out, self.live_docs); @@ -1688,6 +1719,7 @@ pub const Engine = struct { coll.live_bytes <= coll.slab_used, "a collection cannot hold more live bytes than it ever appended", ); + dead_sum += coll.slab_used - coll.live_bytes; try put_u32(gpa, out, @intCast(coll.slab_extents.items.len)); for (coll.slab_extents.items) |e| { try put_u32(gpa, out, e.first); @@ -1699,7 +1731,7 @@ pub const Engine = struct { } } try put_u64(gpa, out, std.hash.XxHash3.hash(0, out.items)); - return live_sum; + return .{ .live = live_sum, .dead = dead_sum }; } fn write_index_catalog( @@ -1764,9 +1796,14 @@ pub const Engine = struct { coll.slab_end = try r.read_u64(); coll.slab_used = try r.read_u64(); coll.live_bytes = try r.read_u64(); - // The engine's total is the sum over collections rather than a - // separately stored field, so the two cannot disagree. + // Both engine totals are sums over collections rather than + // separately stored fields, so neither can disagree with the + // catalog. `slab_used - live_bytes` is this collection's slab + // garbage by definition -- bytes it appended and no longer + // reaches -- which is exactly what the rebuild trigger counts. + if (coll.slab_used < coll.live_bytes) return error.CorruptCatalog; self.live_bytes += coll.live_bytes; + self.dead_bytes += coll.slab_used - coll.live_bytes; const nex = try r.read_u32(); var e: u32 = 0; while (e < nex) : (e += 1) { @@ -1926,7 +1963,8 @@ pub const Engine = struct { try self.catalog_lock.lockShared(self.io); const snapshot_seq = self.seq; const live_before = self.live_bytes; - const live_sum = self.write_catalog(&buf) catch |err| { + const dead_before = self.dead_bytes; + const sums = self.write_catalog(&buf) catch |err| { self.catalog_lock.unlockShared(self.io); return err; }; @@ -1973,9 +2011,16 @@ pub const Engine = struct { // 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, + sums.live == live_before, "the engine's live-byte total must equal the sum over collections", ); + // The same argument, for the total the rebuild trigger reads. This + // is what makes a drop's accounting checkable: charge the engine for + // a dropped collection's bytes and the two sides part company here. + if (self.dead_bytes == dead_before) assert_msg( + sums.dead == dead_before, + "the engine's dead-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| { self.log_lock.unlock(self.io); @@ -2697,8 +2742,9 @@ test "compaction still triggers after a checkpoint has truncated the log" { // then never fires and the doc slab grows without bound. This test exists // because the churn gate measured exactly that: 4.1x live data. // - // Second mutation: drop `dead_bytes` from the watermark, or from the restore - // beside `live_docs` in `open`. Red on the reopened engine below. + // Second mutation: drop the `dead_bytes` accumulation from `read_catalog`. + // Red in "reopen carries the garbage counter across a restart", which is + // where the counter has to survive a restart. var threaded: std.Io.Threaded = .init_single_threaded; defer threaded.deinit(); var env = test_env(&threaded); @@ -2807,6 +2853,14 @@ test "a rebuild leaves the space it reclaimed ready to reuse" { } test "reopen carries the garbage counter across a restart" { + // It carries it by *recomputing* it: `read_catalog` sums + // `slab_used - live_bytes` over the collections the catalog lists, rather + // than trusting the watermark's cached copy. That is a stronger claim than + // the hint was -- a collection dropped since the last checkpoint is simply + // not in the sum, where the hint kept charging the engine for it. + // + // Mutation: drop the accumulation in `read_catalog`; `engine2.dead_bytes` + // reads zero below. var threaded: std.Io.Threaded = .init_single_threaded; defer threaded.deinit(); var env = test_env(&threaded); @@ -2844,6 +2898,87 @@ test "reopen carries the garbage counter across a restart" { defer engine2.deinit(); try testing.expectEqual(dead_before, engine2.dead_bytes); try testing.expectEqual(live_before, engine2.live_bytes); + // And it is the sum over collections on both sides of the restart, not a + // number kept beside them. + const reopened = engine2.get_collection("app", "c").?; + try testing.expectEqual(reopened.slab_used - reopened.live_bytes, engine2.dead_bytes); +} + +test "dropping a collection does not arm compaction" { + // `free_collection` charged the engine's `dead_bytes` with the dropped + // collection's *live* bytes, having just handed every page it owned back to + // the pager on the following line. A drop of a large collection therefore + // asked for a rebuild -- a full copy of every collection that was left -- + // to reclaim space that had already been reclaimed. Its own garbage was + // wrong the other way: it stayed on the engine's books after the pages + // holding it were gone. + // + // Mutation: restore `self.dead_bytes += coll.live_bytes;`, or delete the + // subtraction of `coll.slab_used - coll.live_bytes`. Either one is red on + // the equality below, and the first also re-arms the trigger. + var threaded: std.Io.Threaded = .init_single_threaded; + defer threaded.deinit(); + var env = test_env(&threaded); + const io = env.io; + const gpa = testing.allocator; + + 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 while setting up + try engine.lock(); + defer engine.unlock(); + + // A small collection that survives, and a large one that does not. Both + // hold garbage, so the drop has to keep one collection's and discard the + // other's. + for (0..40) |i| { + var d = try make_doc(gpa, @intCast(i), "x"); + defer d.deinit(); + try engine.insert("app", "keep", &d, &env.gen); + } + for (0..200) |i| { + var d = try make_doc(gpa, @intCast(i), "x"); + defer d.deinit(); + try engine.insert("app", "gone", &d, &env.gen); + } + for (0..10) |i| { + var d = try make_doc(gpa, @intCast(i), "yy"); + defer d.deinit(); + _ = try engine.replace("app", "keep", &d, &env.gen); + } + for (0..200) |i| { + var d = try make_doc(gpa, @intCast(i), "yy"); + defer d.deinit(); + _ = try engine.replace("app", "gone", &d, &env.gen); + } + try engine.commit(); + + const keep = engine.get_collection("app", "keep").?; + const keep_dead = keep.slab_used - keep.live_bytes; + const keep_live = keep.live_bytes; + try testing.expect(keep_dead > 0); + try testing.expect(engine.dead_bytes > keep_dead); + + // Arm the trigger for what the two of them hold between them. + engine.compact_threshold = keep_dead + 1; + engine.note_compact(); + try testing.expect(engine.take_compact()); + + try testing.expect(try engine.drop_collection("app", "gone")); + try testing.expectEqual(keep_dead, engine.dead_bytes); + try testing.expectEqual(keep_live, engine.live_bytes); + + // The next write reconsiders the trigger and finds nothing worth a rebuild: + // what the drop reclaimed is not garbage, it is free. + engine.note_compact(); + try testing.expect(!engine.take_compact()); + + // The catalog agrees, which is what the reopened engine will read. + try engine.checkpoint(); + try testing.expectEqual(keep_dead, engine.dead_bytes); } test "reopen replays log" { diff --git a/src/pager.zig b/src/pager.zig index 5d51f8d..7435389 100644 --- a/src/pager.zig +++ b/src/pager.zig @@ -27,7 +27,8 @@ //! [56..64) u64 freelist_len //! [64..72) u64 prev_generation -- kept intact for fallback //! [72..80) u64 live_docs -- cached hint -//! [80..88) u64 dead_bytes -- cached, drives the rebuild trigger +//! [80..88) u64 dead_bytes -- cached hint; the engine recomputes it +//! from the catalog on open //! [88..4088) reserved (zero) //! [4088..4096) u64 xxhash3 over [0..4088) //! pages 3.. data, handed out by a tail-bump extent allocator