diff --git a/src/db.zig b/src/db.zig index 8454025..5968bde 100644 --- a/src/db.zig +++ b/src/db.zig @@ -231,6 +231,16 @@ pub const Engine = struct { /// rewrite is worth doing — see `note_compact`. live_docs: u64 = 0, dead_docs: u64 = 0, + /// Whether replay must maintain index entries as it goes. + /// + /// A full replay does not: it puts documents in place and lets + /// `build_all_indexes` bulk-pack every index afterwards, which is O(n log n) + /// once instead of per record. After a checkpoint that is wrong -- the + /// indexes arrive already populated, `rebuild_index` skips a non-empty one by + /// design, and the records replayed on top would be invisible to every + /// index. The symptom was a document present in the collection and missing + /// from `_id_`, which after the hashmap goes away means simply missing. + replay_maintains_indexes: bool = false, /// Set to the failing index's own stable name when an upsert is /// rejected by a unique secondary index (error.DuplicateKeyIndex). The /// command reads it while still holding the write lock. @@ -240,16 +250,10 @@ pub const Engine = struct { var log = try storage.Log.open(gpa, io, path); errdefer log.close(); - // The data file sits beside the log. It is recreated empty on every - // open for now: the log is still replayed in full, so nothing durable - // depends on the file yet, and open/close semantics stay exactly what - // they were. The watermark that makes it a checkpoint comes later. + // The data file sits beside the log and is *kept*: a valid watermark in + // it means most of the log never has to be replayed. const data_path = try std.fmt.allocPrint(gpa, "{s}.data", .{log.path}); defer gpa.free(data_path); - std.Io.Dir.cwd().deleteFile(io, data_path) catch |err| switch (err) { - error.FileNotFound => {}, - else => return err, - }; const pager_box = try gpa.create(pgr.Pager); errdefer gpa.destroy(pager_box); @@ -270,10 +274,40 @@ pub const Engine = struct { engine.dbs.deinit(gpa); } - try engine.log.replay(&engine, apply_record); + // A checkpoint, if the data file has one, decides where replay starts. + // Nothing below the watermark needs re-applying: the data file already + // holds its effect. + var replay_from: u64 = 0; + if (engine.pager.loaded.generation != 0) { + engine.read_catalog() catch |err| { + // The image is unusable but the log is not. Warn, drop + // everything loaded, and fall back to a full replay -- the + // database must always open. + std.debug.print( + "multiforadb: WARNING: data file catalog unreadable ({s}); " ++ + "replaying the log in full\n", + .{@errorName(err)}, + ); + engine.reset_after_failed_catalog(); + replay_from = 0; + }; + if (replay_from == 0 and engine.dbs.count() > 0) { + replay_from = engine.pager.loaded.seq; + engine.replay_maintains_indexes = true; + engine.seq = replay_from; + engine.committed_seq = replay_from; + engine.live_docs = engine.pager.loaded.live_docs; + } + } + + try engine.log.replay(&engine, apply_record, replay_from); // Replay registers empty indexes; build them from the live docs - // once replay completes (order-independent). + // once replay completes (order-independent). A checkpointed open finds + // them already populated, and the guard in rebuild_index skips them. try engine.build_all_indexes(); + // Everything replayed is durable by definition -- it was read back off + // the log -- so the commit watermark starts level with the sequence. + engine.committed_seq = engine.seq; return engine; } @@ -1257,6 +1291,318 @@ pub const Engine = struct { } } + // -- checkpoint --------------------------------------------------------- + + /// The catalog: everything about where the engine's structures live that is + /// not recoverable by looking at the pages themselves. + /// + /// Written wholesale into freshly allocated pages at every checkpoint, never + /// mutated in place, so the previous copy stays intact and referenced by the + /// previous watermark until the new one switches over. That is what makes it + /// untearable, and it is why there is no incremental catalog update path. + /// + /// Format (little-endian throughout): + /// u32 magic "MFCT", u32 version, u64 live_docs, u32 db_count + /// per db: u32 name_len, name, u32 coll_count + /// per coll: u32 name_len, name, u64 slab_tail, u64 slab_end, + /// u32 extent_count, (u32 first, u32 pages)*, u32 index_count + /// index 0 is always the implicit _id_ + /// per index: u32 name_len, name, u32 key_count, + /// (u32 path_len, path, u8 descending)*, + /// u8 flags(unique|sparse|multikey|has_ttl), i64 ttl, + /// u32 root, u32 first_leaf, u32 leaf_count, u32 depth, + /// u64 entry_count, u64 ovf_tail, u64 ovf_end, + /// u32 ovf_extent_count, (u32 first, u32 pages)*, + /// u32 node_count, (u32 page)* + /// u64 xxhash3 over everything above + const catalog_magic: u32 = 0x4D464354; // "MFCT" + const catalog_version: u32 = 1; + + fn write_catalog(self: *Engine, out: *std.ArrayListUnmanaged(u8)) !void { + const gpa = self.gpa; + try put_u32(gpa, out, catalog_magic); + try put_u32(gpa, out, catalog_version); + try put_u64(gpa, out, self.live_docs); + try put_u32(gpa, out, @intCast(self.dbs.count())); + var db_it = self.dbs.iterator(); + while (db_it.next()) |db_entry| { + try put_bytes(gpa, out, db_entry.key_ptr.*); + const colls = &db_entry.value_ptr.collections; + try put_u32(gpa, out, @intCast(colls.count())); + var coll_it = colls.iterator(); + while (coll_it.next()) |ce| { + const coll = ce.value_ptr.*; + try put_bytes(gpa, out, ce.key_ptr.*); + try put_u64(gpa, out, coll.slab_tail); + try put_u64(gpa, out, coll.slab_end); + try put_u32(gpa, out, @intCast(coll.slab_extents.items.len)); + for (coll.slab_extents.items) |e| { + try put_u32(gpa, out, e.first); + try put_u32(gpa, out, e.pages); + } + try put_u32(gpa, out, @intCast(coll.indexes.items.len + 1)); + try write_index_catalog(gpa, out, &coll.id_index); + for (coll.indexes.items) |ix| try write_index_catalog(gpa, out, ix); + } + } + try put_u64(gpa, out, std.hash.XxHash3.hash(0, out.items)); + } + + fn write_index_catalog( + gpa: std.mem.Allocator, + out: *std.ArrayListUnmanaged(u8), + ix: *const index.Index, + ) !void { + try put_bytes(gpa, out, ix.name); + try put_u32(gpa, out, @intCast(ix.keys.len)); + for (ix.keys) |k| { + try put_bytes(gpa, out, k.path); + try out.append(gpa, @intFromBool(k.descending)); + } + var flags: u8 = 0; + if (ix.unique) flags |= 1; + if (ix.sparse) flags |= 2; + if (ix.multikey) flags |= 4; + if (ix.ttl != null) flags |= 8; + try out.append(gpa, flags); + try put_u64(gpa, out, @bitCast(ix.ttl orelse 0)); + try put_u32(gpa, out, ix.root); + try put_u32(gpa, out, ix.first_leaf); + try put_u32(gpa, out, ix.leaf_count); + try put_u32(gpa, out, ix.depth); + try put_u64(gpa, out, ix.entry_count); + try put_u64(gpa, out, ix.ovf_tail); + try put_u64(gpa, out, ix.ovf_end); + try put_u32(gpa, out, @intCast(ix.ovf_extents.items.len)); + for (ix.ovf_extents.items) |e| { + try put_u32(gpa, out, e.first); + try put_u32(gpa, out, e.pages); + } + try put_u32(gpa, out, @intCast(ix.node_pages.items.len)); + for (ix.node_pages.items) |pg| try put_u32(gpa, out, pg); + } + + /// Rebuild the catalog from the data file. On any inconsistency this returns + /// an error and the caller falls back to a full replay. + fn read_catalog(self: *Engine) !void { + const wm = self.pager.loaded; + if (wm.catalog_len == 0) return error.NoCatalog; + const buf = self.pager.bytes(@as(u64, wm.catalog_page) << pgr.page_shift, @intCast(wm.catalog_len)); + if (buf.len < 8) return error.CorruptCatalog; + const body = buf[0 .. buf.len - 8]; + if (std.hash.XxHash3.hash(0, body) != std.mem.readInt(u64, buf[buf.len - 8 ..][0..8], .little)) { + return error.CorruptCatalog; + } + var r: Reader = .{ .b = body }; + if (try r.read_u32() != catalog_magic) return error.CorruptCatalog; + if (try r.read_u32() != catalog_version) return error.CorruptCatalog; + self.live_docs = try r.read_u64(); + const ndbs = try r.read_u32(); + var d: u32 = 0; + while (d < ndbs) : (d += 1) { + const db_name = try r.read_bytes(); + const ncolls = try r.read_u32(); + var c: u32 = 0; + while (c < ncolls) : (c += 1) { + const coll_name = try r.read_bytes(); + const coll = try self.get_or_create_collection(db_name, coll_name); + coll.slab_tail = try r.read_u64(); + coll.slab_end = try r.read_u64(); + const nex = try r.read_u32(); + var e: u32 = 0; + while (e < nex) : (e += 1) { + const first = try r.read_u32(); + const pages = try r.read_u32(); + try coll.slab_extents.append(self.gpa, .{ .first = first, .pages = pages }); + } + const nix = try r.read_u32(); + // Index 0 is the implicit _id_, already created by + // get_or_create_collection; the rest are registered here. + try read_index_catalog(self.gpa, &r, &coll.id_index); + var i: u32 = 1; + while (i < nix) : (i += 1) { + const boxed = try self.gpa.create(index.Index); + errdefer self.gpa.destroy(boxed); + boxed.* = try index.Index.init(self.gpa, self.pager, "", &.{}, false, false, null); + read_index_catalog(self.gpa, &r, boxed) catch |err| { + boxed.deinit(self.gpa); + self.gpa.destroy(boxed); + return err; + }; + try coll.indexes.append(self.gpa, boxed); + } + // The docs map is still the authoritative _id -> offset lookup, + // and it is not in the catalog on purpose: the commit that drops + // it would only have to delete the format again. Rebuilt from the + // _id_ tree instead, which the data file already holds. + try self.rebuild_docs_map(coll); + } + } + } + + /// Replace an index's identity and tree position from the catalog. The index + /// arrives freshly initialised, so its own two starting pages are discarded + /// in favour of what was published. + fn read_index_catalog( + gpa: std.mem.Allocator, + r: *Reader, + ix: *index.Index, + ) !void { + const name = try r.read_bytes(); + const nkeys = try r.read_u32(); + if (nkeys == 0 or nkeys > index.max_index_keys) return error.CorruptCatalog; + var keys = try gpa.alloc(index.IndexKey, nkeys); + var built: usize = 0; + errdefer { + for (keys[0..built]) |k| gpa.free(k.path); + gpa.free(keys); + } + while (built < nkeys) : (built += 1) { + const path = try r.read_bytes(); + keys[built] = .{ .path = try gpa.dupe(u8, path), .descending = (try r.read_byte()) != 0 }; + } + const flags = try r.read_byte(); + const ttl_raw: i64 = @bitCast(try r.read_u64()); + + const new_name = try gpa.dupe(u8, name); + errdefer gpa.free(new_name); + + // Swap in the published identity, freeing what init made. + for (ix.keys) |k| gpa.free(k.path); + gpa.free(ix.keys); + gpa.free(ix.name); + ix.name = new_name; + ix.keys = keys; + ix.unique = flags & 1 != 0; + ix.sparse = flags & 2 != 0; + ix.multikey = flags & 4 != 0; + ix.ttl = if (flags & 8 != 0) ttl_raw else null; + ix.root = try r.read_u32(); + ix.first_leaf = try r.read_u32(); + ix.leaf_count = try r.read_u32(); + ix.depth = try r.read_u32(); + ix.entry_count = @intCast(try r.read_u64()); + ix.ovf_tail = try r.read_u64(); + ix.ovf_end = try r.read_u64(); + const novf = try r.read_u32(); + var o: u32 = 0; + while (o < novf) : (o += 1) { + const first = try r.read_u32(); + const pages = try r.read_u32(); + try ix.ovf_extents.append(gpa, .{ .first = first, .pages = pages }); + } + const nnodes = try r.read_u32(); + if (nnodes < 2) return error.CorruptCatalog; + ix.node_pages.clearRetainingCapacity(); + try ix.node_pages.ensureTotalCapacity(gpa, nnodes); + var n: u32 = 0; + while (n < nnodes) : (n += 1) ix.node_pages.appendAssumeCapacity(try r.read_u32()); + } + + /// Add a replayed document's entries to every index that is already + /// populated. Best effort and infallible: replay must not refuse to start, + /// and an index that cannot key this document is reported and left alone -- + /// the same tolerance `rebuild_index` has always had. + fn index_doc_on_replay(self: *Engine, coll: *Collection, doc_bytes: []const u8, off: u64) void { + self.index_one(&coll.id_index, doc_bytes, off); + for (coll.indexes.items) |ix| self.index_one(ix, doc_bytes, off); + } + + fn index_one(self: *Engine, ix: *index.Index, doc_bytes: []const u8, off: u64) void { + var built = ix.build_entries(self.gpa, doc_bytes) catch return; + defer built.deinit(self.gpa); + if (built.multikey) ix.multikey = true; + ix.reserve_for(self.gpa, built.entries.items) catch return; + ix.insert_entries(&built, off); + } + + /// Rebuild the _id -> offset hashmap by walking the _id_ tree. + fn rebuild_docs_map(self: *Engine, coll: *Collection) !void { + var arena = std.heap.ArenaAllocator.init(self.gpa); + defer arena.deinit(); + var it = coll.id_index.iter(); + while (it.next()) |e| { + _ = arena.reset(.retain_capacity); + const id_value = (try bson.get_at(arena.allocator(), coll.doc_bytes(e.off), "_id")) orelse continue; + const id_key = try bson.serialize_value(self.gpa, id_value); + errdefer self.gpa.free(id_key); + try coll.docs.put(self.gpa, id_key, e.off); + } + } + + /// Discard everything a failed catalog load put in place, so the caller can + /// replay the log into a clean engine. + fn reset_after_failed_catalog(self: *Engine) void { + var db_it = self.dbs.iterator(); + while (db_it.next()) |db_entry| { + self.free_db(db_entry.value_ptr); + self.gpa.free(db_entry.key_ptr.*); + } + self.dbs.clearRetainingCapacity(); + self.live_docs = 0; + self.dead_docs = 0; + self.seq = 0; + self.committed_seq = 0; + } + + /// Publish the current state as a checkpoint. + /// + /// The watermark equals the sequence the log has already made durable, never + /// more: `commit` first, then snapshot, and the snapshot is validated against + /// an unchanged `seq` under the log lock -- the same bounded-retry shape + /// compaction has always used. That is the crash-recovery invariant (PLAN + /// D6) reduced to an ordering. + pub fn checkpoint(self: *Engine) !void { + try self.commit(); + + var buf: std.ArrayListUnmanaged(u8) = .empty; + defer buf.deinit(self.gpa); + + var attempt: usize = 0; + const attempt_max = 8; + while (attempt < attempt_max) : (attempt += 1) { + buf.clearRetainingCapacity(); + try self.catalog_lock.lockShared(self.io); + const snapshot_seq = self.seq; + self.write_catalog(&buf) catch |err| { + self.catalog_lock.unlockShared(self.io); + return err; + }; + self.catalog_lock.unlockShared(self.io); + + try self.log_lock.lock(self.io); + if (self.seq != snapshot_seq) { + // A writer landed mid-snapshot; the catalog describes a state + // that no longer matches the log. Retry rather than publish it. + self.log_lock.unlock(self.io); + continue; + } + assert_msg( + snapshot_seq <= self.committed_seq, + "checkpoint watermark past the durable log tail", + ); + const pages: u32 = @intCast((buf.items.len + pgr.page_size - 1) / pgr.page_size); + const first = self.pager.alloc_pages(pages) catch |err| { + self.log_lock.unlock(self.io); + return err; + }; + @memcpy(self.pager.bytes_mut(@as(u64, first) << pgr.page_shift, buf.items.len), buf.items); + self.pager.publish(.{ + .seq = snapshot_seq, + .catalog_page = first, + .catalog_len = buf.items.len, + .live_docs = self.live_docs, + }) catch |err| { + self.log_lock.unlock(self.io); + return err; + }; + self.pager.release_reservation(); + self.log_lock.unlock(self.io); + return; + } + std.debug.print("multiforadb: WARNING: checkpoint gave up after {d} attempts under sustained writes\n", .{attempt_max}); + } + /// Register an (empty) index from a persisted spec document. A repeated /// create record for the same name is an idempotent no-op. fn register_index_from_spec( @@ -1339,6 +1685,12 @@ fn apply_record(ctx: *anyopaque, record: storage.Record, doc: *bson.Document) an var key_owned = false; defer if (!key_owned) self.gpa.free(id_key); + // Engine.seq used to restart at 0 on every open, which was harmless while + // the log was always replayed in full and fatal the moment a watermark + // exists: the first append after an open would reuse a sequence at or below + // it, and the *next* open would discard that record as already-checkpointed. + self.seq = @max(self.seq, record.seq); + switch (record.type) { storage.record_type_upsert => { self.evict_doc(coll, id_key); @@ -1346,10 +1698,13 @@ fn apply_record(ctx: *anyopaque, record: storage.Record, doc: *bson.Document) an defer self.gpa.free(doc_bytes); try coll.slab_reserve(self.gpa, doc_bytes.len); const off = coll.slab_append(doc_bytes); - self.pager.release_reservation(); try coll.docs.put(self.gpa, id_key, off); self.live_docs += 1; key_owned = true; + if (self.replay_maintains_indexes) { + self.index_doc_on_replay(coll, doc_bytes, off); + } + self.pager.release_reservation(); // The _id_ entry is added after replay, in build_all_indexes, // together with the secondary indexes. // @@ -2056,6 +2411,103 @@ test "index survives reopen and compaction" { engine2.unlock(); } +test "a checkpoint lets the next open skip the log it covers" { + // The point of the whole milestone: an open that finds a watermark loads the + // data file and replays only what happened after it, instead of rebuilding + // everything from the log. + // + // Mutation checks, red: publishing a watermark seq of 0; and removing the + // index maintenance in apply_record, which leaves a replayed document + // present in the collection and absent from `_id_` -- which, once the + // hashmap goes, means simply absent. + // + // Not covered, and worth stating rather than implying: removing + // `self.seq = @max(self.seq, record.seq)` from apply_record leaves this + // green. The sequence is seeded from the watermark on a checkpointed open, + // so it only drifts by the records replayed on top -- and every sequence + // reachable from here has the catalog carrying those same records, which + // masks the drift. Observing it needs a crash between a duplicate-sequence + // append and the checkpoint that would have captured it, which is the + // crash-injection harness's job, not this test's. The line stays because a + // log whose sequences are not monotonic has no total order, and + // `committed_seq <= seq` is asserted on every commit. + var threaded: std.Io.Threaded = .init_single_threaded; + defer threaded.deinit(); + var env = test_env(&threaded); + const io = env.io; + const gpa = testing.allocator; + + var tmp = try TmpLog.init(gpa); + defer tmp.deinit(gpa); + const data_path = try std.fmt.allocPrint(gpa, "{s}.data", .{tmp.path}); + defer gpa.free(data_path); + defer std.Io.Dir.cwd().deleteFile(io, data_path) catch {}; + + { + var engine = try Engine.open(gpa, io, tmp.path); + defer engine.deinit(); + try engine.lock(); + var i: i32 = 0; + while (i < 40) : (i += 1) { + var d = try make_user(gpa, i, "a@x.io"); + defer d.deinit(); + try engine.insert("app", "users", &d, &env.gen); + } + engine.unlock(); + try engine.checkpoint(); + try testing.expect(engine.pager.loaded.generation >= 1); + try testing.expectEqual(engine.seq, engine.pager.loaded.seq); + + // Writes after the checkpoint are the ones a reopen must replay. + try engine.lock(); + i = 100; + while (i < 105) : (i += 1) { + var d = try make_user(gpa, i, "b@x.io"); + defer d.deinit(); + try engine.insert("app", "users", &d, &env.gen); + } + engine.unlock(); + } + + // Reopen: the checkpoint is loaded, so only the five later records apply. + { + var engine = try Engine.open(gpa, io, tmp.path); + defer engine.deinit(); + try engine.lock(); + defer engine.unlock(); + try testing.expect(engine.pager.loaded.generation >= 1); + const coll = engine.get_collection("app", "users").?; + try testing.expectEqual(@as(usize, 45), coll.docs.count()); + try testing.expectEqual(@as(usize, 45), coll.id_index.count()); + // And the sequence continued from the watermark rather than restarting. + try testing.expect(engine.seq >= engine.pager.loaded.seq); + } + + // A second reopen, to catch a sequence that restarted: the writes made after + // the first reopen must survive it. + { + var engine = try Engine.open(gpa, io, tmp.path); + defer engine.deinit(); + try engine.lock(); + var d = try make_user(gpa, 500, "c@x.io"); + defer d.deinit(); + try engine.insert("app", "users", &d, &env.gen); + engine.unlock(); + try engine.checkpoint(); + } + { + var engine = try Engine.open(gpa, io, tmp.path); + defer engine.deinit(); + try engine.lock(); + defer engine.unlock(); + const coll = engine.get_collection("app", "users").?; + try testing.expectEqual(@as(usize, 46), coll.docs.count()); + const id_key = try bson.serialize_value(gpa, .{ .int32 = 500 }); + defer gpa.free(id_key); + try testing.expect(engine.get_doc("app", "users", id_key) != null); + } +} + test "index drop survives reopen" { var threaded: std.Io.Threaded = .init_single_threaded; defer threaded.deinit(); @@ -2324,3 +2776,55 @@ test "ttl_sweep spans collections and several TTL indexes on one collection" { try testing.expectEqual(@as(usize, 0), engine.get_collection("app", "events").?.docs.count()); try testing.expectEqual(@as(usize, 1), engine.get_collection("other", "plain").?.docs.count()); } + +// --------------------------------------------------------------------------- +// Catalog encoding helpers +// --------------------------------------------------------------------------- + +fn put_u32(gpa: std.mem.Allocator, out: *std.ArrayListUnmanaged(u8), v: u32) !void { + var b: [4]u8 = undefined; + std.mem.writeInt(u32, &b, v, .little); + try out.appendSlice(gpa, &b); +} + +fn put_u64(gpa: std.mem.Allocator, out: *std.ArrayListUnmanaged(u8), v: u64) !void { + var b: [8]u8 = undefined; + std.mem.writeInt(u64, &b, v, .little); + try out.appendSlice(gpa, &b); +} + +fn put_bytes(gpa: std.mem.Allocator, out: *std.ArrayListUnmanaged(u8), v: []const u8) !void { + try put_u32(gpa, out, @intCast(v.len)); + try out.appendSlice(gpa, v); +} + +/// A bounds-checked cursor over the catalog. Every read is checked because the +/// bytes come off disk: a truncated or scrambled catalog must produce an error +/// the caller can fall back from, never a read past the end. +const Reader = struct { + b: []const u8, + at: usize = 0, + + fn take(self: *Reader, n: usize) ![]const u8 { + if (self.at + n > self.b.len) return error.CorruptCatalog; + defer self.at += n; + return self.b[self.at..][0..n]; + } + + fn read_byte(self: *Reader) !u8 { + return (try self.take(1))[0]; + } + + fn read_u32(self: *Reader) !u32 { + return std.mem.readInt(u32, (try self.take(4))[0..4], .little); + } + + fn read_u64(self: *Reader) !u64 { + return std.mem.readInt(u64, (try self.take(8))[0..8], .little); + } + + fn read_bytes(self: *Reader) ![]const u8 { + const n = try self.read_u32(); + return self.take(n); + } +}; diff --git a/src/storage.zig b/src/storage.zig index e3330dd..9568ead 100644 --- a/src/storage.zig +++ b/src/storage.zig @@ -219,7 +219,14 @@ pub const Log = struct { } /// Replay all valid records from the beginning of the file. - pub fn replay(self: *Log, ctx: *anyopaque, callback: ReplayFn) !void { + /// Replay every record with `seq > from_seq`. Zero replays everything, which + /// is what an engine with no checkpoint does. + /// + /// The walk still visits every block regardless, because that is what leaves + /// `end_pos` correct for the next append. Skipping is about not *applying* + /// records the data file already contains; making opens fast is the job of + /// truncating the log once a checkpoint covers it. + pub fn replay(self: *Log, ctx: *anyopaque, callback: ReplayFn, from_seq: u64) !void { var decomp: std.ArrayListUnmanaged(u8) = .empty; defer decomp.deinit(self.gpa); var pos: u64 = file_header_len; @@ -275,7 +282,7 @@ pub const Log = struct { var idx: usize = 0; while (idx < decomp.items.len) { - idx += try self.parse_record(decomp.items[idx..], pos + idx, ctx, callback); + idx += try self.parse_record(decomp.items[idx..], pos + idx, ctx, callback, from_seq); } pos += total; self.end_pos = pos; @@ -292,6 +299,7 @@ pub const Log = struct { pos: u64, ctx: *anyopaque, callback: ReplayFn, + from_seq: u64, ) !usize { if (bytes.len < 4) return error.InvalidLog; const total: u32 = std.mem.readInt(u32, bytes[0..4], .little); @@ -312,6 +320,9 @@ pub const Log = struct { var idx: usize = header_len - 4; const seq: u64 = std.mem.readInt(u64, payload[8..16], .little); + // Below the checkpoint: the data file already holds its effect. Skipped + // before the BSON parse, which is the expensive part. + if (seq <= from_seq) return total; const rtype = payload[16]; const db = read_cstring(payload, &idx) orelse return error.InvalidLog; const coll = read_cstring(payload, &idx) orelse return error.InvalidLog; @@ -739,7 +750,7 @@ test "append, replay, torn tail" { } }; var ctx = Ctx{ .seen = &seen, .gpa = gpa }; - try log.replay(@ptrCast(&ctx), Ctx.apply); + try log.replay(@ptrCast(&ctx), Ctx.apply, 0); try testing.expectEqualSlices(u8, &[_]u8{ record_type_upsert, 1, record_type_delete, 2 }, seen.items); } @@ -783,7 +794,7 @@ test "record larger than the read chunk replays" { } }; var ctx = Ctx{ .count = &count, .gpa = gpa }; - try log.replay(@ptrCast(&ctx), Ctx.apply); + try log.replay(@ptrCast(&ctx), Ctx.apply, 0); try testing.expectEqual(@as(usize, 1), count); } @@ -832,7 +843,7 @@ test "reject corrupt interior block" { } }; var ctx = Ctx{ .count = &count, .gpa = gpa }; - try testing.expectError(error.InvalidLog, log2.replay(@ptrCast(&ctx), Ctx.apply)); + try testing.expectError(error.InvalidLog, log2.replay(@ptrCast(&ctx), Ctx.apply, 0)); } test "torn tail truncates cleanly and appends overwrite it" { @@ -878,7 +889,7 @@ test "torn tail truncates cleanly and appends overwrite it" { } }; var ctx = Ctx{ .seen = &seen, .gpa = gpa }; - try log2.replay(@ptrCast(&ctx), Ctx.apply); + try log2.replay(@ptrCast(&ctx), Ctx.apply, 0); try testing.expectEqualSlices(u8, &[_]u8{1}, seen.items); // A new append overwrites from the replay end and replays cleanly. @@ -888,7 +899,7 @@ test "torn tail truncates cleanly and appends overwrite it" { var log3 = try Log.open(gpa, io, path); defer log3.close(); ctx.seen = &seen; - try log3.replay(@ptrCast(&ctx), Ctx.apply); + try log3.replay(@ptrCast(&ctx), Ctx.apply, 0); try testing.expectEqualSlices(u8, &[_]u8{ 1, 3 }, seen.items); } @@ -938,7 +949,7 @@ test "Log.create discards a leftover file; Log.open keeps it" { 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 kept.replay(@ptrCast(&ctx), Ctx.apply, 0); try testing.expectEqualSlices(u8, &[_]u8{ 1, 2, 3 }, seen.items); } @@ -959,6 +970,6 @@ test "Log.create discards a leftover file; Log.open keeps it" { seen.clearRetainingCapacity(); var reopened = try Log.open(gpa, io, path); defer reopened.close(); - try reopened.replay(@ptrCast(&ctx), Ctx.apply); + try reopened.replay(@ptrCast(&ctx), Ctx.apply, 0); try testing.expectEqualSlices(u8, &[_]u8{9}, seen.items); }