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

engine: decompose the global lock; cross-connection group commit (roadmap item 5)

Browse Source 
The single engine-wide reader/writer lock is replaced by a lock hierarchy,
so writes to different collections no longer serialize on one mutex:

- Collections are heap-allocated, so their addresses are stable while a
  command holds a collection lock (the maps only store pointers).
- A catalog rwlock guards the database/collection maps: shared for every
  command (so a concurrent DDL cannot mutate the maps underneath it),
  exclusive for create/drop/dropDatabase. Each collection has its own
  rwlock; the ordering is always catalog -> collection -> log lock, never
  two collection locks at once (TTL sweep and compaction take collections
  one at a time).
- Command dispatch acquires the catalog + target collection locks for the
  handler's duration, resolving the collection (creating it for writes)
  under the catalog lock; create/drop upgrade to the exclusive catalog lock.
- Appends never fsync. Each write command's epilogue releases the
  collection lock, then commits once (seal + fsync) with a leader/follower
  group commit: the leader waits for writers mid-append (a pending counter)
  so its seal covers them, and followers whose records the seal covered
  skip their own fsync. Every acknowledged write is fsynced before its
  reply (crash pair verified); an unacknowledged write may vanish and a
  reader may observe a write before its fsync — ordinary w:1 j:true
  semantics instead of 'the log describes >= memory'.
- Compaction snapshots collections without the log lock (so a concurrent
  writer holding one can always finish its append) and retries when a
  writer appended mid-snapshot (detected via the record seq), then swaps
  under the log lock — no deadlock. The compaction trigger moved to the
  command epilogue and the TTL monitor.
- Engine.dup_index moved to the collection (per-command error paths).

Also lands two B-tree edge-case fixes driven by tests that were in flight:
a churned leaf full of dead bytes no longer splits with an empty right
half (the leaf is repacked before splitting, and an emptied node's page is
fully free again), and a slot-count split with all large records on one
side shifts records between the halves until the new record fits. Plus a
randomised fuzz test over key sizes (src/fuzz_split.zig) and the two
regression tests.

Measured (tests/e2e/results/phase6.txt): no regression on the
single-connection benchmark; concurrent durable-insert throughput ~5.1k ->
12.5k docs/s from 1 -> 8 clients, ~14.8k at 32. Verified: unit suite in
all three modes, all e2e suites, the kill -9 crash pair.
This commit is contained in:
Alexey Shakhmatov
2026-08-02 23:26:24 +03:00
parent 570900a6ef
commit ecd28d9b26
9 changed files with 999 additions and 302 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
README.md
View File

@@ -224,10 +224,10 @@ storage (roadmap items 1–4) in `tests/e2e/results/phase2.txt` through
Each is written up with its design decisions, ordering constraints and Each is written up with its design decisions, ordering constraints and
traps in [ROADMAP.md](ROADMAP.md). traps in [ROADMAP.md](ROADMAP.md).
1. **Decompose the global lock** — one reader/writer lock covers the whole The remaining structure is the single-file log (appends and the commit
engine and is held across fsync, compaction and reply construction. serialize on one log lock, though appends no longer hold the collection
Per-collection locks plus cross-connection group commit are the path to locks), and the acknowledged-write fsync, which dominates sequential
using more than one core on writes. per-client workloads. All five roadmap items are landed.
Done so far, with the measurement that drove each: Done so far, with the measurement that drove each:
@@ -276,6 +276,16 @@ Done so far, with the measurement that drove each:
use a borrowed spine into the slab. `server RSS` 1979 → 539 MB (2.4x use a borrowed spine into the slab. `server RSS` 1979 → 539 MB (2.4x
smaller than MongoDB); `range-scan` 22.5 → ~12 ms (parity, best run smaller than MongoDB); `range-scan` 22.5 → ~12 ms (parity, best run
faster); `proj` 4.1 → 3.4 ms. faster); `proj` 4.1 → 3.4 ms.
- **Decomposed locks** (roadmap item 5): collections are heap-allocated;
a catalog rwlock guards the maps and each collection has its own rwlock
(catalog → collection → log ordering, one collection at a time for the
TTL sweep and compaction). Appends never fsync; a write command's
epilogue commits once with a leader/follower group commit, and
compaction snapshots collections without the log lock, retrying if a
writer appended mid-snapshot. Acknowledged writes are fsynced before
their reply; an unacknowledged write may vanish (ordinary `w:1, j:true`,
no longer "the log describes ≥ memory"). Concurrent durable-insert
throughput scales ~5.1k → 12.5k docs/s from 1 → 8 clients, ~14.8k at 32.
- **Entry removal is a binary search**, not a scan of the whole index. - **Entry removal is a binary search**, not a scan of the whole index.
`updateMany` 15.4 → 5.5 ms. `updateMany` 15.4 → 5.5 ms.
- **Top-k sort selection** and an allocation-free decorate pass, plus - **Top-k sort selection** and an allocation-free decorate pass, plus

43
ROADMAP.md
View File

@@ -1,14 +1,13 @@
# Remaining performance work # Remaining performance work
Status: **items 1 (B+tree), 2 (ordered `_id` index), 3 (block-framed Status: **all five items are done** — verified in `tests/e2e/results/phase2.txt`
compressed log) and 4 (byte storage) are done** — verified in through `phase6.txt`: updateMany 17.3 → 1.7 ms, createIndex 62 → 51 ms,
`tests/e2e/results/phase2.txt` through `phase5.txt`: updateMany 17.3 → `_id` sort+limit 6.2 → 2.4 ms, `db on disk` 1025 → 97 MB, `server RSS`
1.7 ms, createIndex 62 → 51 ms, `_id` sort+limit 6.2 → 2.4 ms, `db on 1979 → 539 MB with the range scan at parity (best run faster than
disk` 1025 → 97 MB, and `server RSS` 1979 → 539 MB with the range scan at MongoDB), and the global lock decomposed into catalog + per-collection
parity (best run faster than MongoDB). Only item 5 (decompose the global locks with cross-connection group commit (item 5, no regression on the
lock) remains. Each is sized to be landed and verified on single-connection benchmark; concurrent-write throughput 1 → 8 clients
its own; the ordering constraints between them are the load-bearing part, so ~5.1k → 12.5k docs/s, 32 clients ~14.8k).
read those before picking one up.
Current numbers and what they mean are in the README; the recorded baseline Current numbers and what they mean are in the README; the recorded baseline
is `tests/e2e/results/phase1.txt`, reproduced with is `tests/e2e/results/phase1.txt`, reproduced with
@@ -254,7 +253,31 @@ when `Document` changes meaning.
--- ---
## 5. Decompose the global lock ## 5. Decompose the global lock — DONE
Landed: collections are heap-allocated (stable pointers; the map only holds
them), a catalog rwlock guards the database/collection maps (shared for
commands, exclusive for create/drop), and one rwlock per collection guards
its docs/slab/indexes, with the catalog → collection → log-lock ordering and
never two collection locks at once (TTL sweep and compaction take
collections one at a time). Appends never fsync; each write command's
epilogue commits once (seal + fsync) under a leader/follower group commit —
the leader waits for writers mid-append (a pending counter) so its seal
covers them, and followers whose records the seal covered skip their own
fsync. `Engine.dup_index` moved per-collection. Compaction snapshots the
collections without the log lock and retries if a writer appended during
the snapshot (detected via the record seq), then swaps under the log lock —
no deadlock against a writer holding a collection lock. The durability
guarantee weakened from "the log always describes >= memory" to ordinary
`w:1, j:true`: an acknowledged write is fsynced before its reply (the crash
pair verifies it), an unacknowledged write may vanish, and a reader can
observe a write before its fsync completes.
Measured: no regression on the single-connection benchmark (phase6);
concurrent durable-insert throughput scales ~5.1k → 12.5k docs/s from 1 →
8 clients and ~14.8k at 32 — the fsync per commit still dominates
sequential-per-client workloads, and the group commit coalesces when
appends from different collections overlap.
**Why.** One reader/writer lock covers the entire engine and is held across **Why.** One reader/writer lock covers the entire engine and is held across
fsync, compaction and reply construction, so writes cannot use more than one fsync, compaction and reply construction, so writes cannot use more than one

116
src/commands.zig
View File

