diff --git a/src/assert.zig b/src/assert.zig new file mode 100644 index 0000000..4c48e39 --- /dev/null +++ b/src/assert.zig @@ -0,0 +1,36 @@ +//! Assertions that survive the default ReleaseFast build. +//! +//! `std.debug.assert` lowers to `unreachable`, which in ReleaseFast (this +//! project's default -- see build.zig) is not a skipped check but a promise to +//! the optimizer that the condition holds. That is exactly the wrong lowering +//! for a durability invariant that might actually be false: the compiler is +//! then free to optimize on a lie. So the checks below stay active in every +//! optimize mode. +//! +//! Use these for invariants whose violation means the database is already +//! corrupt, where crashing loudly beats continuing and writing wrong bytes to +//! disk. Every current use sits on a path that already takes a lock or fsyncs, +//! so the branch is noise. Keep `std.debug.assert` for hot inner loops (see +//! index.zig), where the cost is real and a wrong answer is not persistent. + +const std = @import("std"); + +/// Panic unless `ok`. Active in every optimize mode; see the module comment. +pub fn assert(ok: bool) void { + if (!ok) @panic("mongo-lite: assertion failed"); +} + +/// Panic unless `ok`, naming the invariant that broke. Prefer this where the +/// condition alone does not say what went wrong -- the message lands in the +/// crash output, which may be all an operator has to go on. +pub fn assert_msg(ok: bool, comptime message: []const u8) void { + if (!ok) @panic("mongo-lite: assertion failed: " ++ message); +} + +test "assert passes on true and is callable in every mode" { + assert(true); + assert_msg(true, "trivially true"); + // The failing side cannot be tested in-process: it panics by design. + // Its behavior is covered by the invariants it guards in db.zig. + try std.testing.expect(true); +} diff --git a/src/commands.zig b/src/commands.zig index db158ed..60ad20e 100644 --- a/src/commands.zig +++ b/src/commands.zig @@ -10,6 +10,9 @@ const Collection = db.Collection; const query = @import("query.zig"); const update = @import("update.zig"); const index = @import("index.zig"); +// Always active, including in the default ReleaseFast build -- see assert.zig. +const assert = @import("assert.zig").assert; +const assert_msg = @import("assert.zig").assert_msg; pub const Context = struct { gpa: std.mem.Allocator, @@ -153,9 +156,24 @@ pub fn dispatch(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void { 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. + // Durability (seal + fsync) coalesces across concurrent writers. A + // commit error deliberately wins over the handler's captured `result`: + // whether the write reached disk matters more to the client than why + // the write itself was unhappy. + // commit() asserts its own postcondition (committed_seq >= this + // command's seq) internally. Re-checking it here is not possible + // without the log lock, and taking it just to assert would add a real + // race in exchange for a weaker check than the one already made. try ctx.engine.commit(); - if (ctx.engine.take_compact()) try ctx.engine.compact(); + // The write is durable by now, so a compaction failure is a maintenance + // problem and not the client's. Report it and hand the request back + // rather than turning an applied write into an error the client retries. + if (ctx.engine.take_compact()) { + ctx.engine.compact() catch |err| { + std.debug.print("mongo-lite: compaction failed: {s}\n", .{@errorName(err)}); + ctx.engine.request_compact(); + }; + } } return result; } @@ -561,12 +579,11 @@ fn cmd_insert(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void { var write_errors: std.ArrayListUnmanaged(bson.Value) = .empty; defer write_errors.deinit(reply.arena_alloc()); - // Group commit: one fsync for the whole batch instead of one per - // document. end_batch runs on every return path, so even a failed doc - // (writeErrors) or a hard error still syncs what was appended. - ctx.engine.begin_batch(); - defer ctx.engine.end_batch() catch {}; - + // Group commit: one fsync for the whole batch instead of one per document. + // Nothing to open or close here -- appends never sync, and the dispatch + // epilogue is the single commit point. It runs on every return path, so a + // failed doc (writeErrors) or a hard error still syncs what was appended, + // and does it after the collection lock is released. for (docs, 0..) |*doc, i| { if (ctx.engine.insert(db_name, coll_name, doc, ctx.oid_gen)) |_| { inserted += 1; @@ -624,19 +641,24 @@ fn cmd_find(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void { // Sorting and emitting need the documents as trees; materialize the // matched page into the reply arena (the slab itself is never copied). - const coll = ctx.engine.get_collection(db_name, coll_name) orelse { - // A find on a namespace that does not exist is an empty cursor, not - // an error -- and above all not a reply with no `ok` at all, which - // is what returning here without one sends. - const none: []const *const bson.Document = &.{}; - try emit_docs_tree(reply, db_name, coll_name, proj_pairs, none); - return reply.put_ok(); - }; + // A find on a namespace that does not exist is an empty cursor, not an + // error and not a reply missing `ok`: the scan above matched nothing, so + // falling through to the emit at the end of this function says exactly + // that without a second exit path to keep in step with it. + const coll = ctx.engine.get_collection(db_name, coll_name); + // The scan above ran against this same collection with the catalog lock + // held, so a missing collection means nothing matched. Asserted rather than + // left implicit: if that ever stops holding, the loop below silently emits + // an empty page for a query that did match, which is the hardest kind of + // wrong answer to notice. + if (coll == null) assert_msg(matched.items.len == 0, "find matched documents in a collection that does not exist"); // Lives in the reply arena; freed with it. var tree_docs: std.ArrayListUnmanaged(*const bson.Document) = .empty; const arena = reply.arena_alloc(); - for (matched.items) |off| { - try tree_docs.append(arena, try doc_tree(arena, coll, off)); + if (coll) |c| { + for (matched.items) |off| { + try tree_docs.append(arena, try doc_tree(arena, c, off)); + } } if (sort_keys.len > 0 and !index_sorted) { // Selecting the page is much cheaper than ordering everything when @@ -771,10 +793,8 @@ fn cmd_update(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void { var write_errors: std.ArrayListUnmanaged(bson.Value) = .empty; defer write_errors.deinit(reply.arena_alloc()); - // Group commit for multi-document updates: one fsync per command. - ctx.engine.begin_batch(); - defer ctx.engine.end_batch() catch {}; - + // Group commit for multi-document updates: one fsync per command, issued + // by the dispatch epilogue once the collection lock is released. for (specs, 0..) |*spec, si| { const q = doc_arg(spec.get("q")) orelse return bad_value(reply, "update spec requires q"); const u_doc = doc_arg(spec.get("u")) orelse return bad_value(reply, "update spec requires u"); @@ -850,10 +870,8 @@ fn cmd_delete(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void { const specs = try batch_arg(msg, reply, "delete", "deletes") orelse return; var n_deleted: i64 = 0; - // Group commit for multi-document deletes: one fsync per command. - ctx.engine.begin_batch(); - defer ctx.engine.end_batch() catch {}; - + // Group commit for multi-document deletes: one fsync per command, issued + // by the dispatch epilogue once the collection lock is released. for (specs) |*spec| { const q = doc_arg(spec.get("q")) orelse return bad_value(reply, "delete spec requires q"); const limit = int_value(spec.get("limit")) orelse 1; @@ -1478,9 +1496,9 @@ fn duplicate_key_message(ctx: *Context, reply: *wire.Reply, db_name: []const u8, 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); - return e11000_message(reply, db_name, coll_name, index_name, key_text); + index_name = name; + key_text = try render_dup_key(ctx, reply, db_name, coll_name, name, doc); + 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); @@ -2098,7 +2116,6 @@ test "unique index constraint returns 11000 through insert and update" { try testing.expectEqual(@as(i64, 11000), bson.get_pair(up_errs[0].doc, "code").?.int32); } - /// Free a list of serialized ids (each element is gpa-owned). fn free_id_list(gpa: std.mem.Allocator, list: *std.ArrayListUnmanaged([]u8)) void { for (list.items) |id| gpa.free(id); @@ -2184,7 +2201,7 @@ test "indexed queries are equivalent to scans over a mixed corpus" { .{ .key = "_id", .value = .{ .int32 = 3 } }, .{ .key = "a", .value = .{ .double = 30.0 } }, .{ .key = "b", .value = .null }, - .{ .key = "tags", .value = .{ .array = &.{ .{ .string = "a" } } } }, + .{ .key = "tags", .value = .{ .array = &.{.{ .string = "a" }} } }, } }, .{ .doc = &.{ .{ .key = "_id", .value = .{ .string = "s4" } }, @@ -2228,13 +2245,13 @@ test "indexed queries are equivalent to scans over a mixed corpus" { .{ .pairs = &.{.{ .key = "a", .value = .{ .doc = &.{.{ .key = "$lt", .value = .{ .int32 = 30 } }} } }} }, .{ .pairs = &.{.{ .key = "a", .value = .{ .doc = &.{.{ .key = "$lte", .value = .{ .int32 = 30 } }} } }} }, .{ .pairs = &.{.{ .key = "a", .value = .{ .doc = &.{.{ .key = "$in", .value = .{ .array = &.{ .{ .int32 = 10 }, .{ .int32 = 30 } } } }} } }} }, - .{ .pairs = &.{.{ .key = "a", .value = .{ .doc = &.{ .{ .key = "$gt", .value = .{ .int32 = 5 } }, .{ .key = "$lt", .value = .{ .int32 = 25 } } } }} } }, + .{ .pairs = &.{.{ .key = "a", .value = .{ .doc = &.{ .{ .key = "$gt", .value = .{ .int32 = 5 } }, .{ .key = "$lt", .value = .{ .int32 = 25 } } } } }} }, .{ .pairs = &.{.{ .key = "b", .value = .{ .string = "x" } }} }, .{ .pairs = &.{.{ .key = "b", .value = .null }} }, .{ .pairs = &.{ .{ .key = "a", .value = .{ .int32 = 10 } }, .{ .key = "b", .value = .{ .string = "x" } } } }, .{ .pairs = &.{.{ .key = "tags", .value = .{ .string = "a" } }} }, .{ .pairs = &.{.{ .key = "tags", .value = .{ .array = &.{ .{ .string = "a" }, .{ .string = "b" } } } }} }, - .{ .pairs = &.{.{ .key = "tags", .value = .{ .doc = &.{.{ .key = "$all", .value = .{ .array = &.{ .{ .string = "a" } } } }} } }} }, + .{ .pairs = &.{.{ .key = "tags", .value = .{ .doc = &.{.{ .key = "$all", .value = .{ .array = &.{.{ .string = "a" }} } }} } }} }, .{ .pairs = &.{.{ .key = "a", .value = .{ .doc = &.{.{ .key = "$regex", .value = .{ .string = "^1" } }} } }} }, .{ .pairs = &.{.{ .key = "c", .value = .{ .doc = &.{.{ .key = "$exists", .value = .{ .bool = false } }} } }} }, .{ .pairs = &.{.{ .key = "a", .value = .null }} }, diff --git a/src/db.zig b/src/db.zig index 02266a6..02cc870 100644 --- a/src/db.zig +++ b/src/db.zig @@ -9,6 +9,12 @@ const std = @import("std"); const bson = @import("bson.zig"); const storage = @import("storage.zig"); const index = @import("index.zig"); +// Always active, including in the default ReleaseFast build -- see assert.zig +// for why std.debug.assert is the wrong tool for these invariants. +const assert = @import("assert.zig").assert; +// For the durability invariants: a panic carries no expression text, so the +// message is all an operator gets. +const assert_msg = @import("assert.zig").assert_msg; /// One slab segment; slack is bounded by this (a geometric-growth array /// would hold up to 2x its contents after doubling). @@ -73,12 +79,19 @@ pub const Collection = struct { try self.slab.append(gpa, .empty); try self.seg_starts.append(gpa, 0); } - const last = &self.slab.items[self.slab.items.len - 1]; - if (last.items.len + bytes.len > slab_segment_size) { + // Length of the last segment as a value, never as a pointer into + // slab.items: appending the next segment below may reallocate that + // list, which would dangle a pointer taken before the append and + // corrupt the new segment's start offset (and with it every + // doc_bytes lookup in that segment — reads that surfaced as + // InvalidBson, or a crash in Debug builds). + const last_len = self.slab.items[self.slab.items.len - 1].items.len; + if (last_len + bytes.len > slab_segment_size) { try self.slab.append(gpa, .empty); - try self.seg_starts.append(gpa, self.seg_starts.items[self.seg_starts.items.len - 1] + last.items.len); + try self.seg_starts.append(gpa, self.seg_starts.items[self.seg_starts.items.len - 1] + last_len); return self.slab_append(gpa, bytes); } + const last = &self.slab.items[self.slab.items.len - 1]; const off = self.seg_starts.items[self.seg_starts.items.len - 1] + last.items.len; try last.appendSlice(gpa, bytes); return off; @@ -150,17 +163,23 @@ pub const Engine = struct { 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, + /// Atomic because it is set under a *collection* lock (see `note_compact`) + /// but read by the epilogue holding no lock at all. + compact_pending: std.atomic.Value(bool) = std.atomic.Value(bool).init(false), + /// Set while a compaction runs, so only one runs at a time. Compactions + /// share one tmp path and each ends in a rename onto the log, so two at + /// once would publish one compaction's half-written file as the database. + compacting: std.atomic.Value(bool) = std.atomic.Value(bool).init(false), log: storage.Log, dbs: std.StringHashMapUnmanaged(Db), seq: u64, /// Floor for the compaction trigger. The real trigger also scales with - /// the live data size — see `maybe_compact`. + /// the live data size — see `note_compact`. compact_threshold: u64, /// Documents currently resident across every collection, and documents /// superseded or deleted since the last compaction. Their ratio is the /// share of the log that is garbage, which is what decides whether a - /// rewrite is worth doing — see `maybe_compact`. + /// rewrite is worth doing — see `note_compact`. live_docs: u64 = 0, dead_docs: u64 = 0, /// Set to the failing index's own stable name when an upsert is @@ -204,7 +223,12 @@ pub const Engine = struct { /// and secondary indexes (whose entries alias the documents — freed /// first). fn free_collection(self: *Engine, coll: *Collection) void { - // Dropping a collection turns all of its records into garbage. + // Dropping a collection turns all of its records into garbage. The + // engine's live count includes every collection's documents, so it can + // never be smaller than this one's -- and a u64 underflow here would + // read as an astronomically large live count, permanently suppressing + // compaction rather than crashing. + assert_msg(self.live_docs >= coll.docs.count(), "dropping a collection would underflow the engine's live count"); self.live_docs -= coll.docs.count(); self.dead_docs += coll.docs.count(); coll.id_index.deinit(self.gpa); @@ -248,6 +272,8 @@ pub const Engine = struct { for (coll.indexes.items) |*ix| ix.remove_doc(self.gpa, old_bytes, old.key); self.gpa.free(old.key); // This document's log record (and its slab bytes) just became garbage. + // The fetchRemove above succeeded, so a live document was counted. + assert_msg(self.live_docs >= 1, "evicting a document would underflow the engine's live count"); self.live_docs -= 1; self.dead_docs += 1; } @@ -330,6 +356,8 @@ pub const Engine = struct { pub fn commit(self: *Engine) !void { try self.commit_lock.lock(self.io); defer self.commit_lock.unlock(self.io); + // A commit can never have sealed more than was ever appended. + assert_msg(self.committed_seq <= self.seq, "commit claims to have sealed more than was appended"); // Everything this command appended is at or below the current seq. // Read it before waiting, so a leader that sealed before this // command's appends cannot be mistaken for one that covered them. @@ -338,7 +366,13 @@ pub const Engine = struct { self.log_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; + // + // The error propagates rather than being swallowed: this function + // returning success is what tells the caller its write is on disk, so + // reporting success after a failed wait acknowledges a write that was + // never synced. A canceled connection has no reason to wait out + // another writer's commit, so cancelable is right here. + while (self.committing) try self.commit_done.wait(self.io, &self.commit_lock); if (self.committed_seq >= want) { return; // a concurrent commit already synced this writer's records } @@ -349,20 +383,44 @@ pub const Engine = struct { defer { if (!done) { self.committing = false; - self.commit_done.signal(self.io); + // Broadcast: followers sleeping on `committing` all need to + // re-check it, not just one of them. + self.commit_done.broadcast(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; + // + // Uncancelable: `committing` is set, so every other writer is now + // parked behind this leader. Abandoning the commit here would strand + // them for a full extra round trip, and the drain is bounded anyway -- + // an in-flight append only holds log_lock long enough to buffer its + // record. Finishing is strictly better than bailing out. + while (self.pending_appends.load(.acquire) > 0) { + self.commit_done.waitUncancelable(self.io, &self.commit_lock); + } + // The drain is what makes the seal below cover every append in flight. + // Not asserted as pending_appends == 0 here: a new append can start + // at any moment (it increments without commit_lock), so a fresh + // writer can be in flight between the drain's last check and this + // point. The seal still covers every append that wrote bytes before + // the sync — appends serialize with it on log_lock — and any append + // that starts after it is sealed by its own commit. try self.log_lock.lock(self.io); defer self.log_lock.unlock(self.io); try self.log.sync(); // Appends drained above, so the seal covered every record written so // far -- including any that arrived while this leader waited. self.committed_seq = self.seq; + // This writer's own records are now durable: the postcondition the + // caller relies on before it acknowledges the write. Paired with the + // same check in the dispatch epilogue (see commands.zig). + assert_msg(self.committed_seq >= want, "commit returning success without sealing this writer's records"); done = true; self.committing = false; - self.commit_done.signal(self.io); + // Broadcast: every follower waiting on `committing` must wake to see + // it cleared — a single signal would wake only one and strand the + // rest. + self.commit_done.broadcast(self.io); } /// Log an append (and its seq increment) under the log lock, marking @@ -370,13 +428,36 @@ pub const Engine = struct { 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 { + // The increment above pairs with this decrement on every return + // path, so the count can never be zero here. + assert_msg(self.pending_appends.load(.acquire) > 0, "log_append decrementing an already-zero in-flight count"); _ = self.pending_appends.fetchSub(1, .acq_rel); - // Wake a commit leader waiting for in-flight appends. - self.commit_done.signal(self.io); + // Wake a commit leader waiting for in-flight appends. The + // signal must be delivered while holding commit_lock: a leader + // between its pending_appends check and its wait() still holds + // the lock, so a signal here can never land in that window and + // be lost (which froze every writer once a few connections + // committed concurrently). The leader's wait() releases the + // lock, so this lock only blocks until it starts waiting. + // Broadcast rather than signal: if a follower waiting on + // `committing` snatches the single wakeup, the leader would + // sleep forever even with pending_appends back at zero. + // + // lockUncancelable, not lock: this is a cleanup path, and the + // cancelable variant can fail. Swallowing that failure and + // unlocking anyway would release a mutex we never took, which + // Mutex.unlock treats as `unreachable` -- a panic in ReleaseSafe + // and silent memory corruption in the default ReleaseFast build. + self.commit_lock.lockUncancelable(self.io); + self.commit_done.broadcast(self.io); + self.commit_lock.unlock(self.io); } try self.log_lock.lock(self.io); defer self.log_lock.unlock(self.io); self.seq += 1; + // Seqs start at 1 and only ever increase; 0 means "nothing appended", + // which is what committed_seq is compared against. + assert_msg(self.seq > 0, "log_append produced a zero seq"); switch (kind) { .upsert => try self.log.append_upsert(db, coll, doc, self.seq), .delete => try self.log.append_delete(db, coll, doc, self.seq), @@ -385,16 +466,6 @@ pub const Engine = struct { } } - /// 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. /// Generates an ObjectId _id when absent. pub fn insert(self: *Engine, db_name: []const u8, coll_name: []const u8, doc: *const bson.Document, oid_gen: *bson.ObjectIdGen) !void { @@ -787,16 +858,22 @@ 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; - self.compact_pending = true; + self.compact_pending.store(true, .release); } - /// 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. + /// Whether a compaction is wanted; clears the flag atomically so only one + /// of several concurrent writers takes the request. `compact` excludes + /// itself besides, so a caller that wins here still yields to a rewrite + /// already in progress. pub fn take_compact(self: *Engine) bool { - const p = self.compact_pending; - self.compact_pending = false; - return p; + return self.compact_pending.swap(false, .acq_rel); + } + + /// Re-arm the compaction request. For a caller that took the request but + /// could not carry it out (a failed or abandoned rewrite), so the garbage + /// is reconsidered by a later, quieter epilogue instead of being forgotten. + pub fn request_compact(self: *Engine) void { + self.compact_pending.store(true, .release); } /// Rewrite the log with only live documents, atomically swapping the @@ -806,18 +883,37 @@ pub const Engine = struct { /// 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. + /// + /// Only one compaction runs at a time; a second caller returns immediately. pub fn compact(self: *Engine) !void { + // Claim the compaction, or leave it to the one already running. The + // guard lives here rather than in `take_compact` so that every caller + // is covered, including tests that invoke compact directly. Two at once + // would share the tmp path below: one's delete-and-recreate unlinks the + // other's file while it still holds the fd, and then both rename that + // path onto the log -- publishing a half-written file as the database. + // A caller that loses this race has nothing to do anyway: the winner's + // rewrite covers its garbage too. + if (self.compacting.swap(true, .acq_rel)) return; + defer self.compacting.store(false, .release); + const tmp_path = try std.fmt.allocPrint(self.gpa, "{s}.tmp", .{self.log.path}); defer self.gpa.free(tmp_path); - 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); + // Bounded: every attempt rewrites the whole log before the seq check + // below can reject it, so an unbounded retry livelocks under sustained + // writes at one full rewrite per attempt. Giving up re-arms the request + // for a quieter epilogue -- the log stays correct, just larger. + const attempt_max: u32 = 8; + var attempt: u32 = 0; + while (attempt < attempt_max) : (attempt += 1) { + // Truncating create, not open: a tmp file left by a crashed or + // retried compaction is longer than what we are about to write, and + // `open` would keep its tail. Those leftover blocks are intact and + // hash-correct, so replay would apply them as live records once the + // rename publishes this file. + var new_log = try storage.Log.create(self.gpa, self.io, tmp_path); defer new_log.close(); - // 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); @@ -845,13 +941,14 @@ pub const Engine = struct { continue; // a writer appended during the snapshot; retry } errdefer self.log_lock.unlock(self.io); + assert_msg(self.seq == snapshot_seq, "compaction snapshot raced an append"); // Durable before the rename makes it the database. try new_log.sync(); - // After the sync, not before: sync seals the open block, and - // that seal is what moves end_pos past it. Reading the position - // first leaves appends writing over the compacted file's last - // block, which then vanishes on the next replay. - const new_end_pos = new_log.end_pos; + // Nothing may follow the last sealed block: replay walks blocks + // until it runs off the end, so a trailing byte range would be + // applied as live data. Checked here, right where the file is about + // to become the database, rather than trusting the truncating open. + assert_msg(try new_log.file.length(self.io) == new_log.end_pos, "compacted log has bytes past its last sealed block"); try std.Io.Dir.renameAbsolute(tmp_path, self.log.path, self.io); // Persist the rename: fsync the parent directory so the new @@ -864,18 +961,34 @@ pub const Engine = struct { 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; - // The rewritten file was synced before the rename, so everything - // applied so far is durable. - self.committed_seq = self.seq; + // Log.open does not replay, so it starts at an empty file's + // end_pos; continue appending where the compacted file actually + // ends. Read from new_log rather than a local captured earlier: + // the sync above is what seals the last block and moves end_pos + // past it, so a position read before it would leave appends + // overwriting that block, which then vanishes on the next replay. + self.log.end_pos = new_log.end_pos; + // The rewritten file was synced before the rename, and the check + // above proved no append slipped in, so everything up to the + // snapshot's seq is durable. + self.committed_seq = snapshot_seq; + assert_msg(self.committed_seq <= self.seq, "compaction left committed_seq past the log's seq"); // The rewritten log holds only live documents. self.dead_docs = 0; self.gpa.free(old_path); self.log_lock.unlock(self.io); return; } + + // Every attempt lost the race against a concurrent writer. Not an + // error: the log is intact and still correct, only bigger than we would + // like, so hand the request back rather than failing the write whose + // epilogue called us. + self.request_compact(); + std.debug.print( + "mongo-lite: compaction gave up after {d} attempts (concurrent writes); will retry\n", + .{attempt_max}, + ); } /// Re-emit one collection's index specs and documents into the compacted @@ -1388,6 +1501,146 @@ test "concurrent readers and writers on a threaded Io" { } } +test "compact yields to a compaction already in flight" { + // The guard's contract, checked deterministically. Two compactions at once + // share one tmp path and each ends in a rename onto the log, so the second + // truncates and rewrites the file the first is about to publish -- and the + // first then renames whatever the second left there over the live log. + // + // The real interleaving is hard to force: `compact_snapshot_coll` holds each + // collection's write lock while writing its snapshot, so two compactions + // serialize there, and an insert cannot re-arm `compact_pending` while that + // lock is held either. The overlap window is only between the end of one + // snapshot and its rename. Rather than race for it, drive the flag directly + // and pin what the guard promises. + var threaded: std.Io.Threaded = .init_single_threaded; + defer threaded.deinit(); + var env = test_env(&threaded); + const io = env.io; + const gpa = testing.allocator; + + var tmp = try TmpLog.init(gpa); + defer tmp.deinit(gpa); + var engine = try Engine.open(gpa, io, tmp.path); + defer engine.deinit(); + engine.compact_threshold = 1; + + // Leave real garbage behind, so a compaction that ran would be visible: + // `compact` resets dead_docs to zero and nothing else does. + try engine.lock(); + var doc = try make_doc(gpa, 1, "alice"); + defer doc.deinit(); + try engine.insert("app", "users", &doc, &env.gen); + var doc2 = try make_doc(gpa, 1, "alice-again"); + defer doc2.deinit(); + // A replace supersedes the first record, leaving it behind as garbage. + try engine.replace("app", "users", &doc2, &env.gen); + try engine.commit(); + engine.unlock(); + try testing.expect(engine.dead_docs > 0); + const dead_before = engine.dead_docs; + + // With a compaction "in flight", compact must return without rewriting. + engine.compacting.store(true, .release); + try engine.compact(); + try testing.expectEqual(dead_before, engine.dead_docs); + + // With the slot free, the same call does the work -- proving the assertion + // above came from the guard and not from there being nothing to do. + engine.compacting.store(false, .release); + try engine.compact(); + try testing.expectEqual(@as(u64, 0), engine.dead_docs); +} + +test "concurrent writers compacting: the log survives a reopen" { + // Real worker threads driving compaction while other writers append, each + // following the lock sequence the server's dispatch uses (catalog -> + // collection -> release both -> commit -> compact). Every other compaction + // test is single-threaded, so this is the only coverage of the whole write + // path under genuine contention. + // + // What it proves: concurrent compaction leaves a log that replays to + // exactly the right documents. The count is checked exactly -- too few + // means a rewrite was published half-written, too many means a stale tmp + // tail was replayed as live data. + // + // What it does not prove: that either specific race is fixed. Both windows + // are too narrow to hit reliably (see the test above), and this test passes + // with the `compacting` guard removed. It is a smoke test for the path, not + // a regression test for the guard; the deterministic tests above and in + // storage.zig are what pin those two invariants. + const gpa = testing.allocator; + var threaded: std.Io.Threaded = .init(gpa, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var tmp = try TmpLog.init(gpa); + defer tmp.deinit(gpa); + + const writers = 4; + const per_writer: i32 = 60; + const total: i32 = writers * per_writer; + + { + var engine = try Engine.open(gpa, io, tmp.path); + defer engine.deinit(); + // Every write wants a compaction, so writers pile into compact() with + // maximum overlap -- the point of the test. + engine.compact_threshold = 1; + + var next_id = std.atomic.Value(i32).init(1); + + const Worker = struct { + /// The in-lock half of a write command: the collection lock is + /// taken under the catalog lock, and both are released on return -- + /// so the caller's commit runs holding neither, exactly as the + /// server's dispatch epilogue does. + fn insert_locked(e: *Engine, doc: *bson.Document) !void { + try e.lock_catalog(false); + defer e.unlock_catalog(false); + const coll = try e.lock_collection("app", "users", true, true); + if (coll) |c| { + defer e.unlock_collection(c, true); + try e.insert("app", "users", doc, undefined); + } + } + + fn writer(e: *Engine, id_counter: *std.atomic.Value(i32), alloc: std.mem.Allocator) error{Canceled}!void { + while (true) { + const id = id_counter.fetchAdd(1, .monotonic); + if (id > total) return; + var doc = make_doc(alloc, id, "user") catch return error.Canceled; + defer doc.deinit(); + // Ids come from the shared counter, so no insert here can + // legitimately fail; any error is a real defect. + insert_locked(e, &doc) catch return error.Canceled; + e.commit() catch return error.Canceled; + if (e.take_compact()) e.compact() catch return error.Canceled; + } + } + }; + + var group: std.Io.Group = .init; + defer group.cancel(io); + for (0..writers) |_| group.async(io, Worker.writer, .{ &engine, &next_id, gpa }); + try group.await(io); + } + + // Reopen from disk: this replays the log that compaction left behind, which + // is the only place the races above are observable. + var reopened = try Engine.open(gpa, io, tmp.path); + defer reopened.deinit(); + try reopened.lock_read(); + defer reopened.unlock_read(); + const coll = reopened.get_collection("app", "users") orelse return error.TestUnexpectedResult; + try testing.expectEqual(@as(usize, @intCast(total)), coll.docs.count()); + for (1..total + 1) |i| { + const id_key = try bson.serialize_value(gpa, bson.Value{ .int32 = @intCast(i) }); + defer gpa.free(id_key); + try testing.expect(reopened.get_doc("app", "users", id_key) != null); + } +} + // -- index tests ----------------------------------------------------------- /// A spec document for a single-path index, built by serializing and diff --git a/src/lib.zig b/src/lib.zig index 4f23dee..f1f79cd 100644 --- a/src/lib.zig +++ b/src/lib.zig @@ -1,5 +1,6 @@ // mongo-lite core library. Public entry point for tests and the server. +pub const assert = @import("assert.zig"); pub const bson = @import("bson.zig"); pub const wire = @import("wire.zig"); pub const commands = @import("commands.zig"); @@ -11,6 +12,7 @@ pub const update = @import("update.zig"); pub const index = @import("index.zig"); test { + _ = @import("assert.zig"); _ = @import("bson.zig"); _ = @import("wire.zig"); _ = @import("commands.zig"); diff --git a/src/storage.zig b/src/storage.zig index 04865f9..aa4b011 100644 --- a/src/storage.zig +++ b/src/storage.zig @@ -116,13 +116,34 @@ pub const Log = struct { compressed: std.ArrayListUnmanaged(u8), /// LZ4 hash table (positions of recent 4-byte sequences). lz4_table: []u32, - /// Group commit: while set, appends skip the per-block fsync and the - /// caller issues one sync for the whole batch (see Engine.begin_batch / - /// end_batch). Every acknowledged write is still fsynced before the - /// reply, so the crash guarantees are unchanged. - defer_sync: bool = false, + /// Whether an existing file's contents are kept or discarded. + /// + /// `.keep` is the database's own log: its bytes are the database, and + /// `replay` reads them. `.truncate` is for a file being written from + /// scratch (compaction's tmp), where leftover bytes from an earlier, + /// longer file would survive past the new content as intact blocks and be + /// replayed as live records. + const OpenMode = enum { keep, truncate }; + + /// Open the log at `path`, keeping whatever is already there for `replay`. pub fn open(gpa: std.mem.Allocator, io: std.Io, path: []const u8) !Log { + return open_mode(gpa, io, path, .keep); + } + + /// Open `path` as a brand-new empty log, discarding anything already there. + /// + /// Compaction's tmp file must start empty. `open` keeps an existing file's + /// bytes and only rewinds end_pos to the header, so a longer previous tmp + /// (a retried or crashed compaction) would leave valid, hash-correct + /// blocks past the new content -- which `replay` applies as live records + /// once the rename publishes the file as the database, resurrecting + /// documents that were deleted. + pub fn create(gpa: std.mem.Allocator, io: std.Io, path: []const u8) !Log { + return open_mode(gpa, io, path, .truncate); + } + + fn open_mode(gpa: std.mem.Allocator, io: std.Io, path: []const u8, mode: OpenMode) !Log { // Resolve to an absolute path so compaction can rename the file // without depending on the caller's working directory. const abs_path = blk: { @@ -134,9 +155,14 @@ pub const Log = struct { errdefer gpa.free(abs_path); const dir = std.Io.Dir.cwd(); - const file: std.Io.File = dir.openFile(io, abs_path, .{ .mode = .read_write }) catch |err| switch (err) { - error.FileNotFound => try dir.createFile(io, abs_path, .{ .read = true }), - else => return err, + const file: std.Io.File = switch (mode) { + // createFile truncates by default, so this both creates a missing + // file and empties an existing one -- the whole point of .truncate. + .truncate => try dir.createFile(io, abs_path, .{ .read = true, .truncate = true }), + .keep => dir.openFile(io, abs_path, .{ .mode = .read_write }) catch |err| switch (err) { + error.FileNotFound => try dir.createFile(io, abs_path, .{ .read = true }), + else => return err, + }, }; var self: Log = .{ @@ -151,7 +177,6 @@ pub const Log = struct { .block = .empty, .compressed = .empty, .lz4_table = undefined, - .defer_sync = false, }; errdefer { self.scratch.deinit(gpa); @@ -376,9 +401,11 @@ pub const Log = struct { self.log_bytes += total; } - /// One fsync for the whole deferred batch, sealing the current block - /// first. Callers must have set defer_sync, appended, and cleared - /// defer_sync again before the reply. + /// The log's only commit point: seal the open block, then one fsync. + /// + /// Appends never sync (see `append_record`), so nothing is durable until + /// this returns -- which is why Engine.commit calls it exactly once per + /// write command, coalescing every writer in flight into a single fsync. pub fn sync(self: *Log) !void { try self.seal_block(); try self.file.sync(self.io); @@ -817,3 +844,74 @@ test "torn tail truncates cleanly and appends overwrite it" { try log3.replay(@ptrCast(&ctx), Ctx.apply); try testing.expectEqualSlices(u8, &[_]u8{ 1, 3 }, seen.items); } + +test "Log.create discards a leftover file; Log.open keeps it" { + // Compaction reuses one tmp path, so a crashed or retried rewrite can leave + // a *longer* file there. `open` only rewinds end_pos to the header, so the + // predecessor's trailing blocks would survive past the new content -- and + // they are intact and hash-correct, so replay applies them as live records + // once the rename publishes the file. `create` is what prevents that. + var threaded: std.Io.Threaded = .init_single_threaded; + defer threaded.deinit(); + const io = threaded.io(); + const gpa = testing.allocator; + + var tmp = try TmpLog.init(gpa); + defer tmp.deinit(gpa); + const path = tmp.path; + const doc_bytes = [_]u8{ 0x0E, 0, 0, 0, 0x10, '_', 'i', 'd', 0, 42, 0, 0, 0, 0 }; + + // Stand in for the abandoned rewrite: three records, synced, then closed. + { + var old = try Log.open(gpa, io, path); + defer old.close(); + try old.append_upsert("db", "c", &doc_bytes, 1); + try old.append_upsert("db", "c", &doc_bytes, 2); + try old.append_upsert("db", "c", &doc_bytes, 3); + try old.sync(); + try testing.expect(try old.file.length(io) > file_header_len); + } + + const Ctx = struct { + seen: *std.ArrayListUnmanaged(u8), + gpa: std.mem.Allocator, + fn apply(ctx: *anyopaque, record: Record, doc: *bson.Document) anyerror!void { + const self: *@This() = @ptrCast(@alignCast(ctx)); + try self.seen.append(self.gpa, @intCast(record.seq)); + doc.deinit(); + self.gpa.destroy(doc); + } + }; + var seen: std.ArrayListUnmanaged(u8) = .empty; + defer seen.deinit(gpa); + var ctx = Ctx{ .seen = &seen, .gpa = gpa }; + + // `open` keeps the leftover bytes: this is the hazard being guarded against. + { + var kept = try Log.open(gpa, io, path); + defer kept.close(); + try testing.expect(try kept.file.length(io) > file_header_len); + try kept.replay(@ptrCast(&ctx), Ctx.apply); + try testing.expectEqualSlices(u8, &[_]u8{ 1, 2, 3 }, seen.items); + } + + // `create` leaves an empty log: nothing past the header, on disk or in the + // append position, so no stale record can be replayed. + { + var fresh = try Log.create(gpa, io, path); + defer fresh.close(); + try testing.expectEqual(@as(u64, file_header_len), try fresh.file.length(io)); + try testing.expectEqual(@as(u64, file_header_len), fresh.end_pos); + + // One short record where three used to be: replay must see only it, + // proving the old tail is gone rather than merely skipped. + try fresh.append_upsert("db", "c", &doc_bytes, 9); + try fresh.sync(); + try testing.expectEqual(try fresh.file.length(io), fresh.end_pos); + } + seen.clearRetainingCapacity(); + var reopened = try Log.open(gpa, io, path); + defer reopened.close(); + try reopened.replay(@ptrCast(&ctx), Ctx.apply); + try testing.expectEqualSlices(u8, &[_]u8{9}, seen.items); +}