From 148e03ac9f3ad39a019844f4cdf3c3a9f293c16d Mon Sep 17 00:00:00 2001 From: Aleksey Shakhmatov Date: Mon, 3 Aug 2026 21:48:04 +0300 Subject: [PATCH] db: compaction becomes a data-file rebuild `compact` used to re-emit every live document into a fresh log and rename it over the old one. That is the wrong shape twice over now: the log is not where the data lives, and a re-emitted record carries a sequence a later watermark can cover, which would make the next open skip it (PLAN section 4). The log re-emission is deleted; the checkpoint at the end reclaims the log instead. What it reclaims is what a checkpoint cannot. A checkpoint publishes the structures where they already are, and it cannot move a document, because every index leaf holds that document's physical offset. So reclaiming a replaced document's bytes means rewriting the documents *and* repacking every index against the new offsets, together -- which is the whole of `rebuild_collection`. Documents are copied in _id order, so the new slab reads sequentially afterwards. Old extents and old node pages go to the free list rather than being reused immediately, so a crash mid-rebuild simply loses the rebuild: the previous watermark still describes the previous layout, intact. Adds `Collection.slab_used`, because `slab_tail` cannot answer "how many bytes are in use" -- it is an absolute file offset and jumps forward with each new extent. That is also the number the rebuild trigger wants. -- The test is the part worth reading. My first version asserted that every document was still findable and had the replaced contents, and it was nearly useless: two mutations -- not repacking the indexes at all, and not republishing the docs-map offsets -- both left it green. Freed extents go on the free list rather than being overwritten, so a stale offset still reads a perfectly plausible document. What actually distinguishes a repacked index from a stale one is *where* the offset points: after a rebuild every live offset must fall inside an extent the collection currently owns. Asserting that, plus that the index and the map agree, turns all three mutations red -- including repacking `_id_` but forgetting the secondaries. --- src/db.zig | 346 ++++++++++++++++++++++++++++++++------------------ src/index.zig | 24 ++++ 2 files changed, 248 insertions(+), 122 deletions(-) diff --git a/src/db.zig b/src/db.zig index 765d697..bf6fbf6 100644 --- a/src/db.zig +++ b/src/db.zig @@ -43,6 +43,10 @@ pub const Collection = struct { /// extent it falls in. slab_tail: u64, slab_end: u64, + /// Document bytes written into this collection's slab since the last + /// rebuild. `slab_tail` cannot answer that -- it is an absolute file offset, + /// so it jumps forward whenever a fresh extent is taken. + slab_used: u64, /// Secondary indexes (persisted through the log). Heap-allocated, so an /// `*Index` handed out by `find_index` or `create_index` stays valid when /// a sibling index is dropped. Held by value, `orderedRemove` memmoved the @@ -77,6 +81,7 @@ pub const Collection = struct { .slab_extents = .empty, .slab_tail = 0, .slab_end = 0, + .slab_used = 0, .indexes = .empty, .id_index = undefined, }; @@ -138,6 +143,7 @@ pub const Collection = struct { const off = self.slab_tail; @memcpy(self.pager.bytes_mut(off, bytes.len), bytes); self.slab_tail += bytes.len; + self.slab_used += bytes.len; return off; } @@ -1121,140 +1127,121 @@ pub const Engine = struct { /// which makes the snapshot consistent with what the log contains. /// /// Only one compaction runs at a time; a second caller returns immediately. + /// One document's new home during a rebuild: its docs-map key and the offset + /// it was copied to. Named rather than anonymous so the rebuild and the + /// repack agree on the type. + const Moved = struct { key: []const u8, off: u64 }; + + /// Reclaim what a checkpoint cannot: dead document bytes and abandoned + /// pages. + /// + /// A checkpoint publishes the structures where they already are. It cannot + /// move a document, because every index leaf holds that document's physical + /// offset -- so reclaiming a replaced document's bytes means rewriting the + /// documents *and* every index that points at them, together, which is what + /// this does. + /// + /// It used to rewrite the *log* instead: re-emit every live document into a + /// fresh log and rename it over the old one. That is now the wrong shape + /// twice over. The log is no longer where the data lives, and a re-emitted + /// record carries a sequence that a later watermark can cover, which would + /// make the next open skip it (PLAN section 4). The log is simply truncated + /// by the checkpoint at the end. pub fn compact(self: *Engine) !void { - // Claim the compaction, or leave it to the one already running. The - // guard lives here rather than in `take_compact` so that every caller - // is covered, including tests that invoke compact directly. Two at once - // would share the tmp path below: one's delete-and-recreate unlinks the - // other's file while it still holds the fd, and then both rename that - // path onto the log -- publishing a half-written file as the database. - // A caller that loses this race has nothing to do anyway: the winner's - // rewrite covers its garbage too. + // Claim it, or leave it to the one already running. A caller that loses + // this race has nothing to do: the winner's rebuild covers its garbage. if (self.compacting.swap(true, .acq_rel)) return; defer self.compacting.store(false, .release); - const tmp_path = try std.fmt.allocPrint(self.gpa, "{s}.tmp", .{self.log.path}); - defer self.gpa.free(tmp_path); - - // Bounded: every attempt rewrites the whole log before the seq check - // below can reject it, so an unbounded retry livelocks under sustained - // writes at one full rewrite per attempt. Giving up re-arms the request - // for a quieter epilogue -- the log stays correct, just larger. - const attempt_max: u32 = 8; - var attempt: u32 = 0; - while (attempt < attempt_max) : (attempt += 1) { - // Truncating create, not open: a tmp file left by a crashed or - // retried compaction is longer than what we are about to write, and - // `open` would keep its tail. Those leftover blocks are intact and - // hash-correct, so replay would apply them as live records once the - // rename publishes this file. - var new_log = try storage.Log.create(self.gpa, self.io, tmp_path); - defer new_log.close(); - - const snapshot_seq = self.seq; - try self.catalog_lock.lockShared(self.io); - var catalog_err: ?anyerror = null; - var db_it = self.dbs.iterator(); - while (db_it.next()) |db_entry| { - var coll_it = db_entry.value_ptr.collections.iterator(); - while (coll_it.next()) |coll_entry| { - const coll = coll_entry.value_ptr.*; - self.compact_snapshot_coll(coll, &new_log, db_entry.key_ptr.*, coll_entry.key_ptr.*) catch |err| { - catalog_err = err; - break; - }; - } - if (catalog_err != null) break; + try self.catalog_lock.lockShared(self.io); + var rebuild_err: ?anyerror = null; + var db_it = self.dbs.iterator(); + outer: while (db_it.next()) |db_entry| { + var coll_it = db_entry.value_ptr.collections.iterator(); + while (coll_it.next()) |coll_entry| { + self.rebuild_collection(coll_entry.value_ptr.*) catch |err| { + rebuild_err = err; + break :outer; + }; } - self.catalog_lock.unlockShared(self.io); - if (catalog_err) |err| return err; - - // Swap under the log lock, which also blocks appends: verify the - // snapshot saw no interleaved appends before replacing the file. - try self.log_lock.lock(self.io); - if (self.seq != snapshot_seq) { - self.log_lock.unlock(self.io); - continue; // a writer appended during the snapshot; retry - } - errdefer self.log_lock.unlock(self.io); - assert_msg(self.seq == snapshot_seq, "compaction snapshot raced an append"); - // Durable before the rename makes it the database. - try new_log.sync(); - // Nothing may follow the last sealed block: replay walks blocks - // until it runs off the end, so a trailing byte range would be - // applied as live data. Checked here, right where the file is about - // to become the database, rather than trusting the truncating open. - assert_msg(try new_log.file.length(self.io) == new_log.end_pos, "compacted log has bytes past its last sealed block"); - - try std.Io.Dir.renameAbsolute(tmp_path, self.log.path, self.io); - // Persist the rename: fsync the parent directory so the new - // directory entry survives a power loss right after compaction. - const parent = parent_dir(self.log.path); - var dir_file = try std.Io.Dir.cwd().openFile(self.io, parent, .{ .mode = .read_only, .allow_directory = true }); - defer dir_file.close(self.io); - try dir_file.sync(self.io); - - const old_path = try self.gpa.dupe(u8, self.log.path); - self.log.close(); - self.log = try storage.Log.open(self.gpa, self.io, old_path); - // Log.open does not replay, so it starts at an empty file's - // end_pos; continue appending where the compacted file actually - // ends. Read from new_log rather than a local captured earlier: - // the sync above is what seals the last block and moves end_pos - // past it, so a position read before it would leave appends - // overwriting that block, which then vanishes on the next replay. - self.log.end_pos = new_log.end_pos; - // The rewritten file was synced before the rename, and the check - // above proved no append slipped in, so everything up to the - // snapshot's seq is durable. - self.committed_seq = snapshot_seq; - assert_msg(self.committed_seq <= self.seq, "compaction left committed_seq past the log's seq"); - // The rewritten log holds only live documents. - self.dead_docs = 0; - self.gpa.free(old_path); - self.log_lock.unlock(self.io); - return; } + self.catalog_lock.unlockShared(self.io); + if (rebuild_err) |err| return err; - // Every attempt lost the race against a concurrent writer. Not an - // error: the log is intact and still correct, only bigger than we would - // like, so hand the request back rather than failing the write whose - // epilogue called us. - self.request_compact(); - std.debug.print( - "multiforadb: compaction gave up after {d} attempts (concurrent writes); will retry\n", - .{attempt_max}, - ); + self.dead_docs = 0; + // Publish the rebuilt layout, which is also what reclaims the log. Until + // this lands the old pages are still referenced by the previous + // watermark, so a crash mid-rebuild simply loses the rebuild. + try self.checkpoint(); + } + + /// Copy one collection's live documents into fresh extents and rebuild every + /// index against the new offsets. + /// + /// Documents and indexes have to move together: an index leaf holds a + /// physical offset, so a document that moves without its indexes being + /// rebuilt is a stale entry pointing at whatever now occupies those bytes. + fn rebuild_collection(self: *Engine, coll: *Collection) !void { + try coll.lock.lock(self.io); + defer coll.lock.unlock(self.io); + + const old_extents = try self.gpa.dupe(pgr.Extent, coll.slab_extents.items); + defer self.gpa.free(old_extents); + + // Fresh slab. The old extents stay allocated until the free list + // releases them, two generations on. + coll.slab_extents.clearRetainingCapacity(); + coll.slab_tail = 0; + coll.slab_end = 0; + coll.slab_used = 0; + + // Walk in _id order, which is also the order the new slab ends up in -- + // so a later scan reads it sequentially. + var moved: std.ArrayListUnmanaged(Moved) = .empty; + defer moved.deinit(self.gpa); + try moved.ensureTotalCapacity(self.gpa, coll.docs.count()); + + var it = coll.docs.iterator(); + while (it.next()) |entry| { + const bytes = doc_bytes_in(self.pager, entry.value_ptr.*); + try coll.slab_reserve(self.gpa, bytes.len); + const new_off = coll.slab_append(bytes); + self.pager.release_reservation(); + try moved.append(self.gpa, .{ .key = entry.key_ptr.*, .off = new_off }); + } + // Republish the offsets. + for (moved.items) |m| coll.docs.putAssumeCapacity(m.key, m.off); + + // Rebuild every index from the new offsets, bulk-packed. + try self.repack_index(coll, &coll.id_index, moved.items); + for (coll.indexes.items) |ix| try self.repack_index(coll, ix, moved.items); + + for (old_extents) |e| try self.pager.free_pages(e.first, e.pages); + } + + fn repack_index( + self: *Engine, + coll: *Collection, + ix: *index.Index, + moved: []const Moved, + ) !void { + _ = coll; + ix.reset_tree(self.gpa) catch |err| return err; + for (moved) |m| { + ix.append_doc_entries(self.gpa, doc_bytes_in(self.pager, m.off), m.off) catch |err| switch (err) { + error.ParallelArrays => continue, + else => return err, + }; + } + // Duplicates are tolerated here for the same reason they are on open: + // refusing would make a maintenance task able to take the database down. + _ = ix.finish_bulk(self.gpa, false) catch |err| return err; + self.pager.release_reservation(); } /// Re-emit one collection's index specs and documents into the compacted /// log, under the collection's write lock (released on every return /// path, including errors). - fn compact_snapshot_coll( - self: *Engine, - coll: *Collection, - new_log: *storage.Log, - db_name: []const u8, - coll_name: []const u8, - ) !void { - try coll.lock.lock(self.io); - defer coll.lock.unlock(self.io); - // Re-emit the index definitions first: a compacted log that dropped - // them would resurrect the collections without indexes on replay. - for (coll.indexes.items) |ix| { - var spec_bytes: std.ArrayListUnmanaged(u8) = .empty; - defer spec_bytes.deinit(self.gpa); - try ix.write_spec(self.gpa, &spec_bytes); - try new_log.append_index_create(db_name, coll_name, spec_bytes.items, self.seq); - } - var doc_it = coll.docs.iterator(); - while (doc_it.next()) |doc_entry| { - // The slab bytes are the canonical serialization. - const doc_bytes = coll.doc_bytes(doc_entry.value_ptr.*); - try new_log.append_upsert(db_name, coll_name, doc_bytes, self.seq); - } - } - /// Rebuild every empty index from the live documents. Runs after replay /// completes, so it is order-independent: a create record, the documents /// it indexes, and any drop record all replay first. A duplicate under a @@ -1356,6 +1343,7 @@ pub const Engine = struct { try put_bytes(gpa, out, ce.key_ptr.*); try put_u64(gpa, out, coll.slab_tail); try put_u64(gpa, out, coll.slab_end); + try put_u64(gpa, out, coll.slab_used); try put_u32(gpa, out, @intCast(coll.slab_extents.items.len)); for (coll.slab_extents.items) |e| { try put_u32(gpa, out, e.first); @@ -1429,6 +1417,7 @@ pub const Engine = struct { const coll = try self.get_or_create_collection(db_name, coll_name); coll.slab_tail = try r.read_u64(); coll.slab_end = try r.read_u64(); + coll.slab_used = try r.read_u64(); const nex = try r.read_u32(); var e: u32 = 0; while (e < nex) : (e += 1) { @@ -2619,6 +2608,98 @@ test "a checkpoint reclaims the log and the data survives" { } } +test "a rebuild reclaims dead document bytes and keeps every index valid" { + // What a checkpoint cannot do. A replaced document leaves its old bytes + // behind, and they cannot be reclaimed in place because every index leaf + // holds a physical offset -- so the rebuild has to move the documents *and* + // repack the indexes against the new offsets, together. + // + // The assertion that matters is not the size but the second half: after the + // rebuild every document is still findable through both the _id_ index and a + // secondary one. A rebuild that moved documents and left one stale entry + // behind would shrink the file and return wrong answers. + // + // Mutation checks: skip the repack and the lookups go red (stale offsets); + // skip the slab reset and the file never shrinks. + 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); + const data_path = try std.fmt.allocPrint(gpa, "{s}.data", .{tmp.path}); + defer gpa.free(data_path); + defer std.Io.Dir.cwd().deleteFile(io, data_path) catch {}; + + var engine = try Engine.open(gpa, io, tmp.path); + defer engine.deinit(); + + try engine.lock(); + var spec = try index_spec(gpa, "email", "email_1", false, false, null); + defer spec.deinit(); + _ = try engine.create_index("app", "users", &spec); + var i: i32 = 0; + while (i < 60) : (i += 1) { + var d = try make_user(gpa, i, "a@x.io"); + defer d.deinit(); + try engine.insert("app", "users", &d, &env.gen); + } + // Replace every one of them, which is what makes the old bytes garbage. + i = 0; + while (i < 60) : (i += 1) { + var d = try make_user(gpa, i, "b@x.io"); + defer d.deinit(); + try engine.replace("app", "users", &d, &env.gen); + } + engine.unlock(); + + const coll = engine.get_collection("app", "users").?; + const before_used = coll.slab_used; + try engine.compact(); + + // 60 live documents occupy less than the 120 writes that produced them. + try testing.expect(coll.slab_used < before_used); + try testing.expect(coll.slab_used > 0); + try testing.expectEqual(@as(usize, 60), coll.docs.count()); + try testing.expectEqual(@as(usize, 60), coll.id_index.count()); + + // The assertions that matter. Content alone proves nothing here: the old + // extents are only handed to the free list, not overwritten, so a stale + // offset still reads a plausible document. What distinguishes a repacked + // index from a stale one is *where* the offset points -- every live offset + // must fall inside an extent the collection currently owns. + try engine.lock(); + defer engine.unlock(); + i = 0; + while (i < 60) : (i += 1) { + const id_key = try bson.serialize_value(gpa, .{ .int32 = i }); + defer gpa.free(id_key); + + var enc: std.ArrayListUnmanaged(u8) = .empty; + defer enc.deinit(gpa); + try bson.encode_key(.{ .int32 = i }, gpa, &enc); + const from_index = coll.id_index.lookup_exact(enc.items) orelse return error.TestUnexpectedResult; + const from_map = coll.docs.get(id_key) orelse return error.TestUnexpectedResult; + + // The index and the map must agree, which catches either one being left + // behind by the rebuild. + try testing.expectEqual(from_map, from_index); + // And the offset must be in the new slab, which catches both being left + // behind together. + try testing.expect(offset_in_extents(coll, from_index)); + try testing.expect(std.mem.indexOf(u8, coll.doc_bytes(from_index), "b@x.io") != null); + } + // The secondary index too, by the same standard. + var found: std.ArrayListUnmanaged(u64) = .empty; + defer found.deinit(gpa); + const email_ix = coll.find_index("email_1").?; + try email_ix.lookup_eq(gpa, &.{.{ .string = "b@x.io" }}, &found); + try testing.expectEqual(@as(usize, 60), found.items.len); + for (found.items) |off| try testing.expect(offset_in_extents(coll, off)); +} + test "index drop survives reopen" { var threaded: std.Io.Threaded = .init_single_threaded; defer threaded.deinit(); @@ -2888,6 +2969,27 @@ test "ttl_sweep spans collections and several TTL indexes on one collection" { try testing.expectEqual(@as(usize, 1), engine.get_collection("other", "plain").?.docs.count()); } +/// Whether an offset falls inside one of the collection's current slab extents. +/// After a rebuild every live offset must, and that is what tells a repacked +/// index from one still holding pre-rebuild offsets -- the old bytes are on the +/// free list rather than overwritten, so reading them still succeeds. +fn offset_in_extents(coll: *const Collection, off: u64) bool { + for (coll.slab_extents.items) |e| { + const first = @as(u64, e.first) << pgr.page_shift; + const end = first + (@as(u64, e.pages) << pgr.page_shift); + if (off >= first and off < end) return true; + } + return false; +} + +/// Document bytes at an absolute file offset, without needing the Collection. +/// The rebuild works from offsets while the collection's own slab cursors are +/// being replaced under it. +fn doc_bytes_in(pager: *const pgr.Pager, off: u64) []const u8 { + const len: usize = std.mem.readInt(u32, pager.bytes(off, 4)[0..4], .little); + return pager.bytes(off, len); +} + // --------------------------------------------------------------------------- // Catalog encoding helpers // --------------------------------------------------------------------------- diff --git a/src/index.zig b/src/index.zig index 0be73f4..ae0acb2 100644 --- a/src/index.zig +++ b/src/index.zig @@ -579,6 +579,30 @@ pub const Index = struct { return duplicate; } + /// Throw the tree away and start from an empty root, so a rebuild can pack a + /// fresh one. The old pages go on the free list, which withholds them for two + /// generations -- the image that still references them stays intact. + pub fn reset_tree(self: *Index, gpa: std.mem.Allocator) !void { + for (self.node_pages.items) |p| try self.pager.free_pages(p, 1); + for (self.ovf_extents.items) |e| try self.pager.free_pages(e.first, e.pages); + self.node_pages.clearRetainingCapacity(); + self.ovf_extents.clearRetainingCapacity(); + self.ovf_tail = 0; + self.ovf_end = 0; + try self.node_pages.ensureUnusedCapacity(gpa, 2); + try self.pager.reserve_pages(2); + self.node_pages.appendAssumeCapacity(self.pager.alloc_pages_assume_reserved(1)); + self.node_pages.appendAssumeCapacity(self.pager.alloc_pages_assume_reserved(1)); + self.page_mut(0).* = empty_node(0); + self.page_mut(1).* = empty_node(1); + self.root = 1; + self.first_leaf = 1; + self.leaf_count = 1; + self.depth = 0; + self.entry_count = 0; + self.multikey = false; + } + /// Remove every entry for `id`, in one pass over the leaves. Infallible. /// Used directly by remove_id's own callers and as the fallback when /// regeneration cannot locate entries.