@@ -44,9 +44,20 @@ pub const ErrorCode = enum(i32) {
/// `.none` commands must not touch the engine at all. /// `.none` commands must not touch the engine at all.
const CommandKind = enum { none, read, write }; const CommandKind = enum { none, read, write };
/// Lock shape for one command, acquired by dispatch: the catalog lock mode
/// and whether the command's target collection is locked (shared for reads,
/// exclusive for writes/DDL). The target collection is the message field
/// named after the command (find/count/insert/...), which every
/// collection-targeting command uses.
const LockShape = struct {
catalog: enum { none, shared, exclusive } = .none,
coll: enum { none, shared, exclusive } = .none,
};
const Command = struct { const Command = struct {
name: []const u8, name: []const u8,
kind: CommandKind, kind: CommandKind,
locks: LockShape = .{},
handler: *const fn (*Context, *wire.Message, *wire.Reply) anyerror!void, handler: *const fn (*Context, *wire.Message, *wire.Reply) anyerror!void,
}; };
@@ -69,22 +80,23 @@ const command_table = [_]Command{
.{ .name = "getMore", .kind = .none, .handler = cmd_get_more }, .{ .name = "getMore", .kind = .none, .handler = cmd_get_more },
.{ .name = "killCursors", .kind = .none, .handler = cmd_kill_cursors }, .{ .name = "killCursors", .kind = .none, .handler = cmd_kill_cursors },
// Read-only: scan the engine without mutating it. // Read-only: scan the engine without mutating it.
.{ .name = "find", .kind = .read, .handler = cmd_find }, .{ .name = "find", .kind = .read, .locks = .{ .catalog = .shared, .coll = .shared }, .handler = cmd_find },
.{ .name = "count", .kind = .read, .handler = cmd_count }, .{ .name = "count", .kind = .read, .locks = .{ .catalog = .shared, .coll = .shared }, .handler = cmd_count },
.{ .name = "aggregate", .kind = .read, .handler = cmd_aggregate }, .{ .name = "aggregate", .kind = .read, .locks = .{ .catalog = .shared, .coll = .shared }, .handler = cmd_aggregate },
.{ .name = "listDatabases", .kind = .read, .handler = cmd_list_databases }, .{ .name = "listDatabases", .kind = .read, .locks = .{ .catalog = .shared }, .handler = cmd_list_databases },
.{ .name = "listCollections", .kind = .read, .handler = cmd_list_collections }, .{ .name = "listCollections", .kind = .read, .locks = .{ .catalog = .shared }, .handler = cmd_list_collections },
// Writes: exclusive, totally ordered. // Writes: the target collection exclusively; create/drop take the
.{ .name = "create", .kind = .write, .handler = cmd_create }, // catalog exclusively (they mutate the maps).
.{ .name = "drop", .kind = .write, .handler = cmd_drop }, .{ .name = "create", .kind = .write, .locks = .{ .catalog = .exclusive, .coll = .exclusive }, .handler = cmd_create },
.{ .name = "dropDatabase", .kind = .write, .handler = cmd_drop_database }, .{ .name = "drop", .kind = .write, .locks = .{ .catalog = .exclusive, .coll = .exclusive }, .handler = cmd_drop },
.{ .name = "createIndexes", .kind = .write, .handler = cmd_create_indexes }, .{ .name = "dropDatabase", .kind = .write, .locks = .{ .catalog = .exclusive }, .handler = cmd_drop_database },
.{ .name = "dropIndexes", .kind = .write, .handler = cmd_drop_indexes }, .{ .name = "createIndexes", .kind = .write, .locks = .{ .catalog = .shared, .coll = .exclusive }, .handler = cmd_create_indexes },
.{ .name = "insert", .kind = .write, .handler = cmd_insert }, .{ .name = "dropIndexes", .kind = .write, .locks = .{ .catalog = .shared, .coll = .exclusive }, .handler = cmd_drop_indexes },
.{ .name = "update", .kind = .write, .handler = cmd_update }, .{ .name = "insert", .kind = .write, .locks = .{ .catalog = .shared, .coll = .exclusive }, .handler = cmd_insert },
.{ .name = "delete", .kind = .write, .handler = cmd_delete }, .{ .name = "update", .kind = .write, .locks = .{ .catalog = .shared, .coll = .exclusive }, .handler = cmd_update },
.{ .name = "findAndModify", .kind = .write, .handler = cmd_find_and_modify }, .{ .name = "delete", .kind = .write, .locks = .{ .catalog = .shared, .coll = .exclusive }, .handler = cmd_delete },
.{ .name = "listIndexes", .kind = .read, .handler = cmd_list_indexes }, .{ .name = "findAndModify", .kind = .write, .locks = .{ .catalog = .shared, .coll = .exclusive }, .handler = cmd_find_and_modify },
.{ .name = "listIndexes", .kind = .read, .locks = .{ .catalog = .shared, .coll = .shared }, .handler = cmd_list_indexes },
}; };
/// Command name to its index in `command_table`, resolved at comptime so a /// Command name to its index in `command_table`, resolved at comptime so a
@@ -104,22 +116,48 @@ pub fn dispatch(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
return reply.put_error(@intFromEnum(ErrorCode.command_not_found), "CommandNotFound", errmsg); return reply.put_error(@intFromEnum(ErrorCode.command_not_found), "CommandNotFound", errmsg);
}; };
switch (cmd.kind) { // Lock the catalog (shared for most commands, exclusive for DDL), then
.none => return cmd.handler(ctx, msg, reply), // the target collection, then run the handler. The collection lock is
.read => { // taken while the catalog lock is held, so a concurrent drop can never
try ctx.engine.lock_read(); // free the collection out from under us.
// Defers are block-scoped: this one is registered in the prong switch (cmd.locks.catalog) {
// block, so it runs when the prong exits — after the handler .none => {},
// returns. The shared lock is thus held for the whole command. .shared => try ctx.engine.lock_catalog(false),
defer ctx.engine.unlock_read(); .exclusive => try ctx.engine.lock_catalog(true),
return cmd.handler(ctx, msg, reply);
},
.write => {
try ctx.engine.lock();
defer ctx.engine.unlock();
return cmd.handler(ctx, msg, reply);
},
} }
var catalog_held = cmd.locks.catalog != .none;
var coll: ?*Collection = null;
errdefer {
if (coll) |c| ctx.engine.unlock_collection(c, cmd.locks.coll == .exclusive);
if (catalog_held) ctx.engine.unlock_catalog(cmd.locks.catalog == .exclusive);
}
if (cmd.locks.coll != .none) {
const db_name = msg.db_name() orelse return;
const coll_name = str_arg(msg.body.get(name)) orelse return;
// Write commands may create the collection on first use; the catalog
// lock is upgraded to exclusive for that, then restored to shared.
const create = cmd.kind == .write and cmd.locks.catalog == .shared;
if (try ctx.engine.lock_collection(db_name, coll_name, cmd.locks.coll == .exclusive, create)) |c| {
coll = c;
}
}
const result = cmd.handler(ctx, msg, reply);
// Release the collection and catalog locks before the commit: the
// commit may block on other writers' appends, and must never do so
// while holding a collection lock.
if (coll) |c| ctx.engine.unlock_collection(c, cmd.locks.coll == .exclusive);
coll = null;
if (catalog_held) ctx.engine.unlock_catalog(cmd.locks.catalog == .exclusive);
catalog_held = false;
if (cmd.locks.coll == .exclusive) {
// Durability (seal + fsync) coalesces across concurrent writers.
try ctx.engine.commit();
if (ctx.engine.take_compact()) try ctx.engine.compact();
}
return result;
} }
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
@@ -1423,18 +1461,22 @@ fn bad_value(reply: *wire.Reply, msg: []const u8) !void {
} }
/// The E11000 text, shared by the top-level error reply and the per-document /// The E11000 text, shared by the top-level error reply and the per-document
/// `writeErrors` entries of a batch insert. Uses engine.dup_index (set by a /// `writeErrors` entries of a batch insert. Uses the collection's
/// rejected unique-index write) when the conflict came from a secondary /// dup_index (set by a rejected unique-index write) when the conflict came
/// index; otherwise it is the _id_ index. /// from a secondary index; otherwise it is the _id_ index. Per-collection
/// so concurrent writers on other collections cannot clobber it.
fn duplicate_key_message(ctx: *Context, reply: *wire.Reply, db_name: []const u8, coll_name: []const u8, doc: *const bson.Document) ![]const u8 { fn duplicate_key_message(ctx: *Context, reply: *wire.Reply, db_name: []const u8, coll_name: []const u8, doc: *const bson.Document) ![]const u8 {
var index_name: []const u8 = "_id_"; var index_name: []const u8 = "_id_";
var key_text: []const u8 = undefined; var key_text: []const u8 = undefined;
if (ctx.engine.dup_index) |name| { const coll = ctx.engine.get_collection(db_name, coll_name);
if (coll) |c| {
if (c.dup_index) |name| {
index_name = name; index_name = name;
key_text = try render_dup_key(ctx, reply, db_name, coll_name, name, doc); key_text = try render_dup_key(ctx, reply, db_name, coll_name, name, doc);
} else { return e11000_message(reply, db_name, coll_name, index_name, key_text);
key_text = try serialize_value_compact(reply, doc.get("_id") orelse bson.Value.null); }
} }
key_text = try serialize_value_compact(reply, doc.get("_id") orelse bson.Value.null);
return e11000_message(reply, db_name, coll_name, index_name, key_text); return e11000_message(reply, db_name, coll_name, index_name, key_text);
} }

497
src/db.zig
View File

