dualfroz/cobalt
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: extend db

Browse Source
This commit is contained in:
dualfroz committed 2026-09-02 19:42:01 +02:00
1 parent 360df0acea
commit 59073935e4
1 file changed
+127 -45
Download Patch File Download Diff File Expand all files Collapse all files
src/db.zig
+127 -45
View File
@@ -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) {
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) {
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(.{});