dev/MultiforaDB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

db: compaction becomes a data-file rebuild

Browse Source 
`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.
This commit is contained in:
Alexey Shakhmatov
2026-08-03 21:48:04 +03:00
parent 138b7f706f
commit 148e03ac9f
2 changed files with 248 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files

346
src/db.zig
View File

@@ -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
// ---------------------------------------------------------------------------

24
src/index.zig
View File

@@ -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.