diff --git a/README.md b/README.md index 538493e..45c0377 100644 --- a/README.md +++ b/README.md @@ -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 traps in [ROADMAP.md](ROADMAP.md). -1. **Decompose the global lock** — one reader/writer lock covers the whole - engine and is held across fsync, compaction and reply construction. - Per-collection locks plus cross-connection group commit are the path to - using more than one core on writes. +The remaining structure is the single-file log (appends and the commit +serialize on one log lock, though appends no longer hold the collection +locks), and the acknowledged-write fsync, which dominates sequential +per-client workloads. All five roadmap items are landed. 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 smaller than MongoDB); `range-scan` 22.5 → ~12 ms (parity, best run 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. `updateMany` 15.4 → 5.5 ms. - **Top-k sort selection** and an allocation-free decorate pass, plus diff --git a/ROADMAP.md b/ROADMAP.md index 3c8de12..041980a 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -1,14 +1,13 @@ # Remaining performance work -Status: **items 1 (B+tree), 2 (ordered `_id` index), 3 (block-framed -compressed log) and 4 (byte storage) are done** — verified in -`tests/e2e/results/phase2.txt` through `phase5.txt`: updateMany 17.3 → -1.7 ms, createIndex 62 → 51 ms, `_id` sort+limit 6.2 → 2.4 ms, `db on -disk` 1025 → 97 MB, and `server RSS` 1979 → 539 MB with the range scan at -parity (best run faster than MongoDB). Only item 5 (decompose the global -lock) remains. Each is sized to be landed and verified on -its own; the ordering constraints between them are the load-bearing part, so -read those before picking one up. +Status: **all five items are done** — verified in `tests/e2e/results/phase2.txt` +through `phase6.txt`: updateMany 17.3 → 1.7 ms, createIndex 62 → 51 ms, +`_id` sort+limit 6.2 → 2.4 ms, `db on disk` 1025 → 97 MB, `server RSS` +1979 → 539 MB with the range scan at parity (best run faster than +MongoDB), and the global lock decomposed into catalog + per-collection +locks with cross-connection group commit (item 5, no regression on the +single-connection benchmark; concurrent-write throughput 1 → 8 clients +~5.1k → 12.5k docs/s, 32 clients ~14.8k). Current numbers and what they mean are in the README; the recorded baseline 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 fsync, compaction and reply construction, so writes cannot use more than one diff --git a/src/commands.zig b/src/commands.zig index d0e1385..52e5f8e 100644 --- a/src/commands.zig +++ b/src/commands.zig @@ -44,9 +44,20 @@ pub const ErrorCode = enum(i32) { /// `.none` commands must not touch the engine at all. 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 { name: []const u8, kind: CommandKind, + locks: LockShape = .{}, 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 = "killCursors", .kind = .none, .handler = cmd_kill_cursors }, // Read-only: scan the engine without mutating it. - .{ .name = "find", .kind = .read, .handler = cmd_find }, - .{ .name = "count", .kind = .read, .handler = cmd_count }, - .{ .name = "aggregate", .kind = .read, .handler = cmd_aggregate }, - .{ .name = "listDatabases", .kind = .read, .handler = cmd_list_databases }, - .{ .name = "listCollections", .kind = .read, .handler = cmd_list_collections }, - // Writes: exclusive, totally ordered. - .{ .name = "create", .kind = .write, .handler = cmd_create }, - .{ .name = "drop", .kind = .write, .handler = cmd_drop }, - .{ .name = "dropDatabase", .kind = .write, .handler = cmd_drop_database }, - .{ .name = "createIndexes", .kind = .write, .handler = cmd_create_indexes }, - .{ .name = "dropIndexes", .kind = .write, .handler = cmd_drop_indexes }, - .{ .name = "insert", .kind = .write, .handler = cmd_insert }, - .{ .name = "update", .kind = .write, .handler = cmd_update }, - .{ .name = "delete", .kind = .write, .handler = cmd_delete }, - .{ .name = "findAndModify", .kind = .write, .handler = cmd_find_and_modify }, - .{ .name = "listIndexes", .kind = .read, .handler = cmd_list_indexes }, + .{ .name = "find", .kind = .read, .locks = .{ .catalog = .shared, .coll = .shared }, .handler = cmd_find }, + .{ .name = "count", .kind = .read, .locks = .{ .catalog = .shared, .coll = .shared }, .handler = cmd_count }, + .{ .name = "aggregate", .kind = .read, .locks = .{ .catalog = .shared, .coll = .shared }, .handler = cmd_aggregate }, + .{ .name = "listDatabases", .kind = .read, .locks = .{ .catalog = .shared }, .handler = cmd_list_databases }, + .{ .name = "listCollections", .kind = .read, .locks = .{ .catalog = .shared }, .handler = cmd_list_collections }, + // Writes: the target collection exclusively; create/drop take the + // catalog exclusively (they mutate the maps). + .{ .name = "create", .kind = .write, .locks = .{ .catalog = .exclusive, .coll = .exclusive }, .handler = cmd_create }, + .{ .name = "drop", .kind = .write, .locks = .{ .catalog = .exclusive, .coll = .exclusive }, .handler = cmd_drop }, + .{ .name = "dropDatabase", .kind = .write, .locks = .{ .catalog = .exclusive }, .handler = cmd_drop_database }, + .{ .name = "createIndexes", .kind = .write, .locks = .{ .catalog = .shared, .coll = .exclusive }, .handler = cmd_create_indexes }, + .{ .name = "dropIndexes", .kind = .write, .locks = .{ .catalog = .shared, .coll = .exclusive }, .handler = cmd_drop_indexes }, + .{ .name = "insert", .kind = .write, .locks = .{ .catalog = .shared, .coll = .exclusive }, .handler = cmd_insert }, + .{ .name = "update", .kind = .write, .locks = .{ .catalog = .shared, .coll = .exclusive }, .handler = cmd_update }, + .{ .name = "delete", .kind = .write, .locks = .{ .catalog = .shared, .coll = .exclusive }, .handler = cmd_delete }, + .{ .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 @@ -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); }; - switch (cmd.kind) { - .none => return cmd.handler(ctx, msg, reply), - .read => { - try ctx.engine.lock_read(); - // Defers are block-scoped: this one is registered in the prong - // block, so it runs when the prong exits — after the handler - // returns. The shared lock is thus held for the whole command. - defer ctx.engine.unlock_read(); - return cmd.handler(ctx, msg, reply); - }, - .write => { - try ctx.engine.lock(); - defer ctx.engine.unlock(); - return cmd.handler(ctx, msg, reply); - }, + // Lock the catalog (shared for most commands, exclusive for DDL), then + // the target collection, then run the handler. The collection lock is + // taken while the catalog lock is held, so a concurrent drop can never + // free the collection out from under us. + switch (cmd.locks.catalog) { + .none => {}, + .shared => try ctx.engine.lock_catalog(false), + .exclusive => try ctx.engine.lock_catalog(true), } + 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 -/// `writeErrors` entries of a batch insert. Uses engine.dup_index (set by a -/// rejected unique-index write) when the conflict came from a secondary -/// index; otherwise it is the _id_ index. +/// `writeErrors` entries of a batch insert. Uses the collection's +/// dup_index (set by a rejected unique-index write) when the conflict came +/// 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 { var index_name: []const u8 = "_id_"; 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; key_text = try render_dup_key(ctx, reply, db_name, coll_name, name, doc); - } else { - 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); + } } + 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); } diff --git a/src/db.zig b/src/db.zig index 72ad1fb..647de36 100644 --- a/src/db.zig +++ b/src/db.zig @@ -14,6 +14,8 @@ const index = @import("index.zig"); /// would hold up to 2x its contents after doubling). const slab_segment_size = 8 * 1024 * 1024; +const LogKind = enum { upsert, delete, index_create, index_drop }; + pub const Collection = struct { /// Documents live as canonical BSON bytes in a per-collection slab of /// fixed segments; the map holds each document's flat slab offset. @@ -29,6 +31,14 @@ pub const Collection = struct { seg_starts: std.ArrayListUnmanaged(u64), /// Secondary indexes (persisted through the log). 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 /// 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 @@ -104,16 +114,40 @@ pub const Collection = 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 { gpa: std.mem.Allocator, io: std.Io, - // One writer at a time (log append + fsync, map mutation); many - // concurrent readers (find/count/aggregate scans). Writer-preferring: - // a queued writer blocks new readers rather than starving. + // Legacy whole-engine lock, used by the unit tests' explicit + // lock()/lock_read() calls. The server uses the finer-grained locks + // below: catalog (maps), per-collection (docs/slab/indexes), and + // log_lock (append + commit). 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, dbs: std.StringHashMapUnmanaged(Db), seq: u64, @@ -181,13 +215,14 @@ pub const Engine = struct { for (coll.slab.items) |*seg| seg.deinit(self.gpa); coll.slab.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. fn free_db(self: *Engine, db: *Db) void { var coll_it = db.collections.iterator(); 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.*); } db.collections.deinit(self.gpa); @@ -235,19 +270,121 @@ pub const Engine = struct { self.rwlock.unlockShared(self.io); } - /// Group commit: defer per-record fsyncs until end_batch. Callers must - /// 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, - /// so an acknowledged multi-write command is durable as a unit — the - /// same crash guarantee as the old fsync-per-record, with one sync per - /// command instead of one per document. - pub fn begin_batch(self: *Engine) void { - self.log.defer_sync = true; + // -- per-collection locking (the server's command dispatch) ------------ + + /// Lock the catalog for a command's duration. + pub fn lock_catalog(self: *Engine, exclusive: bool) !void { + if (exclusive) { + try self.catalog_lock.lock(self.io); + } else { + try self.catalog_lock.lockShared(self.io); + } } - pub fn end_batch(self: *Engine) !void { - self.log.defer_sync = false; + pub fn unlock_catalog(self: *Engine, exclusive: bool) void { + 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(); + 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. @@ -291,7 +428,7 @@ pub const Engine = struct { const id_key = try bson.serialize_value(self.gpa, id_value); var stored = false; 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, // before anything is logged or mutated. @@ -325,7 +462,7 @@ pub const Engine = struct { for (built_list.items) |*b| { if (!b.ix.unique) continue; 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; }; } @@ -336,9 +473,9 @@ pub const Engine = struct { try b.ix.reserve_for(self.gpa, b.built.entries.items); } - // 5. Log (and sync) before anything becomes visible. - self.seq += 1; - try self.log.append_upsert(db_name, coll_name, doc_bytes, self.seq); + // 5. Log (and sync) before anything becomes visible. The append + // takes the log lock; durability (fsync) is the command's commit. + try self.log_append(.upsert, db_name, coll_name, doc_bytes); // 6. Replace drops the old document (and its index entries). if (mode == .replace) self.evict_doc(coll, id_key); @@ -353,7 +490,7 @@ pub const Engine = struct { b.ix.insert_entries(&b.built); } stored = true; - try self.maybe_compact(); + self.note_compact(); } /// 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 { 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; // 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; defer id_doc.deinit(self.gpa); 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, self.seq); + try self.log_append(.delete, db_name, coll_name, id_doc.items); self.evict_doc(coll, id_key); // Deletes grow the log too. Without this a delete-heavy workload // never compacts, because only upsert and ttl_sweep used to check. - try self.maybe_compact(); + self.note_compact(); return true; } 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; - 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 { @@ -403,8 +539,8 @@ pub const Engine = struct { 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; - var removed = db.collections.fetchRemove(coll_name) orelse return false; - self.free_collection(&removed.value); + const removed = db.collections.fetchRemove(coll_name) orelse return false; + self.free_collection(removed.value); self.gpa.free(removed.key); return true; } @@ -451,8 +587,7 @@ pub const Engine = struct { var spec_bytes: std.ArrayListUnmanaged(u8) = .empty; defer spec_bytes.deinit(self.gpa); try ix.write_spec(self.gpa, &spec_bytes); - self.seq += 1; - try self.log.append_index_create(db_name, coll_name, spec_bytes.items, self.seq); + try self.log_append(.index_create, db_name, coll_name, spec_bytes.items); coll.indexes.appendAssumeCapacity(ix); committed = true; @@ -463,15 +598,14 @@ pub const Engine = struct { /// 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 { 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; const name_pairs = [_]bson.Pair{.{ .key = "name", .value = .{ .string = index_name } }}; var name_doc: std.ArrayListUnmanaged(u8) = .empty; defer name_doc.deinit(self.gpa); 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, self.seq); + try self.log_append(.index_drop, db_name, coll_name, name_doc.items); _ = coll.remove_index(self.gpa, index_name); return true; @@ -488,6 +622,30 @@ pub const Engine = struct { /// the next sweep. pub fn ttl_sweep(self: *Engine, now_ms: i64) !usize { 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 // that `Entry.id` points at, which would leave the rest of the batch // pointing into freed memory. @@ -497,57 +655,43 @@ pub const Engine = struct { ids.deinit(self.gpa); } - 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| { - for (ids.items) |id| self.gpa.free(id); - ids.clearRetainingCapacity(); - - for (coll_entry.value_ptr.indexes.items) |*ix| { - const ttl = ix.ttl orelse continue; - const cutoff: i128 = @as(i128, now_ms) - @as(i128, ttl) * 1000; - // bson compare order ranks datetime above null, numbers - // and strings and below only timestamp and maxKey, so - // datetimes form a contiguous band in the encoded key - // order: seek the minimum datetime and stop when the - // 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; - } + for (coll.indexes.items) |*ix| { + const ttl = ix.ttl orelse continue; + const cutoff: i128 = @as(i128, now_ms) - @as(i128, ttl) * 1000; + // bson compare order ranks datetime above null, numbers + // and strings and below only timestamp and maxKey, so + // datetimes form a contiguous band in the encoded key + // order: seek the minimum datetime and stop when the + // 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)); } } - // A TTL-only workload never reaches the threshold check in `upsert`, - // so the log would otherwise grow without bound. - if (deleted > 0) try self.maybe_compact(); - return deleted; + if (ids.items.len == 0) return 0; + + // 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; + + 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 { @@ -570,13 +714,15 @@ pub const Engine = struct { try self.dbs.put(self.gpa, db_key, .{ .collections = .empty }); 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); 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); 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 @@ -612,7 +758,12 @@ pub const Engine = struct { /// perfectly compact file for nothing. `compact` reports how much it /// reclaimed; when that is little, we back the baseline off /// 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 on-disk size: a compressed log would otherwise stay under any // 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 // for by the space it reclaims. 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. - /// Callers must hold the write lock. + /// Whether a compaction is wanted; clears the flag. Racy by design (two + /// 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 { const tmp_path = try std.fmt.allocPrint(self.gpa, "{s}.tmp", .{self.log.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 (db_it.next()) |db_entry| { - var coll_it = db_entry.value_ptr.collections.iterator(); - while (coll_it.next()) |coll_entry| { - // Re-emit the index definitions first: a compacted log that - // dropped them would resurrect the collections without - // indexes on replay. - for (coll_entry.value_ptr.indexes.items) |*ix| { - var spec_bytes: std.ArrayListUnmanaged(u8) = .empty; - defer spec_bytes.deinit(self.gpa); - try ix.write_spec(self.gpa, &spec_bytes); - try new_log.append_index_create(db_entry.key_ptr.*, coll_entry.key_ptr.*, spec_bytes.items, self.seq); - } - var doc_it = coll_entry.value_ptr.docs.iterator(); - while (doc_it.next()) |doc_entry| { - // The slab bytes are the canonical serialization. - const doc_bytes = coll_entry.value_ptr.doc_bytes(doc_entry.value_ptr.*); - try new_log.append_upsert(db_entry.key_ptr.*, coll_entry.key_ptr.*, doc_bytes, self.seq); + while (true) { + 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; + + const snapshot_seq = self.seq; + try self.catalog_lock.lockShared(self.io); + var catalog_err: ?anyerror = null; + var db_it = self.dbs.iterator(); + while (db_it.next()) |db_entry| { + var coll_it = db_entry.value_ptr.collections.iterator(); + while (coll_it.next()) |coll_entry| { + const coll = coll_entry.value_ptr.*; + self.compact_snapshot_coll(coll, &new_log, db_entry.key_ptr.*, coll_entry.key_ptr.*) catch |err| { + catalog_err = err; + break; + }; } + if (catalog_err != null) break; } + 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); - // 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); - // A batch may span a compaction (a large insert crossing the - // threshold): keep the deferred-sync mode on the swapped-in log so - // the remaining batch records stay grouped with the same command. - const deferred = self.log.defer_sync; - self.log.close(); - self.log = try storage.Log.open(self.gpa, self.io, old_path); - self.log.defer_sync = deferred; - // 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; - // The rewritten log holds only live documents. - self.dead_docs = 0; - self.gpa.free(old_path); + /// Re-emit one collection's index specs and documents into the compacted + /// log, under the collection's write lock (released on every return + /// path, including errors). + fn compact_snapshot_coll(self: *Engine, coll: *Collection, new_log: *storage.Log, db_name: []const u8, coll_name: []const u8) !void { + try coll.lock.lock(self.io); + defer coll.lock.unlock(self.io); + // Re-emit the index definitions first: a compacted log that dropped + // them would resurrect the collections without indexes on replay. + for (coll.indexes.items) |*ix| { + var spec_bytes: std.ArrayListUnmanaged(u8) = .empty; + defer spec_bytes.deinit(self.gpa); + try ix.write_spec(self.gpa, &spec_bytes); + try new_log.append_index_create(db_name, coll_name, spec_bytes.items, self.seq); + } + var doc_it = coll.docs.iterator(); + while (doc_it.next()) |doc_entry| { + // The slab bytes are the canonical serialization. + const doc_bytes = coll.doc_bytes(doc_entry.value_ptr.*); + try new_log.append_upsert(db_name, coll_name, doc_bytes, self.seq); + } } /// Rebuild every empty index from the live documents. Runs after replay @@ -705,10 +898,10 @@ pub const Engine = struct { while (db_it.next()) |db_entry| { var coll_it = db_entry.value_ptr.collections.iterator(); while (coll_it.next()) |coll_entry| { - for (coll_entry.value_ptr.indexes.items) |*ix| { - try self.rebuild_index(coll_entry.value_ptr, 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, &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(); try engine.insert("app", "c", &d, &env.gen); } + try engine.commit(); 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); // 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(); 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); // 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" { @@ -1077,6 +1278,9 @@ test "compaction rewrites log and keeps data" { docs[i] = try make_doc(gpa, @intCast(i + 1), "user-{d}"); 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(); } @@ -1088,6 +1292,7 @@ test "compaction rewrites log and keeps data" { var extra = try make_doc(gpa, 5, "eve"); defer extra.deinit(); try engine2.insert("app", "users", &extra, &env.gen); + try engine2.commit(); engine2.unlock(); 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"); defer d2.deinit(); 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). var d1b = try make_user(gpa, 1, "a@x.io"); diff --git a/src/fuzz_split.zig b/src/fuzz_split.zig new file mode 100644 index 0000000..9c36ae6 --- /dev/null +++ b/src/fuzz_split.zig @@ -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); +} diff --git a/src/index.zig b/src/index.zig index dc9a175..dc8c958 100644 --- a/src/index.zig +++ b/src/index.zig @@ -83,6 +83,9 @@ const page_size = 4096; const page_data = page_size - 32; /// Records longer than a quarter of a node spill to the overflow slab. 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 /// 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). pub fn reserve_for(self: *Index, gpa: std.mem.Allocator, entries: []const Entry) !void { 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; for (entries) |e| { const rec_len: u64 = e.key.len + e.id.len; if (rec_len > inline_limit) overflow_bytes += rec_len; } - const extra_nodes: u64 = n + n / 100 + self.depth + 2; - 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))); + try self.overflow.ensureUnusedCapacity(gpa, @intCast(overflow_bytes)); } /// Insert pre-built entries (maintaining order) and drain the batch. @@ -387,7 +403,10 @@ pub const Index = struct { 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); return duplicate; } @@ -604,22 +623,25 @@ pub const Index = struct { key: []const u8, id: []const u8 = "", child: u32 = 0, - /// When set, key+id already live in the overflow slab and are - /// referenced rather than copied (moved leaf records, promoted - /// separators). - spill_off: u64 = 0, + /// When set, key+id already live in the overflow slab at this + /// offset and are referenced rather than copied (moved leaf + /// records, promoted separators). Optional rather than a 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 { key: []const u8, - spill_off: u64, + spill_off: ?u64, }; const Split = struct { /// The separator to promote, stable across node-array growth (a /// slice into the overflow slab, or a copy in the promo buffer). key: []const u8, - spill_off: u64, + spill_off: ?u64, right: u32, }; @@ -644,6 +666,14 @@ pub const Index = struct { 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 { 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 rec_len: u64 = rec.key.len + rec.id.len; var s: Slot = .{ - .off = rec.spill_off, + .off = 0, .key_len = @intCast(rec.key.len), .extra = if (rec.id.len > 0) @intCast(rec.id.len) else rec.child, .spill = false, ._pad = 0, }; - if (rec.spill_off != 0) { + if (rec.spill_off) |off| { + s.off = off; s.spill = true; } else if (rec_len > inline_limit) { 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]; std.mem.copyForwards(u8, dst_slots, src_slots); node.count -= 1; + if (node.count == 0) node.data_start = page_data; } /// 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 }; const k = self.key_of(node_id, i); @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). @@ -892,6 +956,17 @@ pub const Index = struct { fn insert_rec(self: *Index, node_id: u32, key: []const u8, id: []const u8) ?Split { const node = &self.nodes.items[node_id]; 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)) { self.store_record(node_id, self.leaf_pos(node_id, key, id), .{ .key = key, .id = id }); self.entry_count += 1; @@ -904,48 +979,72 @@ pub const Index = struct { return self.insert_separator(node_id, res); } - /// Split a full leaf: the right half (including any records equal to - /// the boundary key — lookups scan whole key bands, so equal keys may - /// live on both sides) moves to a new leaf, the boundary key is - /// promoted, and the new record is stored in whichever half holds its - /// position. - fn split_leaf(self: *Index, leaf_id: u32, key: []const u8, id: []const u8) ?Split { - const node = &self.nodes.items[leaf_id]; - const mid = @max(1, node.count / 2); + /// Split a full leaf around the record being inserted. The new record + /// takes its place in the leaf's key order and the merged run is cut + /// where the two halves come closest to equal cost, so the half the new + /// record lands in is guaranteed to have room for it. Records equal to + /// the boundary key may end up on both sides; lookups scan whole key + /// bands, so that is fine. The right half moves to a new leaf and its + /// first key is promoted. + 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 node = &self.nodes.items[leaf_id]; const right = &self.nodes.items[right_id]; right.is_leaf = 1; right.next = node.next; right.prev = leaf_id; right.parent = node.parent; } - // Move the right half; spilled records keep their slab reference. - var i: u32 = mid; - while (i < node.count) : (i += 1) { - const s = get_slot(&self.nodes.items[leaf_id], i); - self.store_record(right_id, @intCast(self.nodes.items[right_id].count), .{ - .key = self.key_of(leaf_id, i), - .id = self.id_of(leaf_id, i), - .spill_off = if (s.spill) s.off else 0, + // Move the merged tail; spilled records keep their slab reference. + var m: u32 = mid; + while (m < n) : (m += 1) { + const at: u32 = self.nodes.items[right_id].count; + if (m == pos) { + self.store_record(right_id, at, .{ .key = key, .id = id }); + continue; + } + 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. - self.repack_keep_prefix(leaf_id, mid); - // Link the chain. - const node2 = &self.nodes.items[leaf_id]; - if (node2.next != 0) self.nodes.items[node2.next].prev = right_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 }); + // Keep the merged head in the left page: its own records, plus the + // new one when that is the half it belongs to. + if (pos < mid) { + self.repack_keep_prefix(leaf_id, mid - 1); + self.store_record(leaf_id, pos, .{ .key = key, .id = id }); } 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; + + const sep = self.stable_key(right_id, 0); 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 // to a stack buffer so they survive. 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); 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)) { self.store_record(node_id, self.separator_pos(node_id, key), .{ .key = key, @@ -970,55 +1075,79 @@ pub const Index = struct { self.nodes.items[split.right].parent = node_id; return null; } - const up = self.split_internal(node_id); - // 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; + return self.split_internal(node_id, key, split.spill_off, split.right); } - /// Split a full internal node: the middle separator is promoted, the - /// first half stays, the rest moves to a new right node. - fn split_internal(self: *Index, node_id: u32) Split { - const node = &self.nodes.items[node_id]; - const s = node.count / 2; - // Copy the promoted key before the repack rewrites the page. - const mid_key = self.stable_key(node_id, s); + /// Split a full internal node around the separator being inserted: the + /// new separator joins the node's order, the merged run is cut where + /// the halves come closest to equal cost (see split_leaf), and the + /// separator at the cut is promoted -- its child becoming the right + /// node's first child. + fn split_internal(self: *Index, node_id: u32, key: []const u8, spill_off: ?u64, child: u32) Split { + 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 = &self.nodes.items[right_id]; - right.is_leaf = 0; - right.parent = node.parent; - right.first_child = get_slot(node, s).extra; - // The moved subtrees now live under the right node. - self.nodes.items[right.first_child].parent = right_id; - } - var i: u32 = s + 1; - while (i < node.count) : (i += 1) { - const slot_i = get_slot(&self.nodes.items[node_id], i); - self.nodes.items[slot_i.extra].parent = right_id; - self.store_record(right_id, @intCast(self.nodes.items[right_id].count), .{ - .key = self.key_of(node_id, i), - .child = slot_i.extra, - .spill_off = if (slot_i.spill) slot_i.off else 0, + self.nodes.items[right_id].is_leaf = 0; + self.nodes.items[right_id].parent = self.nodes.items[node_id].parent; + // The promoted separator's child heads the right node. + const mid_child: u32 = if (mid == pos) + child + else + get_slot(&self.nodes.items[node_id], if (mid < pos) mid else mid - 1).extra; + self.nodes.items[right_id].first_child = mid_child; + self.nodes.items[mid_child].parent = right_id; + + var m: u32 = mid + 1; + while (m < n) : (m += 1) { + const at: u32 = self.nodes.items[right_id].count; + if (m == pos) { + self.store_record(right_id, at, .{ .key = key, .child = child, .spill_off = spill_off }); + 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); - return .{ .key = mid_key.key, .spill_off = mid_key.spill_off, .right = right_id }; + if (pos < mid) { + 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 @@ -1097,9 +1226,9 @@ pub const Index = struct { /// The internal root emptied: swap in a fresh empty leaf as the root. fn replace_root_with_leaf(self: *Index) void { - self.root = self.alloc_node(); - self.nodes.items[self.root].is_leaf = 1; - self.nodes.items[self.root].parent = 0; + // Removal reserves no capacity, so this must not allocate a node: + // re-use the emptied root page as the empty root leaf. + self.nodes.items[self.root] = empty_node(1); self.first_leaf = self.root; self.leaf_count = 1; self.depth = 0; @@ -1114,6 +1243,10 @@ pub const Index = struct { self.staging.clearRetainingCapacity(); } 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 { id: u32, @@ -1126,7 +1259,7 @@ pub const Index = struct { var prev: u32 = 0; var lit = self.staging.items; while (lit.len > 0) { - const leaf = self.alloc_node(); + const leaf = try self.alloc_node_grow(gpa); self.nodes.items[leaf].is_leaf = 1; const first_key = lit[0].key; // Fill until the next record would not fit. @@ -1158,7 +1291,7 @@ pub const Index = struct { defer next.deinit(gpa); var i: usize = 0; 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[level.items[i].id].parent = node; const first_key = level.items[i].first_key; @@ -1172,7 +1305,6 @@ pub const Index = struct { self.store_record(node, @intCast(slots), .{ .key = level.items[j].first_key, .child = level.items[j].id, - .spill_off = 0, }); self.nodes.items[level.items[j].id].parent = node; slots += 1; @@ -2360,6 +2492,106 @@ test "TTL spec rejects compound keys and bad expireAfterSeconds" { 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" { // The implicit _id_ index is a normal Index (keys = [_id: 1]) passed to // plan separately from the secondaries. Its encoded keys are canonical, diff --git a/src/server.zig b/src/server.zig index 753b6b2..ef8435a 100644 --- a/src/server.zig +++ b/src/server.zig @@ -58,23 +58,26 @@ pub const Server = struct { }; /// 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 -/// serialized with commands exactly like any other write; a sweep failure is -/// logged rather than fatal, since the next one will retry. +/// the group is canceled. The sweep takes the catalog lock and one +/// collection's write lock at a time, exactly like a write command; a sweep +/// failure is logged rather than fatal, since the next one will retry. fn ttl_monitor(io: std.Io, server: *Server) error{Canceled}!void { const interval: std.Io.Duration = .fromSeconds(server.ttl_sweep_secs); while (true) { // Sleep first: at startup the engine has just replayed the log, and // 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 server.engine.lock(); - defer server.engine.unlock(); const now_ms = std.Io.Timestamp.now(io, .real).toMilliseconds(); _ = server.engine.ttl_sweep(now_ms) catch |err| { std.debug.print("mongo-lite: TTL sweep failed: {s}\n", .{@errorName(err)}); continue; }; + if (server.engine.take_compact()) { + server.engine.compact() catch |err| { + std.debug.print("mongo-lite: compaction failed: {s}\n", .{@errorName(err)}); + }; + } } } diff --git a/src/storage.zig b/src/storage.zig index eaae302..04865f9 100644 --- a/src/storage.zig +++ b/src/storage.zig @@ -341,12 +341,8 @@ pub const Log = struct { try self.seal_block(); } try self.block.appendSlice(self.gpa, buf.items); - - if (!self.defer_sync) { - // Durable before the reply: seal this block and fsync. - try self.seal_block(); - try self.file.sync(self.io); - } + // No sync here: durability is the commit point (Log.sync), which + // runs once per write command and coalesces across connections. } /// 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_delete("db1", "coll1", &doc_bytes, 2); + try log.sync(); var seen: std.ArrayListUnmanaged(u8) = .empty; defer seen.deinit(gpa); @@ -697,6 +694,7 @@ test "record larger than the read chunk replays" { defer out.deinit(gpa); try bson.write_doc(&pairs, gpa, &out); try log.append_upsert("db", "big", out.items, 1); + try log.sync(); var count: usize = 0; 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 }; try log.append_upsert("db", "c", &doc_bytes, 1); try log.append_upsert("db", "c", &doc_bytes, 2); // a second block + try log.sync(); log.close(); // 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); 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.sync(); try log.append_upsert("db", "c", &doc_bytes, 2); + try log.sync(); log.close(); // 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. try log2.append_upsert("db", "c", &doc_bytes, 3); + try log2.sync(); seen.clearRetainingCapacity(); var log3 = try Log.open(gpa, io, path); defer log3.close(); diff --git a/tests/e2e/results/phase6.txt b/tests/e2e/results/phase6.txt new file mode 100644 index 0000000..381befc --- /dev/null +++ b/tests/e2e/results/phase6.txt @@ -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: }) 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).