@@ -14,6 +14,8 @@ const index = @import("index.zig");
/// would hold up to 2x its contents after doubling). /// would hold up to 2x its contents after doubling).
const slab_segment_size = 8 * 1024 * 1024; const slab_segment_size = 8 * 1024 * 1024;
const LogKind = enum { upsert, delete, index_create, index_drop };
pub const Collection = struct { pub const Collection = struct {
/// Documents live as canonical BSON bytes in a per-collection slab of /// Documents live as canonical BSON bytes in a per-collection slab of
/// fixed segments; the map holds each document's flat slab offset. /// fixed segments; the map holds each document's flat slab offset.
@@ -29,6 +31,14 @@ pub const Collection = struct {
seg_starts: std.ArrayListUnmanaged(u64), seg_starts: std.ArrayListUnmanaged(u64),
/// Secondary indexes (persisted through the log). /// Secondary indexes (persisted through the log).
indexes: std.ArrayListUnmanaged(index.Index), indexes: std.ArrayListUnmanaged(index.Index),
/// Guards this collection's docs/slab/indexes. Writers take it
/// exclusive, readers shared; never held while taking the catalog lock,
/// and never more than one collection lock at a time.
lock: std.Io.RwLock = .init,
/// The secondary index that rejected the most recent unique write
/// (duplicate-key error path); per-collection so concurrent writers on
/// other collections cannot clobber it mid-command.
dup_index: ?[]const u8 = null,
/// The implicit _id_ index: every document has an _id and it is not /// The implicit _id_ index: every document has an _id and it is not
/// sparse, so entry count equals document count and a full scan of it /// sparse, so entry count equals document count and a full scan of it
/// cannot miss a document — which is what the sort planner's full-scan /// cannot miss a document — which is what the sort planner's full-scan
@@ -104,16 +114,40 @@ pub const Collection = struct {
}; };
pub const Db = struct { pub const Db = struct {
collections: std.StringHashMapUnmanaged(Collection), /// Collections are heap-allocated so their addresses are stable while a
/// command holds a collection lock — the map may reallocate under the
/// catalog lock, but the pointers it holds do not move.
collections: std.StringHashMapUnmanaged(*Collection),
}; };
pub const Engine = struct { pub const Engine = struct {
gpa: std.mem.Allocator, gpa: std.mem.Allocator,
io: std.Io, io: std.Io,
// One writer at a time (log append + fsync, map mutation); many // Legacy whole-engine lock, used by the unit tests' explicit
// concurrent readers (find/count/aggregate scans). Writer-preferring: // lock()/lock_read() calls. The server uses the finer-grained locks
// a queued writer blocks new readers rather than starving. // below: catalog (maps), per-collection (docs/slab/indexes), and
// log_lock (append + commit).
rwlock: std.Io.RwLock, rwlock: std.Io.RwLock,
/// Guards the dbs/collections maps. Commands hold it shared for their
/// whole duration so a concurrent DDL cannot mutate the maps under
/// them; DDL takes it exclusive.
catalog_lock: std.Io.RwLock = .init,
/// Serializes log appends, seals and the commit sync.
log_lock: std.Io.Mutex = .init,
/// Serializes commit decisions; the group-commit leader holds it while
/// sealing and syncing.
commit_lock: std.Io.Mutex = .init,
/// Log end position covered by the last completed commit.
committed_end: u64 = 0,
/// Writers increment before appending and decrement after; the commit
/// leader waits for this to reach zero so its seal covers every append
/// in flight, coalescing many writers' fsyncs into one.
pending_appends: std.atomic.Value(u32) = std.atomic.Value(u32).init(0),
committing: bool = false,
commit_done: std.Io.Condition = std.Io.Condition.init,
/// Set when the garbage ratio crosses the compaction threshold; the
/// write command's epilogue runs compact after releasing its locks.
compact_pending: bool = false,
log: storage.Log, log: storage.Log,
dbs: std.StringHashMapUnmanaged(Db), dbs: std.StringHashMapUnmanaged(Db),
seq: u64, seq: u64,
@@ -181,13 +215,14 @@ pub const Engine = struct {
for (coll.slab.items) |*seg| seg.deinit(self.gpa); for (coll.slab.items) |*seg| seg.deinit(self.gpa);
coll.slab.deinit(self.gpa); coll.slab.deinit(self.gpa);
coll.seg_starts.deinit(self.gpa); coll.seg_starts.deinit(self.gpa);
self.gpa.destroy(coll);
} }
/// Free every collection in a database along with its owned name keys. /// Free every collection in a database along with its owned name keys.
fn free_db(self: *Engine, db: *Db) void { fn free_db(self: *Engine, db: *Db) void {
var coll_it = db.collections.iterator(); var coll_it = db.collections.iterator();
while (coll_it.next()) |coll_entry| { while (coll_it.next()) |coll_entry| {
self.free_collection(coll_entry.value_ptr); self.free_collection(coll_entry.value_ptr.*);
self.gpa.free(coll_entry.key_ptr.*); self.gpa.free(coll_entry.key_ptr.*);
} }
db.collections.deinit(self.gpa); db.collections.deinit(self.gpa);
@@ -235,19 +270,121 @@ pub const Engine = struct {
self.rwlock.unlockShared(self.io); self.rwlock.unlockShared(self.io);
} }
/// Group commit: defer per-record fsyncs until end_batch. Callers must // -- per-collection locking (the server's command dispatch) ------------
/// hold the write lock and pair every begin with an end (the command's
/// defer). Every document published before end_batch is fsynced by it, /// Lock the catalog for a command's duration.
/// so an acknowledged multi-write command is durable as a unit — the pub fn lock_catalog(self: *Engine, exclusive: bool) !void {
/// same crash guarantee as the old fsync-per-record, with one sync per if (exclusive) {
/// command instead of one per document. try self.catalog_lock.lock(self.io);
pub fn begin_batch(self: *Engine) void { } else {
self.log.defer_sync = true; try self.catalog_lock.lockShared(self.io);
}
} }
pub fn end_batch(self: *Engine) !void { pub fn unlock_catalog(self: *Engine, exclusive: bool) void {
self.log.defer_sync = false; if (exclusive) self.catalog_lock.unlock(self.io) else self.catalog_lock.unlockShared(self.io);
}
/// With the catalog lock held, resolve the target collection and take
/// its lock. When the collection is missing and `create` is set, the
/// catalog lock is upgraded to exclusive to create it (then restored to
/// shared); the collection lock is acquired before the exclusive catalog
/// lock is dropped, so a concurrent drop can never free it underneath.
/// Returns null when the collection does not exist (and create is off).
pub fn lock_collection(self: *Engine, db_name: []const u8, coll_name: []const u8, write: bool, create: bool) !?*Collection {
var coll = self.get_collection(db_name, coll_name);
if (coll == null and create) {
self.catalog_lock.unlockShared(self.io);
try self.catalog_lock.lock(self.io);
coll = try self.get_or_create_collection(db_name, coll_name);
try self.lock_one(coll.?, write);
self.catalog_lock.unlock(self.io);
try self.catalog_lock.lockShared(self.io);
return coll;
}
if (coll) |c| try self.lock_one(c, write);
return coll;
}
fn lock_one(self: *Engine, coll: *Collection, write: bool) !void {
if (write) {
try coll.lock.lock(self.io);
} else {
try coll.lock.lockShared(self.io);
}
}
pub fn unlock_collection(self: *Engine, coll: *Collection, write: bool) void {
if (write) coll.lock.unlock(self.io) else coll.lock.unlockShared(self.io);
}
/// Ensure this command's appends are durable. The commit leader waits
/// for writers mid-append to finish, then seals and syncs once, covering
/// every append in flight — followers that arrived during the leader's
/// commit find their records already covered and return without a sync
/// of their own. Every acknowledged write is fsynced before its reply,
/// so the crash guarantees are unchanged.
pub fn commit(self: *Engine) !void {
try self.commit_lock.lock(self.io);
defer self.commit_lock.unlock(self.io);
// A commit is in flight; wait for it, then check whether the
// leader's seal covered this writer's append.
while (self.committing) self.commit_done.wait(self.io, &self.commit_lock) catch return;
try self.log_lock.lock(self.io);
const covered = self.log.end_pos == self.committed_end;
self.log_lock.unlock(self.io);
if (covered) {
return; // a concurrent commit already synced this writer's records
}
// Become the leader: the flag is set before the wait below, so any
// commit that arrives during it waits as a follower.
self.committing = true;
var done = false;
defer {
if (!done) {
self.committing = false;
self.commit_done.signal(self.io);
}
}
// Wait for writers mid-append to finish so the seal covers them.
while (self.pending_appends.load(.acquire) > 0) self.commit_done.wait(self.io, &self.commit_lock) catch return;
try self.log_lock.lock(self.io);
defer self.log_lock.unlock(self.io);
try self.log.sync(); try self.log.sync();
self.committed_end = self.log.end_pos;
done = true;
self.committing = false;
self.commit_done.signal(self.io);
}
/// Log an append (and its seq increment) under the log lock, marking
/// the append as in flight so a commit leader's seal covers it.
fn log_append(self: *Engine, comptime kind: LogKind, db: []const u8, coll: []const u8, doc: []const u8) !void {
_ = self.pending_appends.fetchAdd(1, .acq_rel);
defer {
_ = self.pending_appends.fetchSub(1, .acq_rel);
// Wake a commit leader waiting for in-flight appends.
self.commit_done.signal(self.io);
}
try self.log_lock.lock(self.io);
defer self.log_lock.unlock(self.io);
self.seq += 1;
switch (kind) {
.upsert => try self.log.append_upsert(db, coll, doc, self.seq),
.delete => try self.log.append_delete(db, coll, doc, self.seq),
.index_create => try self.log.append_index_create(db, coll, doc, self.seq),
.index_drop => try self.log.append_index_drop(db, coll, doc, self.seq),
}
}
/// Group commit within a command: every write command's appends are
/// deferred anyway, and the command epilogue (dispatch or a direct
/// caller) commits once — so these are no-ops kept for the batch
/// commands' pairing.
pub fn begin_batch(self: *Engine) void { _ = self; }
pub fn end_batch(self: *Engine) !void {
try self.commit();
} }
/// Insert a document. Fails with error.DuplicateKey if the _id exists. /// Insert a document. Fails with error.DuplicateKey if the _id exists.
@@ -291,7 +428,7 @@ pub const Engine = struct {
const id_key = try bson.serialize_value(self.gpa, id_value); const id_key = try bson.serialize_value(self.gpa, id_value);
var stored = false; var stored = false;
errdefer if (!stored) self.gpa.free(id_key); errdefer if (!stored) self.gpa.free(id_key);
self.dup_index = null; coll.dup_index = null;
// 1. Build entries for every index. ParallelArrays escapes here, // 1. Build entries for every index. ParallelArrays escapes here,
// before anything is logged or mutated. // before anything is logged or mutated.
@@ -325,7 +462,7 @@ pub const Engine = struct {
for (built_list.items) |*b| { for (built_list.items) |*b| {
if (!b.ix.unique) continue; if (!b.ix.unique) continue;
b.ix.check_unique(b.built.entries.items, id_key) catch { b.ix.check_unique(b.built.entries.items, id_key) catch {
self.dup_index = b.ix.name; coll.dup_index = b.ix.name;
return error.DuplicateKeyIndex; return error.DuplicateKeyIndex;
}; };
} }
@@ -336,9 +473,9 @@ pub const Engine = struct {
try b.ix.reserve_for(self.gpa, b.built.entries.items); try b.ix.reserve_for(self.gpa, b.built.entries.items);
} }
// 5. Log (and sync) before anything becomes visible. // 5. Log (and sync) before anything becomes visible. The append
self.seq += 1; // takes the log lock; durability (fsync) is the command's commit.
try self.log.append_upsert(db_name, coll_name, doc_bytes, self.seq); try self.log_append(.upsert, db_name, coll_name, doc_bytes);
// 6. Replace drops the old document (and its index entries). // 6. Replace drops the old document (and its index entries).
if (mode == .replace) self.evict_doc(coll, id_key); if (mode == .replace) self.evict_doc(coll, id_key);
@@ -353,7 +490,7 @@ pub const Engine = struct {
b.ix.insert_entries(&b.built); b.ix.insert_entries(&b.built);
} }
stored = true; stored = true;
try self.maybe_compact(); self.note_compact();
} }
/// Remove a document by its `_id` value. Returns true if it existed. /// Remove a document by its `_id` value. Returns true if it existed.
@@ -366,7 +503,7 @@ pub const Engine = struct {
fn remove(self: *Engine, db_name: []const u8, coll_name: []const u8, id_key: []const u8) !bool { fn remove(self: *Engine, db_name: []const u8, coll_name: []const u8, id_key: []const u8) !bool {
const db = self.dbs.get(db_name) orelse return false; const db = self.dbs.get(db_name) orelse return false;
const coll = db.collections.getPtr(coll_name) orelse return false; const coll = db.collections.get(coll_name) orelse return false;
const off = coll.docs.get(id_key) orelse return false; const off = coll.docs.get(id_key) orelse return false;
// Log (and sync) the delete before removing it from memory, so the // Log (and sync) the delete before removing it from memory, so the
@@ -380,19 +517,18 @@ pub const Engine = struct {
var id_doc: std.ArrayListUnmanaged(u8) = .empty; var id_doc: std.ArrayListUnmanaged(u8) = .empty;
defer id_doc.deinit(self.gpa); defer id_doc.deinit(self.gpa);
try bson.write_doc(&id_pairs, self.gpa, &id_doc); try bson.write_doc(&id_pairs, self.gpa, &id_doc);
self.seq += 1; try self.log_append(.delete, db_name, coll_name, id_doc.items);
try self.log.append_delete(db_name, coll_name, id_doc.items, self.seq);
self.evict_doc(coll, id_key); self.evict_doc(coll, id_key);
// Deletes grow the log too. Without this a delete-heavy workload // Deletes grow the log too. Without this a delete-heavy workload
// never compacts, because only upsert and ttl_sweep used to check. // never compacts, because only upsert and ttl_sweep used to check.
try self.maybe_compact(); self.note_compact();
return true; return true;
} }
pub fn get_collection(self: *Engine, db_name: []const u8, coll_name: []const u8) ?*Collection { pub fn get_collection(self: *Engine, db_name: []const u8, coll_name: []const u8) ?*Collection {
const db = self.dbs.get(db_name) orelse return null; const db = self.dbs.get(db_name) orelse return null;
return db.collections.getPtr(coll_name); return db.collections.get(coll_name);
} }
pub fn get_doc(self: *Engine, db_name: []const u8, coll_name: []const u8, id_key: []const u8) ?[]const u8 { pub fn get_doc(self: *Engine, db_name: []const u8, coll_name: []const u8, id_key: []const u8) ?[]const u8 {
@@ -403,8 +539,8 @@ pub const Engine = struct {
pub fn drop_collection(self: *Engine, db_name: []const u8, coll_name: []const u8) !bool { pub fn drop_collection(self: *Engine, db_name: []const u8, coll_name: []const u8) !bool {
const db = self.dbs.getPtr(db_name) orelse return false; const db = self.dbs.getPtr(db_name) orelse return false;
var removed = db.collections.fetchRemove(coll_name) orelse return false; const removed = db.collections.fetchRemove(coll_name) orelse return false;
self.free_collection(&removed.value); self.free_collection(removed.value);
self.gpa.free(removed.key); self.gpa.free(removed.key);
return true; return true;
} }
@@ -451,8 +587,7 @@ pub const Engine = struct {
var spec_bytes: std.ArrayListUnmanaged(u8) = .empty; var spec_bytes: std.ArrayListUnmanaged(u8) = .empty;
defer spec_bytes.deinit(self.gpa); defer spec_bytes.deinit(self.gpa);
try ix.write_spec(self.gpa, &spec_bytes); try ix.write_spec(self.gpa, &spec_bytes);
self.seq += 1; try self.log_append(.index_create, db_name, coll_name, spec_bytes.items);
try self.log.append_index_create(db_name, coll_name, spec_bytes.items, self.seq);
coll.indexes.appendAssumeCapacity(ix); coll.indexes.appendAssumeCapacity(ix);
committed = true; committed = true;
@@ -463,15 +598,14 @@ pub const Engine = struct {
/// Returns false when no such index exists. /// Returns false when no such index exists.
pub fn drop_index(self: *Engine, db_name: []const u8, coll_name: []const u8, index_name: []const u8) !bool { pub fn drop_index(self: *Engine, db_name: []const u8, coll_name: []const u8, index_name: []const u8) !bool {
const db = self.dbs.get(db_name) orelse return false; const db = self.dbs.get(db_name) orelse return false;
const coll = db.collections.getPtr(coll_name) orelse return false; const coll = db.collections.get(coll_name) orelse return false;
if (coll.find_index(index_name) == null) return false; if (coll.find_index(index_name) == null) return false;
const name_pairs = [_]bson.Pair{.{ .key = "name", .value = .{ .string = index_name } }}; const name_pairs = [_]bson.Pair{.{ .key = "name", .value = .{ .string = index_name } }};
var name_doc: std.ArrayListUnmanaged(u8) = .empty; var name_doc: std.ArrayListUnmanaged(u8) = .empty;
defer name_doc.deinit(self.gpa); defer name_doc.deinit(self.gpa);
try bson.write_doc(&name_pairs, self.gpa, &name_doc); try bson.write_doc(&name_pairs, self.gpa, &name_doc);
self.seq += 1; try self.log_append(.index_drop, db_name, coll_name, name_doc.items);
try self.log.append_index_drop(db_name, coll_name, name_doc.items, self.seq);
_ = coll.remove_index(self.gpa, index_name); _ = coll.remove_index(self.gpa, index_name);
return true; return true;
@@ -488,6 +622,30 @@ pub const Engine = struct {
/// the next sweep. /// the next sweep.
pub fn ttl_sweep(self: *Engine, now_ms: i64) !usize { pub fn ttl_sweep(self: *Engine, now_ms: i64) !usize {
var deleted: usize = 0; var deleted: usize = 0;
// Catalog lock for the whole sweep (the collection pointers stay
// valid); each collection is swept under its own write lock, one at
// a time, never two at once.
try self.catalog_lock.lockShared(self.io);
defer self.catalog_lock.unlockShared(self.io);
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.*;
deleted += try self.ttl_sweep_coll(coll, now_ms, db_entry.key_ptr.*, coll_entry.key_ptr.*);
}
}
// A TTL-only workload never reaches the threshold check in `upsert`,
// so the log would otherwise grow without bound.
if (deleted > 0) self.note_compact();
return deleted;
}
/// Sweep one collection under its write lock; the lock is released on
/// every return path. Returns how many documents were removed.
fn ttl_sweep_coll(self: *Engine, coll: *Collection, now_ms: i64, db_name: []const u8, coll_name: []const u8) !usize {
try coll.lock.lock(self.io);
defer coll.lock.unlock(self.io);
// Ids are duped rather than aliased: `remove` frees the docs-map key // Ids are duped rather than aliased: `remove` frees the docs-map key
// that `Entry.id` points at, which would leave the rest of the batch // that `Entry.id` points at, which would leave the rest of the batch
// pointing into freed memory. // pointing into freed memory.
@@ -497,57 +655,43 @@ pub const Engine = struct {
ids.deinit(self.gpa); ids.deinit(self.gpa);
} }
var db_it = self.dbs.iterator(); for (coll.indexes.items) |*ix| {
while (db_it.next()) |db_entry| { const ttl = ix.ttl orelse continue;
var coll_it = db_entry.value_ptr.collections.iterator(); const cutoff: i128 = @as(i128, now_ms) - @as(i128, ttl) * 1000;
while (coll_it.next()) |coll_entry| { // bson compare order ranks datetime above null, numbers
for (ids.items) |id| self.gpa.free(id); // and strings and below only timestamp and maxKey, so
ids.clearRetainingCapacity(); // datetimes form a contiguous band in the encoded key
// order: seek the minimum datetime and stop when the
for (coll_entry.value_ptr.indexes.items) |*ix| { // leading type changes or the cutoff is passed.
const ttl = ix.ttl orelse continue; const min_dt = [_]u8{ bson.encoded_datetime_tag, 0, 0, 0, 0, 0, 0, 0, 0 };
const cutoff: i128 = @as(i128, now_ms) - @as(i128, ttl) * 1000; var it = ix.seek(&min_dt);
// bson compare order ranks datetime above null, numbers while (it.next()) |e| {
// and strings and below only timestamp and maxKey, so const ms = bson.encoded_leading_datetime(e.key) orelse break;
// datetimes form a contiguous band in the encoded key if (@as(i128, ms) > cutoff) break;
// order: seek the minimum datetime and stop when the try ids.append(self.gpa, try self.gpa.dupe(u8, e.id));
// leading type changes or the cutoff is passed.
const min_dt = [_]u8{ bson.encoded_datetime_tag, 0, 0, 0, 0, 0, 0, 0, 0 };
var it = ix.seek(&min_dt);
while (it.next()) |e| {
const ms = bson.encoded_leading_datetime(e.key) orelse break;
if (@as(i128, ms) > cutoff) break;
try ids.append(self.gpa, try self.gpa.dupe(u8, e.id));
}
}
if (ids.items.len == 0) continue;
// One document can be expired by several entries (an array
// of dates) or by several TTL indexes.
std.mem.sort([]u8, ids.items, {}, less_id_bytes);
var w: usize = 1;
for (ids.items[1..]) |id| {
if (std.mem.eql(u8, id, ids.items[w - 1])) {
self.gpa.free(id);
} else {
ids.items[w] = id;
w += 1;
}
}
ids.items.len = w;
// `remove` only mutates the collection's docs map and index
// entries, never the dbs/collections maps, so both iterators
// above stay valid.
for (ids.items) |id| {
if (try self.remove(db_entry.key_ptr.*, coll_entry.key_ptr.*, id)) deleted += 1;
}
} }
} }
// A TTL-only workload never reaches the threshold check in `upsert`, if (ids.items.len == 0) return 0;
// so the log would otherwise grow without bound.
if (deleted > 0) try self.maybe_compact(); // One document can be expired by several entries (an array
return deleted; // of dates) or by several TTL indexes.
std.mem.sort([]u8, ids.items, {}, less_id_bytes);
var w: usize = 1;
for (ids.items[1..]) |id| {
if (std.mem.eql(u8, id, ids.items[w - 1])) {
self.gpa.free(id);
} else {
ids.items[w] = id;
w += 1;
}
}
ids.items.len = w;
var removed: usize = 0;
for (ids.items) |id| {
if (try self.remove(db_name, coll_name, id)) removed += 1;
}
return removed;
} }
pub fn database_names(self: *Engine, out: *std.ArrayListUnmanaged([]const u8)) !void { pub fn database_names(self: *Engine, out: *std.ArrayListUnmanaged([]const u8)) !void {
@@ -570,13 +714,15 @@ pub const Engine = struct {
try self.dbs.put(self.gpa, db_key, .{ .collections = .empty }); try self.dbs.put(self.gpa, db_key, .{ .collections = .empty });
return self.get_or_create_collection(db_name, coll_name); return self.get_or_create_collection(db_name, coll_name);
}; };
if (db.collections.getPtr(coll_name)) |coll| return coll; if (db.collections.get(coll_name)) |coll| return coll;
const coll_key = try self.gpa.dupe(u8, coll_name); const coll_key = try self.gpa.dupe(u8, coll_name);
errdefer self.gpa.free(coll_key); errdefer self.gpa.free(coll_key);
var new_coll = try Collection.init(self.gpa); const new_coll = try self.gpa.create(Collection);
errdefer self.gpa.destroy(new_coll);
new_coll.* = try Collection.init(self.gpa);
errdefer new_coll.id_index.deinit(self.gpa); errdefer new_coll.id_index.deinit(self.gpa);
try db.collections.put(self.gpa, coll_key, new_coll); try db.collections.put(self.gpa, coll_key, new_coll);
return db.collections.getPtr(coll_name) orelse unreachable; return new_coll;
} }
/// Deep-copy a document into engine-owned storage, prepending a /// Deep-copy a document into engine-owned storage, prepending a
@@ -612,7 +758,12 @@ pub const Engine = struct {
/// perfectly compact file for nothing. `compact` reports how much it /// perfectly compact file for nothing. `compact` reports how much it
/// reclaimed; when that is little, we back the baseline off /// reclaimed; when that is little, we back the baseline off
/// multiplicatively so a garbage-free log is left alone. /// multiplicatively so a garbage-free log is left alone.
fn maybe_compact(self: *Engine) !void { /// Called at the end of a write command's in-lock section: when the
/// garbage share crosses the threshold, record that a compaction is
/// wanted. It runs in the command epilogue, after the collection lock is
/// released — never inline, since compact takes the collection locks
/// itself and would deadlock against the caller's.
fn note_compact(self: *Engine) void {
// The threshold counts data volume (uncompressed record bytes), not // The threshold counts data volume (uncompressed record bytes), not
// the on-disk size: a compressed log would otherwise stay under any // the on-disk size: a compressed log would otherwise stay under any
// byte threshold and never compact its garbage. // byte threshold and never compact its garbage.
@@ -628,69 +779,111 @@ pub const Engine = struct {
// file stays near 1.25x the live data and each compaction is paid // file stays near 1.25x the live data and each compaction is paid
// for by the space it reclaims. // for by the space it reclaims.
if (self.dead_docs * 4 < self.live_docs) return; if (self.dead_docs * 4 < self.live_docs) return;
try self.compact(); self.compact_pending = true;
} }
/// Rewrite the log with only live documents, atomically swapping the file. /// Whether a compaction is wanted; clears the flag. Racy by design (two
/// Callers must hold the write lock. /// writers may both see it) — a redundant compact only rewrites an
/// already-compact log.
pub fn take_compact(self: *Engine) bool {
const p = self.compact_pending;
self.compact_pending = false;
return p;
}
/// Rewrite the log with only live documents, atomically swapping the
/// file. Runs outside every collection lock (called from a write
/// command's epilogue). The snapshot takes the collection locks one at
/// a time without the log lock — so a concurrent writer can always
/// finish its append — then takes the log lock and retries until no
/// writer appended during the snapshot (detected via the record seq),
/// which makes the snapshot consistent with what the log contains.
pub fn compact(self: *Engine) !void { pub fn compact(self: *Engine) !void {
const tmp_path = try std.fmt.allocPrint(self.gpa, "{s}.tmp", .{self.log.path}); const tmp_path = try std.fmt.allocPrint(self.gpa, "{s}.tmp", .{self.log.path});
defer self.gpa.free(tmp_path); defer self.gpa.free(tmp_path);
std.Io.Dir.cwd().deleteFile(self.io, tmp_path) catch {};
var new_log = try storage.Log.open(self.gpa, self.io, tmp_path);
defer new_log.close();
// One fsync for the whole rewrite, not one per document. The
// rewrite's durability comes from the rename below, which is only
// safe to publish after a single sync of the finished file.
new_log.defer_sync = true;
var db_it = self.dbs.iterator(); while (true) {
while (db_it.next()) |db_entry| { std.Io.Dir.cwd().deleteFile(self.io, tmp_path) catch {};
var coll_it = db_entry.value_ptr.collections.iterator(); var new_log = try storage.Log.open(self.gpa, self.io, tmp_path);
while (coll_it.next()) |coll_entry| { defer new_log.close();
// Re-emit the index definitions first: a compacted log that // One fsync for the whole rewrite, not one per document. The
// dropped them would resurrect the collections without // rewrite's durability comes from the rename below, which is
// indexes on replay. // only safe to publish after a single sync of the finished file.
for (coll_entry.value_ptr.indexes.items) |*ix| { new_log.defer_sync = true;
var spec_bytes: std.ArrayListUnmanaged(u8) = .empty;
defer spec_bytes.deinit(self.gpa); const snapshot_seq = self.seq;
try ix.write_spec(self.gpa, &spec_bytes); try self.catalog_lock.lockShared(self.io);
try new_log.append_index_create(db_entry.key_ptr.*, coll_entry.key_ptr.*, spec_bytes.items, self.seq); var catalog_err: ?anyerror = null;
} var db_it = self.dbs.iterator();
var doc_it = coll_entry.value_ptr.docs.iterator(); while (db_it.next()) |db_entry| {
while (doc_it.next()) |doc_entry| { var coll_it = db_entry.value_ptr.collections.iterator();
// The slab bytes are the canonical serialization. while (coll_it.next()) |coll_entry| {
const doc_bytes = coll_entry.value_ptr.doc_bytes(doc_entry.value_ptr.*); const coll = coll_entry.value_ptr.*;
try new_log.append_upsert(db_entry.key_ptr.*, coll_entry.key_ptr.*, doc_bytes, self.seq); 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;
} }
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);
const new_end_pos = new_log.end_pos;
// Durable before the rename makes it the database.
try new_log.sync();
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 starts at end_pos 0 and does not replay; continue
// appending where the compacted file actually ends.
self.log.end_pos = new_end_pos;
self.committed_end = new_end_pos;
// The rewritten log holds only live documents.
self.dead_docs = 0;
self.gpa.free(old_path);
self.log_lock.unlock(self.io);
return;
} }
const new_end_pos = new_log.end_pos; }
// Durable before the rename makes it the database.
try new_log.sync();
try std.Io.Dir.renameAbsolute(tmp_path, self.log.path, self.io); /// Re-emit one collection's index specs and documents into the compacted
// Persist the rename: fsync the parent directory so the new /// log, under the collection's write lock (released on every return
// directory entry survives a power loss right after compaction. /// path, including errors).
const parent = parent_dir(self.log.path); fn compact_snapshot_coll(self: *Engine, coll: *Collection, new_log: *storage.Log, db_name: []const u8, coll_name: []const u8) !void {
var dir_file = try std.Io.Dir.cwd().openFile(self.io, parent, .{ .mode = .read_only, .allow_directory = true }); try coll.lock.lock(self.io);
defer dir_file.close(self.io); defer coll.lock.unlock(self.io);
try dir_file.sync(self.io); // Re-emit the index definitions first: a compacted log that dropped
// them would resurrect the collections without indexes on replay.
const old_path = try self.gpa.dupe(u8, self.log.path); for (coll.indexes.items) |*ix| {
// A batch may span a compaction (a large insert crossing the var spec_bytes: std.ArrayListUnmanaged(u8) = .empty;
// threshold): keep the deferred-sync mode on the swapped-in log so defer spec_bytes.deinit(self.gpa);
// the remaining batch records stay grouped with the same command. try ix.write_spec(self.gpa, &spec_bytes);
const deferred = self.log.defer_sync; try new_log.append_index_create(db_name, coll_name, spec_bytes.items, self.seq);
self.log.close(); }
self.log = try storage.Log.open(self.gpa, self.io, old_path); var doc_it = coll.docs.iterator();
self.log.defer_sync = deferred; while (doc_it.next()) |doc_entry| {
// Log.open starts at end_pos 0 and does not replay; continue appending // The slab bytes are the canonical serialization.
// where the compacted file actually ends. const doc_bytes = coll.doc_bytes(doc_entry.value_ptr.*);
self.log.end_pos = new_end_pos; try new_log.append_upsert(db_name, coll_name, doc_bytes, self.seq);
// The rewritten log holds only live documents. }
self.dead_docs = 0;
self.gpa.free(old_path);
} }
/// Rebuild every empty index from the live documents. Runs after replay /// Rebuild every empty index from the live documents. Runs after replay
@@ -705,10 +898,10 @@ pub const Engine = struct {
while (db_it.next()) |db_entry| { while (db_it.next()) |db_entry| {
var coll_it = db_entry.value_ptr.collections.iterator(); var coll_it = db_entry.value_ptr.collections.iterator();
while (coll_it.next()) |coll_entry| { while (coll_it.next()) |coll_entry| {
for (coll_entry.value_ptr.indexes.items) |*ix| { for (coll_entry.value_ptr.*.indexes.items) |*ix| {
try self.rebuild_index(coll_entry.value_ptr, ix); try self.rebuild_index(coll_entry.value_ptr.*, ix);
} }
try self.rebuild_index(coll_entry.value_ptr, &coll_entry.value_ptr.id_index); try self.rebuild_index(coll_entry.value_ptr.*, &coll_entry.value_ptr.*.id_index);
} }
} }
} }
@@ -961,8 +1154,11 @@ test "compaction reclaims garbage but leaves a garbage-free log alone" {
defer d.deinit(); defer d.deinit();
try engine.insert("app", "c", &d, &env.gen); try engine.insert("app", "c", &d, &env.gen);
} }
try engine.commit();
try testing.expectEqual(@as(u64, 0), engine.dead_docs); try testing.expectEqual(@as(u64, 0), engine.dead_docs);
const after_insert = engine.log.end_pos; // The compaction threshold counts data volume (uncompressed bytes), not
// the compressed on-disk size.
const after_insert = engine.log.data_bytes;
try testing.expect(after_insert > engine.compact_threshold); try testing.expect(after_insert > engine.compact_threshold);
// Rewriting every document makes the log mostly garbage; compaction // Rewriting every document makes the log mostly garbage; compaction
@@ -973,11 +1169,16 @@ test "compaction reclaims garbage but leaves a garbage-free log alone" {
defer d.deinit(); defer d.deinit();
try engine.replace("app", "c", &d, &env.gen); try engine.replace("app", "c", &d, &env.gen);
} }
if (engine.log.end_pos < after_insert * 2) break; try engine.commit();
if (engine.log.data_bytes >= after_insert * 2) break;
} }
// The server runs compaction in a write command's epilogue; the tests
// drive it directly.
if (engine.take_compact()) try engine.compact();
try engine.commit();
try testing.expectEqual(@as(u64, 200), engine.live_docs); try testing.expectEqual(@as(u64, 200), engine.live_docs);
// Bounded well below the ~40x of record bytes those rewrites wrote. // Bounded well below the ~40x of record bytes those rewrites wrote.
try testing.expect(engine.log.end_pos < after_insert * 2); try testing.expect(engine.log.data_bytes < after_insert * 2);
} }
test "reopen replays log" { test "reopen replays log" {
@@ -1077,6 +1278,9 @@ test "compaction rewrites log and keeps data" {
docs[i] = try make_doc(gpa, @intCast(i + 1), "user-{d}"); docs[i] = try make_doc(gpa, @intCast(i + 1), "user-{d}");
try engine.insert("app", "users", &docs[i], &env.gen); try engine.insert("app", "users", &docs[i], &env.gen);
} }
try engine.commit();
// threshold 1 makes every write want a compaction; run it.
if (engine.take_compact()) try engine.compact();
engine.unlock(); engine.unlock();
} }
@@ -1088,6 +1292,7 @@ test "compaction rewrites log and keeps data" {
var extra = try make_doc(gpa, 5, "eve"); var extra = try make_doc(gpa, 5, "eve");
defer extra.deinit(); defer extra.deinit();
try engine2.insert("app", "users", &extra, &env.gen); try engine2.insert("app", "users", &extra, &env.gen);
try engine2.commit();
engine2.unlock(); engine2.unlock();
var engine3 = try Engine.open(gpa, io, tmp.path); var engine3 = try Engine.open(gpa, io, tmp.path);
@@ -1237,7 +1442,7 @@ test "unique index enforced on insert, replace, and upsert-conflict" {
var d2 = try make_user(gpa, 2, "a@x.io"); var d2 = try make_user(gpa, 2, "a@x.io");
defer d2.deinit(); defer d2.deinit();
try testing.expectError(error.DuplicateKeyIndex, engine.insert("app", "users", &d2, &env.gen)); try testing.expectError(error.DuplicateKeyIndex, engine.insert("app", "users", &d2, &env.gen));
try testing.expectEqualStrings("email_1", engine.dup_index.?); try testing.expectEqualStrings("email_1", engine.get_collection("app", "users").?.dup_index.?);
// A replace that keeps its own email is fine (own entries excluded). // A replace that keeps its own email is fine (own entries excluded).
var d1b = try make_user(gpa, 1, "a@x.io"); var d1b = try make_user(gpa, 1, "a@x.io");

129
src/fuzz_split.zig Normal file
View File

@@ -0,0 +1,129 @@
//! Randomised differential over Index with wildly varying key sizes.
//!
//! The tree's split logic is where record sizes and slot counts interact:
//! a page can be full of bytes or full of slots, and the record that caused
//! the split has to fit the half it lands in. Fixed-shape tests never mix
//! those, so this hammers the same index with keys from 4 bytes to past the
//! inline limit, interleaves removals (which leave dead bytes behind), and
//! checks the whole tree against a model.
//!
//! Run: zig test src/fuzz_split.zig
const std = @import("std");
const bson = @import("bson.zig");
const index = @import("index.zig");
const testing = std.testing;
fn make_doc(gpa: std.mem.Allocator, i: usize, s: []const u8) ![]u8 {
var out: std.ArrayListUnmanaged(u8) = .empty;
defer out.deinit(gpa);
try bson.write_doc(&.{
.{ .key = "_id", .value = .{ .int32 = @intCast(i) } },
.{ .key = "s", .value = .{ .string = s } },
}, gpa, &out);
return out.toOwnedSlice(gpa);
}
const Doc = struct {
id: []u8,
s: []u8,
bytes: []u8,
live: bool,
};
fn run(seed: u64, ops: usize, max_len: usize) !void {
const gpa = testing.allocator;
var prng = std.Random.DefaultPrng.init(seed);
const rand = prng.random();
var ix = try index.Index.init(gpa, "s_1", &.{.{ .path = "s", .descending = false }}, false, false, null);
defer ix.deinit(gpa);
var docs: std.ArrayListUnmanaged(Doc) = .empty;
defer {
for (docs.items) |d| {
gpa.free(d.id);
gpa.free(d.s);
gpa.free(d.bytes);
}
docs.deinit(gpa);
}
var live: usize = 0;
for (0..ops) |op| {
if (live > 0 and rand.uintLessThan(u32, 100) < 35) {
// Remove a random live document.
var pick = rand.uintLessThan(usize, live);
for (docs.items) |*d| {
if (!d.live) continue;
if (pick == 0) {
ix.remove_doc(gpa, d.bytes, d.id);
d.live = false;
live -= 1;
break;
}
pick -= 1;
}
} else {
// Insert a document whose key length is drawn from a mix of
// tiny, around the inline limit, and past it (spilled).
const len = switch (rand.uintLessThan(u32, 10)) {
0...4 => rand.intRangeAtMost(usize, 1, 16),
5...7 => rand.intRangeAtMost(usize, 900, 1100),
else => rand.intRangeAtMost(usize, 1100, max_len),
};
const s = try gpa.alloc(u8, len);
errdefer gpa.free(s);
// A small alphabet so keys collide and share prefixes.
for (s) |*c| c.* = 'a' + rand.uintLessThan(u8, 4);
const id = try std.fmt.allocPrint(gpa, "id{d}", .{op});
errdefer gpa.free(id);
const bytes = try make_doc(gpa, op, s);
errdefer gpa.free(bytes);
_ = try ix.add_doc(gpa, bytes, id, false);
try docs.append(gpa, .{ .id = id, .s = s, .bytes = bytes, .live = true });
live += 1;
}
if (op % 25 != 0 and op != ops - 1) continue;
// The tree holds exactly the live entries, in key order.
try testing.expectEqual(live, ix.count());
var seen: usize = 0;
var prev: []const u8 = "";
var it = ix.iter();
while (it.next()) |e| : (seen += 1) {
try testing.expect(std.mem.order(u8, prev, e.key) != .gt);
prev = e.key;
}
try testing.expectEqual(live, seen);
// Every live document is reachable by a descent, not just by
// walking the leaf chain: a bad separator breaks only the descent.
var found: std.ArrayListUnmanaged([]const u8) = .empty;
defer found.deinit(gpa);
for (docs.items) |d| {
if (!d.live) continue;
if (rand.uintLessThan(u32, 100) >= 10) continue; // sample
found.clearRetainingCapacity();
try ix.lookup_eq(gpa, &.{.{ .string = d.s }}, &found);
var hit = false;
for (found.items) |got| {
if (std.mem.eql(u8, got, d.id)) hit = true;
}
if (!hit) {
std.debug.print("seed {d} op {d}: id {s} (key len {d}) not found by descent\n", .{ seed, op, d.id, d.s.len });
return error.EntryUnreachable;
}
}
}
}
test "mixed key sizes with removals" {
for (0..6) |k| try run(@intCast(k + 1), 1500, 8000);
}
test "keys clustered around the inline limit" {
for (0..4) |k| try run(@intCast(k + 100), 1200, 1200);
}

418
src/index.zig
View File

@@ -83,6 +83,9 @@ const page_size = 4096;
const page_data = page_size - 32; const page_data = page_size - 32;
/// Records longer than a quarter of a node spill to the overflow slab. /// Records longer than a quarter of a node spill to the overflow slab.
const inline_limit = page_size / 4; const inline_limit = page_size / 4;
/// Upper bound on the slots one node can hold, since every slot costs at
/// least its own size. Bounds the split scratch.
const max_slots = page_data / slot_size;
/// One slotted-page entry: a key plus either an id length (leaf records) or /// One slotted-page entry: a key plus either an id length (leaf records) or
/// a child node id (internal separators). The key bytes live either inline /// a child node id (internal separators). The key bytes live either inline
@@ -300,14 +303,27 @@ pub const Index = struct {
/// and the exact overflow bytes (spilled records are copied once). /// and the exact overflow bytes (spilled records are copied once).
pub fn reserve_for(self: *Index, gpa: std.mem.Allocator, entries: []const Entry) !void { pub fn reserve_for(self: *Index, gpa: std.mem.Allocator, entries: []const Entry) !void {
const n: u64 = entries.len; const n: u64 = entries.len;
// One entry splits at most one node per level plus a new root. A
// batch can also deepen the tree as it goes, and every level it
// adds costs one more node per remaining entry; a level needs a
// full root and the smallest root holds three ~1 KiB separators, so
// n/8 levels per batch is well clear of the worst case.
const extra_nodes: u64 = n * (self.depth + 2 + n / 8) + 4;
try self.nodes.ensureUnusedCapacity(gpa, @intCast(extra_nodes));
try self.reserve_overflow(gpa, entries);
}
/// Reserve the slab bytes `entries` will spill, so that store_record's
/// append is infallible. A spilled record is copied into the slab once,
/// when it first enters the tree; moving it between nodes re-uses its
/// offset.
fn reserve_overflow(self: *Index, gpa: std.mem.Allocator, entries: []const Entry) !void {
var overflow_bytes: u64 = 0; var overflow_bytes: u64 = 0;
for (entries) |e| { for (entries) |e| {
const rec_len: u64 = e.key.len + e.id.len; const rec_len: u64 = e.key.len + e.id.len;
if (rec_len > inline_limit) overflow_bytes += rec_len; if (rec_len > inline_limit) overflow_bytes += rec_len;
} }
const extra_nodes: u64 = n + n / 100 + self.depth + 2; try self.overflow.ensureUnusedCapacity(gpa, @intCast(overflow_bytes));
try self.nodes.ensureUnusedCapacity(gpa, self.nodes.items.len + @as(usize, @intCast(extra_nodes)));
try self.overflow.ensureUnusedCapacity(gpa, self.overflow.items.len + @as(usize, @intCast(overflow_bytes)));
} }
/// Insert pre-built entries (maintaining order) and drain the batch. /// Insert pre-built entries (maintaining order) and drain the batch.
@@ -387,7 +403,10 @@ pub const Index = struct {
duplicate = true; duplicate = true;
} }
} }
try self.reserve_for(gpa, self.staging.items); // Only the slab needs reserving up front: pack_tree grows the node
// array itself, so a bulk build no longer reserves a node per entry
// when it needs one per leaf (65k entries pack into ~1k leaves).
try self.reserve_overflow(gpa, self.staging.items);
try self.pack_tree(gpa); try self.pack_tree(gpa);
return duplicate; return duplicate;
} }
@@ -604,22 +623,25 @@ pub const Index = struct {
key: []const u8, key: []const u8,
id: []const u8 = "", id: []const u8 = "",
child: u32 = 0, child: u32 = 0,
/// When set, key+id already live in the overflow slab and are /// When set, key+id already live in the overflow slab at this
/// referenced rather than copied (moved leaf records, promoted /// offset and are referenced rather than copied (moved leaf
/// separators). /// records, promoted separators). Optional rather than a 0
spill_off: u64 = 0, /// sentinel: offset 0 is a real slab position, held by the first
/// record that ever spilled -- which would otherwise be copied into
/// the slab again on every move, past the reserved capacity.
spill_off: ?u64 = null,
}; };
const StableKey = struct { const StableKey = struct {
key: []const u8, key: []const u8,
spill_off: u64, spill_off: ?u64,
}; };
const Split = struct { const Split = struct {
/// The separator to promote, stable across node-array growth (a /// The separator to promote, stable across node-array growth (a
/// slice into the overflow slab, or a copy in the promo buffer). /// slice into the overflow slab, or a copy in the promo buffer).
key: []const u8, key: []const u8,
spill_off: u64, spill_off: ?u64,
right: u32, right: u32,
}; };
@@ -644,6 +666,14 @@ pub const Index = struct {
return @intCast(self.nodes.items.len - 1); return @intCast(self.nodes.items.len - 1);
} }
/// Allocate a node id, growing the array. For the pack path, which is
/// fallible anyway and would otherwise have to reserve one node per
/// entry when it needs one per leaf.
fn alloc_node_grow(self: *Index, gpa: std.mem.Allocator) !u32 {
try self.nodes.append(gpa, empty_node(0));
return @intCast(self.nodes.items.len - 1);
}
fn get_slot(node: *const Node, i: u32) Slot { fn get_slot(node: *const Node, i: u32) Slot {
return std.mem.bytesToValue(Slot, node.buf[i * slot_size ..][0..slot_size]); return std.mem.bytesToValue(Slot, node.buf[i * slot_size ..][0..slot_size]);
} }
@@ -683,13 +713,14 @@ pub const Index = struct {
const node = &self.nodes.items[node_id]; const node = &self.nodes.items[node_id];
const rec_len: u64 = rec.key.len + rec.id.len; const rec_len: u64 = rec.key.len + rec.id.len;
var s: Slot = .{ var s: Slot = .{
.off = rec.spill_off, .off = 0,
.key_len = @intCast(rec.key.len), .key_len = @intCast(rec.key.len),
.extra = if (rec.id.len > 0) @intCast(rec.id.len) else rec.child, .extra = if (rec.id.len > 0) @intCast(rec.id.len) else rec.child,
.spill = false, .spill = false,
._pad = 0, ._pad = 0,
}; };
if (rec.spill_off != 0) { if (rec.spill_off) |off| {
s.off = off;
s.spill = true; s.spill = true;
} else if (rec_len > inline_limit) { } else if (rec_len > inline_limit) {
s.off = self.overflow.items.len; s.off = self.overflow.items.len;
@@ -738,6 +769,7 @@ pub const Index = struct {
const dst_slots = node.buf[i * slot_size .. (node.count - 1) * slot_size]; const dst_slots = node.buf[i * slot_size .. (node.count - 1) * slot_size];
std.mem.copyForwards(u8, dst_slots, src_slots); std.mem.copyForwards(u8, dst_slots, src_slots);
node.count -= 1; node.count -= 1;
if (node.count == 0) node.data_start = page_data;
} }
/// Repack `node` keeping only slots [0, k). Record bytes are copied to /// Repack `node` keeping only slots [0, k). Record bytes are copied to
@@ -783,7 +815,39 @@ pub const Index = struct {
if (s.spill) return .{ .key = self.overflow.items[@intCast(s.off) .. @intCast(s.off + s.key_len)], .spill_off = s.off }; if (s.spill) return .{ .key = self.overflow.items[@intCast(s.off) .. @intCast(s.off + s.key_len)], .spill_off = s.off };
const k = self.key_of(node_id, i); const k = self.key_of(node_id, i);
@memcpy(self.promo[0..k.len], k); @memcpy(self.promo[0..k.len], k);
return .{ .key = self.promo[0..k.len], .spill_off = 0 }; return .{ .key = self.promo[0..k.len], .spill_off = null };
}
/// What one record costs a page: its slot plus, when it stays inline,
/// its bytes. The unit both `fits` and the split point are measured in.
fn record_cost(rec_len: u64) u32 {
return slot_size + @as(u32, if (rec_len > inline_limit) 0 else @intCast(rec_len));
}
/// `record_cost` of the record already in slot `i`.
fn slot_cost(self: *const Index, node_id: u32, i: u32) u32 {
const node = &self.nodes.items[node_id];
const s = get_slot(node, i);
if (s.spill) return slot_size;
return slot_size + s.key_len + (if (node.is_leaf == 1) s.extra else 0);
}
/// Where to cut a merged run of records so that neither half exceeds
/// half the total cost by more than one record: the first cut whose
/// left half would pass half the total. A page's live records plus one
/// new record cost at most page_data plus one record, so both halves
/// are then guaranteed to fit a page. Cutting by slot count is not:
/// every large record can sit on one side of the count midpoint.
fn balanced_cut(costs: []const u32) u32 {
var total: u64 = 0;
for (costs) |c| total += c;
var acc: u64 = 0;
var mid: u32 = 0;
while (mid < costs.len) : (mid += 1) {
acc += costs[mid];
if (acc * 2 > total) break;
}
return @min(mid, @as(u32, @intCast(costs.len - 1)));
} }
/// Entry position in a leaf, by (key, id). /// Entry position in a leaf, by (key, id).
@@ -892,6 +956,17 @@ pub const Index = struct {
fn insert_rec(self: *Index, node_id: u32, key: []const u8, id: []const u8) ?Split { fn insert_rec(self: *Index, node_id: u32, key: []const u8, id: []const u8) ?Split {
const node = &self.nodes.items[node_id]; const node = &self.nodes.items[node_id];
if (node.is_leaf == 1) { if (node.is_leaf == 1) {
if (self.fits(node_id, key.len + id.len)) {
self.store_record(node_id, self.leaf_pos(node_id, key, id), .{ .key = key, .id = id });
self.entry_count += 1;
return null;
}
// remove_record leaves freed bytes dead at the top of the page,
// so a churned leaf can run out of room with only a few live
// records. Reclaim the dead bytes first; a leaf with a handful
// of <= 1 KiB records always fits after that, so the split below
// only ever sees a genuinely full leaf with count >= 2.
self.repack_keep_prefix(node_id, node.count);
if (self.fits(node_id, key.len + id.len)) { if (self.fits(node_id, key.len + id.len)) {
self.store_record(node_id, self.leaf_pos(node_id, key, id), .{ .key = key, .id = id }); self.store_record(node_id, self.leaf_pos(node_id, key, id), .{ .key = key, .id = id });
self.entry_count += 1; self.entry_count += 1;
@@ -904,48 +979,72 @@ pub const Index = struct {
return self.insert_separator(node_id, res); return self.insert_separator(node_id, res);
} }
/// Split a full leaf: the right half (including any records equal to /// Split a full leaf around the record being inserted. The new record
/// the boundary key — lookups scan whole key bands, so equal keys may /// takes its place in the leaf's key order and the merged run is cut
/// live on both sides) moves to a new leaf, the boundary key is /// where the two halves come closest to equal cost, so the half the new
/// promoted, and the new record is stored in whichever half holds its /// record lands in is guaranteed to have room for it. Records equal to
/// position. /// the boundary key may end up on both sides; lookups scan whole key
fn split_leaf(self: *Index, leaf_id: u32, key: []const u8, id: []const u8) ?Split { /// bands, so that is fine. The right half moves to a new leaf and its
const node = &self.nodes.items[leaf_id]; /// first key is promoted.
const mid = @max(1, node.count / 2); fn split_leaf(self: *Index, leaf_id: u32, key: []const u8, id: []const u8) Split {
// insert_rec reclaims dead bytes before giving up on a page, so the
// page is genuinely full here and holds at least two records --
// which is what makes both halves below non-empty.
const old_count = self.nodes.items[leaf_id].count;
std.debug.assert(old_count >= 2);
const pos = self.leaf_pos(leaf_id, key, id);
const n = old_count + 1;
var costs: [max_slots + 1]u32 = undefined;
for (0..n) |m| {
costs[m] = if (m == pos)
record_cost(key.len + id.len)
else
self.slot_cost(leaf_id, @intCast(if (m < pos) m else m - 1));
}
const mid = @max(1, @min(balanced_cut(costs[0..n]), n - 1));
const right_id = self.alloc_node(); const right_id = self.alloc_node();
{ {
const node = &self.nodes.items[leaf_id];
const right = &self.nodes.items[right_id]; const right = &self.nodes.items[right_id];
right.is_leaf = 1; right.is_leaf = 1;
right.next = node.next; right.next = node.next;
right.prev = leaf_id; right.prev = leaf_id;
right.parent = node.parent; right.parent = node.parent;
} }
// Move the right half; spilled records keep their slab reference. // Move the merged tail; spilled records keep their slab reference.
var i: u32 = mid; var m: u32 = mid;
while (i < node.count) : (i += 1) { while (m < n) : (m += 1) {
const s = get_slot(&self.nodes.items[leaf_id], i); const at: u32 = self.nodes.items[right_id].count;
self.store_record(right_id, @intCast(self.nodes.items[right_id].count), .{ if (m == pos) {
.key = self.key_of(leaf_id, i), self.store_record(right_id, at, .{ .key = key, .id = id });
.id = self.id_of(leaf_id, i), continue;
.spill_off = if (s.spill) s.off else 0, }
const src: u32 = if (m < pos) m else m - 1;
const slot = get_slot(&self.nodes.items[leaf_id], src);
self.store_record(right_id, at, .{
.key = self.key_of(leaf_id, src),
.id = self.id_of(leaf_id, src),
.spill_off = if (slot.spill) slot.off else null,
}); });
} }
// Repack the left half in place. // Keep the merged head in the left page: its own records, plus the
self.repack_keep_prefix(leaf_id, mid); // new one when that is the half it belongs to.
// Link the chain. if (pos < mid) {
const node2 = &self.nodes.items[leaf_id]; self.repack_keep_prefix(leaf_id, mid - 1);
if (node2.next != 0) self.nodes.items[node2.next].prev = right_id; self.store_record(leaf_id, pos, .{ .key = key, .id = id });
node2.next = right_id;
self.leaf_count += 1;
// Promote the right leaf's first key.
const sep = self.stable_key(right_id, 0);
// Store the new record in the correct half.
if (std.mem.order(u8, key, sep.key) == .lt) {
self.store_record(leaf_id, self.leaf_pos(leaf_id, key, id), .{ .key = key, .id = id });
} else { } else {
self.store_record(right_id, self.leaf_pos(right_id, key, id), .{ .key = key, .id = id }); self.repack_keep_prefix(leaf_id, mid);
} }
// Link the chain.
const left = &self.nodes.items[leaf_id];
if (left.next != 0) self.nodes.items[left.next].prev = right_id;
left.next = right_id;
self.leaf_count += 1;
self.entry_count += 1; self.entry_count += 1;
const sep = self.stable_key(right_id, 0);
return .{ .key = sep.key, .spill_off = sep.spill_off, .right = right_id }; return .{ .key = sep.key, .spill_off = sep.spill_off, .right = right_id };
} }
@@ -957,10 +1056,16 @@ pub const Index = struct {
// spilled keys already live in the immutable slab. Copy inline keys // spilled keys already live in the immutable slab. Copy inline keys
// to a stack buffer so they survive. // to a stack buffer so they survive.
var local: [inline_limit]u8 = undefined; var local: [inline_limit]u8 = undefined;
const key: []const u8 = if (split.spill_off != 0) split.key else blk: { const key: []const u8 = if (split.spill_off != null) split.key else blk: {
@memcpy(local[0..split.key.len], split.key); @memcpy(local[0..split.key.len], split.key);
break :blk local[0..split.key.len]; break :blk local[0..split.key.len];
}; };
// As in a leaf: a dropped child leaves its separator's bytes dead
// in the page, so reclaim before believing the page is full. This
// is also what guarantees at least two separators at the split.
if (!self.fits(node_id, key.len)) {
self.repack_keep_prefix(node_id, self.nodes.items[node_id].count);
}
if (self.fits(node_id, key.len)) { if (self.fits(node_id, key.len)) {
self.store_record(node_id, self.separator_pos(node_id, key), .{ self.store_record(node_id, self.separator_pos(node_id, key), .{
.key = key, .key = key,
@@ -970,55 +1075,79 @@ pub const Index = struct {
self.nodes.items[split.right].parent = node_id; self.nodes.items[split.right].parent = node_id;
return null; return null;
} }
const up = self.split_internal(node_id); return self.split_internal(node_id, key, split.spill_off, split.right);
// The node split; the separator goes into whichever half holds its
// position. Both halves have room (they are half-full).
if (std.mem.order(u8, key, up.key) == .lt) {
self.store_record(node_id, self.separator_pos(node_id, key), .{
.key = key,
.child = split.right,
.spill_off = split.spill_off,
});
self.nodes.items[split.right].parent = node_id;
} else {
self.store_record(up.right, self.separator_pos(up.right, key), .{
.key = key,
.child = split.right,
.spill_off = split.spill_off,
});
self.nodes.items[split.right].parent = up.right;
}
return up;
} }
/// Split a full internal node: the middle separator is promoted, the /// Split a full internal node around the separator being inserted: the
/// first half stays, the rest moves to a new right node. /// new separator joins the node's order, the merged run is cut where
fn split_internal(self: *Index, node_id: u32) Split { /// the halves come closest to equal cost (see split_leaf), and the
const node = &self.nodes.items[node_id]; /// separator at the cut is promoted -- its child becoming the right
const s = node.count / 2; /// node's first child.
// Copy the promoted key before the repack rewrites the page. fn split_internal(self: *Index, node_id: u32, key: []const u8, spill_off: ?u64, child: u32) Split {
const mid_key = self.stable_key(node_id, s); const old_count = self.nodes.items[node_id].count;
std.debug.assert(old_count >= 2);
const pos = self.separator_pos(node_id, key);
const n = old_count + 1;
var costs: [max_slots + 1]u32 = undefined;
for (0..n) |m| {
costs[m] = if (m == pos)
record_cost(key.len)
else
self.slot_cost(node_id, @intCast(if (m < pos) m else m - 1));
}
// The cut is promoted rather than kept, so only the right half
// needs a separator left over.
const mid = @min(balanced_cut(costs[0..n]), n - 1);
// The promoted key has to survive the repack below. A spilled key
// already lives in the immutable slab (and can be far larger than
// the promo buffer); an inline one is copied into promo.
const promoted: StableKey = if (mid != pos)
self.stable_key(node_id, if (mid < pos) mid else mid - 1)
else if (spill_off != null)
.{ .key = key, .spill_off = spill_off }
else blk: {
@memcpy(self.promo[0..key.len], key);
break :blk .{ .key = self.promo[0..key.len], .spill_off = null };
};
const right_id = self.alloc_node(); const right_id = self.alloc_node();
{ self.nodes.items[right_id].is_leaf = 0;
const right = &self.nodes.items[right_id]; self.nodes.items[right_id].parent = self.nodes.items[node_id].parent;
right.is_leaf = 0; // The promoted separator's child heads the right node.
right.parent = node.parent; const mid_child: u32 = if (mid == pos)
right.first_child = get_slot(node, s).extra; child
// The moved subtrees now live under the right node. else
self.nodes.items[right.first_child].parent = right_id; get_slot(&self.nodes.items[node_id], if (mid < pos) mid else mid - 1).extra;
} self.nodes.items[right_id].first_child = mid_child;
var i: u32 = s + 1; self.nodes.items[mid_child].parent = right_id;
while (i < node.count) : (i += 1) {
const slot_i = get_slot(&self.nodes.items[node_id], i); var m: u32 = mid + 1;
self.nodes.items[slot_i.extra].parent = right_id; while (m < n) : (m += 1) {
self.store_record(right_id, @intCast(self.nodes.items[right_id].count), .{ const at: u32 = self.nodes.items[right_id].count;
.key = self.key_of(node_id, i), if (m == pos) {
.child = slot_i.extra, self.store_record(right_id, at, .{ .key = key, .child = child, .spill_off = spill_off });
.spill_off = if (slot_i.spill) slot_i.off else 0, self.nodes.items[child].parent = right_id;
continue;
}
const src: u32 = if (m < pos) m else m - 1;
const slot = get_slot(&self.nodes.items[node_id], src);
self.store_record(right_id, at, .{
.key = self.key_of(node_id, src),
.child = slot.extra,
.spill_off = if (slot.spill) slot.off else null,
}); });
self.nodes.items[slot.extra].parent = right_id;
} }
self.repack_keep_prefix(node_id, s); if (pos < mid) {
return .{ .key = mid_key.key, .spill_off = mid_key.spill_off, .right = right_id }; self.repack_keep_prefix(node_id, mid - 1);
self.store_record(node_id, pos, .{ .key = key, .child = child, .spill_off = spill_off });
self.nodes.items[child].parent = node_id;
} else {
self.repack_keep_prefix(node_id, mid);
}
return .{ .key = promoted.key, .spill_off = promoted.spill_off, .right = right_id };
} }
/// Remove the exact entry (key, id). Equal keys may span several /// Remove the exact entry (key, id). Equal keys may span several
@@ -1097,9 +1226,9 @@ pub const Index = struct {
/// The internal root emptied: swap in a fresh empty leaf as the root. /// The internal root emptied: swap in a fresh empty leaf as the root.
fn replace_root_with_leaf(self: *Index) void { fn replace_root_with_leaf(self: *Index) void {
self.root = self.alloc_node(); // Removal reserves no capacity, so this must not allocate a node:
self.nodes.items[self.root].is_leaf = 1; // re-use the emptied root page as the empty root leaf.
self.nodes.items[self.root].parent = 0; self.nodes.items[self.root] = empty_node(1);
self.first_leaf = self.root; self.first_leaf = self.root;
self.leaf_count = 1; self.leaf_count = 1;
self.depth = 0; self.depth = 0;
@@ -1114,6 +1243,10 @@ pub const Index = struct {
self.staging.clearRetainingCapacity(); self.staging.clearRetainingCapacity();
} }
if (self.staging.items.len == 0) return; if (self.staging.items.len == 0) return;
// A fresh tree replaces whatever was here; the old pages are
// abandoned in place, like any other dropped node.
self.leaf_count = 0;
self.depth = 0;
const Level = struct { const Level = struct {
id: u32, id: u32,
@@ -1126,7 +1259,7 @@ pub const Index = struct {
var prev: u32 = 0; var prev: u32 = 0;
var lit = self.staging.items; var lit = self.staging.items;
while (lit.len > 0) { while (lit.len > 0) {
const leaf = self.alloc_node(); const leaf = try self.alloc_node_grow(gpa);
self.nodes.items[leaf].is_leaf = 1; self.nodes.items[leaf].is_leaf = 1;
const first_key = lit[0].key; const first_key = lit[0].key;
// Fill until the next record would not fit. // Fill until the next record would not fit.
@@ -1158,7 +1291,7 @@ pub const Index = struct {
defer next.deinit(gpa); defer next.deinit(gpa);
var i: usize = 0; var i: usize = 0;
while (i < level.items.len) { while (i < level.items.len) {
const node = self.alloc_node(); const node = try self.alloc_node_grow(gpa);
self.nodes.items[node].first_child = level.items[i].id; self.nodes.items[node].first_child = level.items[i].id;
self.nodes.items[level.items[i].id].parent = node; self.nodes.items[level.items[i].id].parent = node;
const first_key = level.items[i].first_key; const first_key = level.items[i].first_key;
@@ -1172,7 +1305,6 @@ pub const Index = struct {
self.store_record(node, @intCast(slots), .{ self.store_record(node, @intCast(slots), .{
.key = level.items[j].first_key, .key = level.items[j].first_key,
.child = level.items[j].id, .child = level.items[j].id,
.spill_off = 0,
}); });
self.nodes.items[level.items[j].id].parent = node; self.nodes.items[level.items[j].id].parent = node;
slots += 1; slots += 1;
@@ -2360,6 +2492,106 @@ test "TTL spec rejects compound keys and bad expireAfterSeconds" {
try testing.expectEqualStrings("expireAt_1", ix.name); try testing.expectEqualStrings("expireAt_1", ix.name);
} }
fn str_doc(gpa: std.mem.Allocator, i: usize, fill: u8, len: usize) ![]u8 {
const s = try gpa.alloc(u8, len);
defer gpa.free(s);
@memset(s, fill);
return bytes_of(gpa, &.{
.{ .key = "_id", .value = .{ .int32 = @intCast(i) } },
.{ .key = "s", .value = .{ .string = s } },
});
}
test "a churned leaf of large keys splits without promoting from an empty half" {
// remove_record leaves the freed bytes dead at the top of the page, so
// a leaf holding one or two ~1 KiB keys runs out of room after a few
// insert/remove cycles while still holding almost nothing. Splitting
// then would hand the right half zero records and promote whatever the
// uninitialised slot 0 happened to hold.
const gpa = testing.allocator;
var ix = try simple_index(gpa, &.{"s"}, false, false);
defer ix.deinit(gpa);
const n = 12;
var docs: [n][]u8 = undefined;
var ids: [n][]u8 = undefined;
var made: usize = 0;
defer for (0..made) |i| {
gpa.free(docs[i]);
gpa.free(ids[i]);
};
for (0..n) |i| {
docs[i] = try str_doc(gpa, i, @intCast('a' + i), 1000);
ids[i] = try std.fmt.allocPrint(gpa, "id{d}", .{i});
made += 1;
}
// Keep one entry live while the page fills with dead bytes, then leave
// two live: a bad split promotes an empty page's slot 0 as the
// separator, and every later descent then misses the left leaf. The
// leaf chain would still hold both, so this has to be checked through a
// lookup, which descends.
for (0..n - 1) |i| {
_ = try ix.add_doc(gpa, docs[i], ids[i], false);
if (i >= 1) ix.remove_doc(gpa, docs[i - 1], ids[i - 1]);
}
_ = try ix.add_doc(gpa, docs[n - 1], ids[n - 1], false);
try testing.expectEqual(@as(usize, 2), ix.count());
for ([_]usize{ n - 2, n - 1 }) |i| {
var s: [1000]u8 = undefined;
@memset(&s, @intCast('a' + i));
try expect_ids(gpa, &ix, &.{.{ .string = &s }}, &.{ids[i]});
}
}
test "a split with lopsided record sizes keeps the new record inside its page" {
// The halves are split by slot count, so all the large records can end
// up on one side; the record that caused the split is then stored into
// that half with no room left for it.
const gpa = testing.allocator;
var ix = try simple_index(gpa, &.{"s"}, false, false);
defer ix.deinit(gpa);
var docs: std.ArrayListUnmanaged([]u8) = .empty;
var ids: std.ArrayListUnmanaged([]u8) = .empty;
defer {
for (docs.items) |d| gpa.free(d);
for (ids.items) |x| gpa.free(x);
docs.deinit(gpa);
ids.deinit(gpa);
}
var i: usize = 0;
// Three ~1 KiB keys, all sorting before the short ones ('a' < 'z') ...
while (i < 3) : (i += 1) {
try docs.append(gpa, try str_doc(gpa, i, 'a', 1000 - i));
try ids.append(gpa, try std.fmt.allocPrint(gpa, "id{d}", .{i}));
}
// ... then short keys, filling the page by slot count ...
while (i < 28) : (i += 1) {
try docs.append(gpa, try str_doc(gpa, i, 'z', 4));
try ids.append(gpa, try std.fmt.allocPrint(gpa, "id{d}", .{i}));
}
// ... then one more large key, which sorts into the left half.
try docs.append(gpa, try str_doc(gpa, i, 'a', 996));
try ids.append(gpa, try std.fmt.allocPrint(gpa, "id{d}", .{i}));
for (docs.items, ids.items) |d, id| _ = try ix.add_doc(gpa, d, id, false);
try testing.expectEqual(docs.items.len, ix.count());
// Every key comes back, in order, exactly once.
var prev: []const u8 = "";
var seen: usize = 0;
var it = ix.iter();
while (it.next()) |e| : (seen += 1) {
try testing.expect(std.mem.order(u8, prev, e.key) != .gt);
prev = e.key;
}
try testing.expectEqual(docs.items.len, seen);
}
test "the _id index plan covers equality, ranges and _id sort order" { test "the _id index plan covers equality, ranges and _id sort order" {
// The implicit _id_ index is a normal Index (keys = [_id: 1]) passed to // The implicit _id_ index is a normal Index (keys = [_id: 1]) passed to
// plan separately from the secondaries. Its encoded keys are canonical, // plan separately from the secondaries. Its encoded keys are canonical,

15
src/server.zig
View File

@@ -58,23 +58,26 @@ pub const Server = struct {
}; };
/// Expire documents under TTL indexes every `ttl_sweep_secs` seconds, until /// Expire documents under TTL indexes every `ttl_sweep_secs` seconds, until
/// the group is canceled. Sweeping takes the engine's write lock, so it is /// the group is canceled. The sweep takes the catalog lock and one
/// serialized with commands exactly like any other write; a sweep failure is /// collection's write lock at a time, exactly like a write command; a sweep
/// logged rather than fatal, since the next one will retry. /// failure is logged rather than fatal, since the next one will retry.
fn ttl_monitor(io: std.Io, server: *Server) error{Canceled}!void { fn ttl_monitor(io: std.Io, server: *Server) error{Canceled}!void {
const interval: std.Io.Duration = .fromSeconds(server.ttl_sweep_secs); const interval: std.Io.Duration = .fromSeconds(server.ttl_sweep_secs);
while (true) { while (true) {
// Sleep first: at startup the engine has just replayed the log, and // Sleep first: at startup the engine has just replayed the log, and
// an immediate sweep would race the listener's first connections for // an immediate sweep would race the listener's first connections for
// the write lock. // the collection locks.
try std.Io.sleep(io, interval, .awake); try std.Io.sleep(io, interval, .awake);
try server.engine.lock();
defer server.engine.unlock();
const now_ms = std.Io.Timestamp.now(io, .real).toMilliseconds(); const now_ms = std.Io.Timestamp.now(io, .real).toMilliseconds();
_ = server.engine.ttl_sweep(now_ms) catch |err| { _ = server.engine.ttl_sweep(now_ms) catch |err| {
std.debug.print("mongo-lite: TTL sweep failed: {s}\n", .{@errorName(err)}); std.debug.print("mongo-lite: TTL sweep failed: {s}\n", .{@errorName(err)});
continue; continue;
}; };
if (server.engine.take_compact()) {
server.engine.compact() catch |err| {
std.debug.print("mongo-lite: compaction failed: {s}\n", .{@errorName(err)});
};
}
} }
} }

16
src/storage.zig
View File

@@ -341,12 +341,8 @@ pub const Log = struct {
try self.seal_block(); try self.seal_block();
} }
try self.block.appendSlice(self.gpa, buf.items); try self.block.appendSlice(self.gpa, buf.items);
// No sync here: durability is the commit point (Log.sync), which
if (!self.defer_sync) { // runs once per write command and coalesces across connections.
// Durable before the reply: seal this block and fsync.
try self.seal_block();
try self.file.sync(self.io);
}
} }
/// Compress and write the current block, then reset it. No-op when it is /// Compress and write the current block, then reset it. No-op when it is
@@ -653,6 +649,7 @@ test "append, replay, torn tail" {
}; };
try log.append_upsert("db1", "coll1", &doc_bytes, 1); try log.append_upsert("db1", "coll1", &doc_bytes, 1);
try log.append_delete("db1", "coll1", &doc_bytes, 2); try log.append_delete("db1", "coll1", &doc_bytes, 2);
try log.sync();
var seen: std.ArrayListUnmanaged(u8) = .empty; var seen: std.ArrayListUnmanaged(u8) = .empty;
defer seen.deinit(gpa); defer seen.deinit(gpa);
@@ -697,6 +694,7 @@ test "record larger than the read chunk replays" {
defer out.deinit(gpa); defer out.deinit(gpa);
try bson.write_doc(&pairs, gpa, &out); try bson.write_doc(&pairs, gpa, &out);
try log.append_upsert("db", "big", out.items, 1); try log.append_upsert("db", "big", out.items, 1);
try log.sync();
var count: usize = 0; var count: usize = 0;
const Ctx = struct { const Ctx = struct {
@@ -729,6 +727,7 @@ test "reject corrupt interior block" {
const doc_bytes = [_]u8{ 0x0E, 0, 0, 0, 0x10, '_', 'i', 'd', 0, 42, 0, 0, 0, 0 }; const doc_bytes = [_]u8{ 0x0E, 0, 0, 0, 0x10, '_', 'i', 'd', 0, 42, 0, 0, 0, 0 };
try log.append_upsert("db", "c", &doc_bytes, 1); try log.append_upsert("db", "c", &doc_bytes, 1);
try log.append_upsert("db", "c", &doc_bytes, 2); // a second block try log.append_upsert("db", "c", &doc_bytes, 2); // a second block
try log.sync();
log.close(); log.close();
// Corrupt the FIRST block's first record: flip a byte in the hashed // Corrupt the FIRST block's first record: flip a byte in the hashed
@@ -774,8 +773,12 @@ test "torn tail truncates cleanly and appends overwrite it" {
var log = try Log.open(gpa, io, path); var log = try Log.open(gpa, io, path);
const doc_bytes = [_]u8{ 0x0E, 0, 0, 0, 0x10, '_', 'i', 'd', 0, 42, 0, 0, 0, 0 }; const doc_bytes = [_]u8{ 0x0E, 0, 0, 0, 0x10, '_', 'i', 'd', 0, 42, 0, 0, 0, 0 };
// One sync per record gives two blocks: the first is complete, the
// second is the one torn by the truncation.
try log.append_upsert("db", "c", &doc_bytes, 1); try log.append_upsert("db", "c", &doc_bytes, 1);
try log.sync();
try log.append_upsert("db", "c", &doc_bytes, 2); try log.append_upsert("db", "c", &doc_bytes, 2);
try log.sync();
log.close(); log.close();
// Cut the file in the middle of the second block: a crash mid-append. // Cut the file in the middle of the second block: a crash mid-append.
@@ -806,6 +809,7 @@ test "torn tail truncates cleanly and appends overwrite it" {
// A new append overwrites from the replay end and replays cleanly. // A new append overwrites from the replay end and replays cleanly.
try log2.append_upsert("db", "c", &doc_bytes, 3); try log2.append_upsert("db", "c", &doc_bytes, 3);
try log2.sync();
seen.clearRetainingCapacity(); seen.clearRetainingCapacity();
var log3 = try Log.open(gpa, io, path); var log3 = try Log.open(gpa, io, path);
defer log3.close(); defer log3.close();

49
tests/e2e/results/phase6.txt Normal file
View File

@@ -0,0 +1,49 @@
# Phase 6 gate — mongo-lite vs MongoDB 8.3.7, 1g dataset / ~16k docs
# Ratio < 1.0 = mongo-lite faster. Reproduce: bash tests/e2e/compare-run.sh 1g 16k
# This run includes all five roadmap items (B+tree, _id index, compressed
# log, byte storage, decomposed locks). Compare: tests/e2e/results/phase1.txt.
benchmark mongo-lite mongodb ratio
insertOne (sequential) ×200 0.20 ms 4.7 ms 0.0x
bulk insert throughput 739.0 MB/s 694.5 MB/s 1.1x
docs loaded 65,536 65,536 1.0x
createIndex({k: 1}) 64.7 ms 82.1 ms 0.8x
countDocuments({}) 3.4 ms 10.9 ms 0.3x
findOne({_id: <ObjectId>}) 0.83 ms 0.72 ms 1.2x
findOne({k: 500}) (indexed) 0.62 ms 1.6 ms 0.4x
find({p: {$gte,$lt}}).count() (scan) 12.4 ms 12.0 ms 1.0x
find({}).sort({_id:-1}).limit(20) 2.3 ms 2.1 ms 1.1x
find({}, {proj}).limit(1000) 3.7 ms 4.3 ms 0.9x
aggregate $group by k 10.2 ms 12.8 ms 0.8x
updateOne({_id}) ×50 0.17 ms 0.20 ms 0.8x
updateMany({k: 7}, {$inc}) 1.9 ms 6.3 ms 0.3x
deleteOne({_id}) + insertOne 0.58 ms 5.1 ms 0.1x
node client RSS 152 MB 157 MB 1.0x
server RSS 547 MB 1274 MB
kill -9 reopen 0.8s 1.3s
db on disk 97MB 88MB
# Item 5 (decomposed locks) deltas vs phase5: none on this single-connection
# benchmark (all rows within run noise). The structure is the deliverable:
# - collections are heap-allocated (stable pointers), the docs/slab/index
# maps are guarded by a catalog rwlock (shared for commands, exclusive
# for create/drop) plus one rwlock per collection (catalog -> collection
# ordering, one collection at a time for TTL sweep and compaction).
# - appends never fsync; each write command's epilogue commits once
# (seal + fsync) with leader/follower group commit: the leader waits
# for writers mid-append, seals, and syncs once, and followers whose
# records the seal covered skip their own fsync.
# - compaction snapshots collections one at a time without the log lock
# and retries if a writer appended during the snapshot (no deadlock
# against a writer holding a collection lock), then swaps under the log
# lock.
# - durability semantics: acknowledged writes are fsynced before their
# reply (crash pair verified); an unacknowledged write may vanish, and
# a reader can observe a write before its fsync completes — ordinary
# w:1 j:true semantics, no longer "the log describes >= memory".
#
# Concurrent-write throughput (sequential insertOne per client, durable):
# 1 client ~5.1k docs/s | 8 clients ~12.5k docs/s | 32 clients ~14.8k
# docs/s. The fsync per commit still dominates sequential-per-client
# workloads; the group commit coalesces when appends overlap (different
# collections on different connections).