diff --git a/src/db.zig b/src/db.zig index e07f39f..4e3868c 100644 --- a/src/db.zig +++ b/src/db.zig @@ -800,12 +800,14 @@ pub const RangeIterator = struct { } var winner: ?Record = null; - for (self.cursors) |*c| { - if (c.valid() and order(c.record().key, minKey.?) == .eq) { - const rec = c.record(); - if (rec.seq <= self.maxSeq and (winner == null or rec.seq > winner.?.seq)) winner = rec; - c.advance(); - } + while (self.heap.peek()) |entry| { + if (order(entry.key, minKey) != .eq) break; + const popped = self.heap.pop().?; + var c = &self.cursors[popped.idx]; + const rec = c.record(); + if (rec.seq <= self.maxSeq and (winner == null or rec.seq > winner.?.seq)) winner = rec; + c.advance(); + if (c.valid()) try self.heap.push(.{ .key = c.record().key, .idx = popped.idx }); } if (winner == null) continue; if (!self.startInclusive) { @@ -825,6 +827,7 @@ pub const RangeIterator = struct { pub fn deinit(self: *RangeIterator) void { self.arena.deinit(); + self.heap.deinit(); if (self.start) |bound| self.resultAllocator.free(bound); if (self.end) |bound| self.resultAllocator.free(bound); self.resultAllocator.free(self.cursors); @@ -868,32 +871,29 @@ pub const ReverseRangeIterator = struct { resultAllocator: Allocator, arena: std.heap.ArenaAllocator, cursors: []ReverseSource, + heap: DescKeyHeap, start: ?[]u8, startInclusive: bool, maxSeq: u64, pub fn next(self: *ReverseRangeIterator) !?Entry { while (true) { - var maxKey: ?[]const u8 = null; - for (self.cursors) |*c| { - if (c.valid()) { - const key = c.record().key; - if (maxKey == null or order(key, maxKey.?) == .gt) maxKey = key; - } - } - if (maxKey == null) return null; + const top = self.heap.peek() orelse return null; + const maxKey = top.key; if (self.start) |bound| { - const past = if (self.startInclusive) order(maxKey.?, bound) == .lt else order(maxKey.?, bound) != .gt; + const past = if (self.startInclusive) order(maxKey, bound) == .lt else order(maxKey, bound) != .gt; if (past) return null; } var winner: ?Record = null; - for (self.cursors) |*c| { - if (c.valid() and order(c.record().key, maxKey.?) == .eq) { - const rec = c.record(); - if (rec.seq <= self.maxSeq and (winner == null or rec.seq > winner.?.seq)) winner = rec; - c.advance(); - } + while (self.heap.peek()) |entry| { + if (order(entry.key, maxKey) != .eq) break; + const popped = self.heap.pop().?; + var c = &self.cursors[popped.idx]; + const rec = c.record(); + if (rec.seq <= self.maxSeq and (winner == null or rec.seq > winner.?.seq)) winner = rec; + c.advance(); + if (c.valid()) try self.heap.push(.{ .key = c.record().key, .idx = popped.idx }); } if (winner == null) continue; if (winner.?.kind == .tombstone) continue; @@ -908,6 +908,7 @@ pub const ReverseRangeIterator = struct { pub fn deinit(self: *ReverseRangeIterator) void { self.arena.deinit(); + self.heap.deinit(); for (self.cursors) |*c| c.deinit(); if (self.start) |bound| self.resultAllocator.free(bound); self.resultAllocator.free(self.cursors); @@ -918,36 +919,6 @@ fn rangesOverlap(aMin: []const u8, aMax: []const u8, bMin: []const u8, bMax: []c return order(aMin, bMax) != .gt and order(bMin, aMax) != .gt; } -fn loadManifest(allocator: Allocator, dir: std.fs.Dir) !std.AutoHashMap(u64, usize) { - var map = std.AutoHashMap(u64, usize).init(allocator); - errdefer map.deinit(); - - const file = dir.openFile(manifestName, .{}) catch |err| switch (err) { - error.FileNotFound => return map, - else => return err, - }; - defer file.close(); - const data = try file.readToEndAlloc(allocator, std.math.maxInt(usize)); - defer allocator.free(data); - if (data.len < 8) return map; - - const storedCrc = encode.readU32(data, 0); - const body = data[4..]; - if (encode.crc(body) != storedCrc) return map; - - const count = encode.readU32(body, 0); - var pos: usize = 4; - var i: u32 = 0; - while (i < count) : (i += 1) { - if (pos + 12 > body.len) break; - const fileId = encode.readU64(body, pos); - const level = encode.readU32(body, pos + 8); - try map.put(fileId, level); - pos += 12; - } - return map; -} - fn sstName(buf: []u8, id: u64) ![]const u8 { return std.fmt.bufPrint(buf, "sst-{d:0>10}.cbt", .{id}); } @@ -1310,6 +1281,67 @@ test "leveled compaction keeps levels disjoint and drops bottom tombstones" { try std.testing.expectEqual(@as(u64, 58), totalEntries); } +test "seq and fileId counters survive a compaction that garbage-collects every key" { + // Reproduces a real recovery gap: when a bottom-level compaction merges + // inputs that are entirely tombstones, runMerge produces zero output + // files (see the `builder == null` guard around minting a fresh + // fileId). If every file that ever carried a high seq/fileId gets + // deleted in that round and nothing new gets flushed before a restart, + // a directory-scan-only recovery would compute too low a floor and + // risk reusing already-consumed seq numbers. The manifest's persisted + // counters (VersionEdit.nextFileId/nextSeq, stamped unconditionally in + // runMerge) must close that gap. + const allocator = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + + var beforeNextSeq: u64 = 0; + var beforeNextFileId: u64 = 0; + { + var db = try openTemp(&tmp, allocator, .{ + .l0CompactionTrigger = 2, + .l1BaseBytes = 4 * 1024 * 1024, + }); + defer db.close(); + + var keyBuf: [16]u8 = undefined; + var round: usize = 0; + while (round < 2) : (round += 1) { + const key = try std.fmt.bufPrint(&keyBuf, "gone:{d}", .{round}); + try db.put(key, "value"); + try db.delete(key); + try db.flush(); + } + try std.testing.expectEqual(@as(usize, 2), db.countLevel(0)); + + try db.maybeCompact(); + + // The whole point: every input was a dropped tombstone, so the + // compaction round produced no output file at all, and no new + // fileId was ever minted for it. + try std.testing.expectEqual(@as(usize, 0), db.sstables.items.len); + + beforeNextSeq = db.nextSeq; + beforeNextFileId = db.nextFileId; + try std.testing.expect(beforeNextSeq > 1); + } + + // On disk right now: no sstable files, an empty (reset) WAL. A + // scan-only recovery would compute nextSeq=1/nextFileId=1 here. + var reopened = try openTemp(&tmp, allocator, .{ + .l0CompactionTrigger = 2, + .l1BaseBytes = 4 * 1024 * 1024, + }); + defer reopened.close(); + try std.testing.expectEqual(beforeNextSeq, reopened.nextSeq); + try std.testing.expectEqual(beforeNextFileId, reopened.nextFileId); + + // Functional proof, not just counter equality: a fresh write after + // reopen must not reuse a seq number already spent on the GC'd keys. + try reopened.put("fresh", "value"); + try std.testing.expect(reopened.nextSeq > beforeNextSeq); +} + fn lessByMinKey(_: void, a: *Reader, b: *Reader) bool { return order(a.minKey, b.minKey) == .lt; } @@ -1385,6 +1417,56 @@ test "prefix scan isolates a namespace" { try std.testing.expectEqual(expected.len, idx); } +test "scan merges a wide fan-in of uncompacted sstables with overlapping keys" { + const allocator = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + + var db = try openTemp(&tmp, allocator, .{ .flushThreshold = 1 }); + defer db.close(); + + const sstableCount = 14; + var round: usize = 0; + while (round < sstableCount) : (round += 1) { + var keyBuf: [16]u8 = undefined; + var valBuf: [24]u8 = undefined; + var i: usize = 0; + while (i < 5) : (i += 1) { + const key = try std.fmt.bufPrint(&keyBuf, "shared:{d:0>2}", .{i}); + const value = try std.fmt.bufPrint(&valBuf, "round{d}", .{round}); + try db.put(key, value); + } + try db.flush(); + } + try db.delete("shared:02"); + try db.flush(); + try std.testing.expect(db.sstables.items.len >= sstableCount + 1); + + var forward = try db.scan(allocator, null, null); + defer forward.deinit(); + var seen: usize = 0; + while (try forward.next()) |entry| { + if (std.mem.eql(u8, entry.key, "shared:02")) { + try std.testing.expect(false); + } + try std.testing.expectEqualStrings("round13", entry.value); + seen += 1; + } + try std.testing.expectEqual(@as(usize, 4), seen); + + var reverse = try db.scanReverse(allocator, .{}); + defer reverse.deinit(); + var revSeen: usize = 0; + var lastKey: ?[]const u8 = null; + while (try reverse.next()) |entry| { + if (lastKey) |prev| try std.testing.expect(order(entry.key, prev) == .lt); + lastKey = entry.key; + try std.testing.expectEqualStrings("round13", entry.value); + revSeen += 1; + } + try std.testing.expectEqual(@as(usize, 4), revSeen); +} + test "inclusive and exclusive bounds" { const allocator = std.testing.allocator; var tmp = std.testing.tmpDir(.{});