feat: add parseManifestId and recordBodyLen to version

src/version.zig

@@ -314,3 +314,162 @@ fn writeCurrent(dir: std.fs.Dir, manifestName: []const u8) !void {
     try dir.rename(tmpName, currentName);
 }
 
+fn parseManifestId(name: []const u8) ?u64 {
+    if (!std.mem.startsWith(u8, name, manifestPrefix)) return null;
+    return std.fmt.parseInt(u64, name[manifestPrefix.len..], 10) catch null;
+}
+
+fn recordBodyLen(data: []const u8, bodyStart: usize) ?usize {
+    if (bodyStart + 4 > data.len) return null;
+    const addedCount = encode.readU32(data, bodyStart);
+    var p = bodyStart + 4;
+    var i: u32 = 0;
+    while (i < addedCount) : (i += 1) {
+        if (p + 12 > data.len) return null;
+        p += 12;
+    }
+    if (p + 4 > data.len) return null;
+    const deletedCount = encode.readU32(data, p);
+    p += 4;
+    var j: u32 = 0;
+    while (j < deletedCount) : (j += 1) {
+        if (p + 8 > data.len) return null;
+        p += 8;
+    }
+    // Trailing counter watermarks: one presence flag byte plus an optional
+    // u64 each, for nextFileId then nextSeq.
+    var k: u32 = 0;
+    while (k < 2) : (k += 1) {
+        if (p + 1 > data.len) return null;
+        const present = data[p];
+        p += 1;
+        if (present != 0) {
+            if (p + 8 > data.len) return null;
+            p += 8;
+        }
+    }
+    return p - bodyStart;
+}
+
+fn replayRecord(state: *std.AutoHashMap(u64, usize), counters: *Counters, body: []const u8) !void {
+    const addedCount = encode.readU32(body, 0);
+    var p: usize = 4;
+    var i: u32 = 0;
+    while (i < addedCount) : (i += 1) {
+        const fileId = encode.readU64(body, p);
+        const level = encode.readU32(body, p + 8);
+        try state.put(fileId, level);
+        p += 12;
+    }
+    const deletedCount = encode.readU32(body, p);
+    p += 4;
+    var j: u32 = 0;
+    while (j < deletedCount) : (j += 1) {
+        const fileId = encode.readU64(body, p);
+        _ = state.remove(fileId);
+        p += 8;
+    }
+    if (p < body.len) {
+        const hasNextFileId = body[p];
+        p += 1;
+        if (hasNextFileId != 0) {
+            counters.nextFileId = encode.readU64(body, p);
+            p += 8;
+        }
+        const hasNextSeq = body[p];
+        p += 1;
+        if (hasNextSeq != 0) {
+            counters.nextSeq = encode.readU64(body, p);
+            p += 8;
+        }
+    }
+}
+
+test "fresh directory bootstraps an empty version set" {
+    const allocator = std.testing.allocator;
+    var tmp = std.testing.tmpDir(.{});
+    defer tmp.cleanup();
+
+    var opened = try VersionSet.open(allocator, tmp.dir);
+    defer opened.state.deinit();
+    defer opened.set.close();
+
+    try std.testing.expectEqual(@as(usize, 0), opened.state.count());
+    try std.testing.expectEqual(@as(u64, 1), opened.set.manifestId);
+    try std.testing.expect(tmp.dir.access(currentName, .{}) != error.FileNotFound);
+}
+
+test "logAndApply records adds and survives reopen" {
+    const allocator = std.testing.allocator;
+    var tmp = std.testing.tmpDir(.{});
+    defer tmp.cleanup();
+
+    {
+        var opened = try VersionSet.open(allocator, tmp.dir);
+        defer opened.state.deinit();
+        defer opened.set.close();
+
+        var edit = VersionEdit.init(allocator);
+        defer edit.deinit();
+        try edit.addFile(1, 0);
+        try edit.addFile(2, 0);
+        try opened.set.logAndApply(&edit, &opened.state);
+
+        try std.testing.expectEqual(@as(usize, 2), opened.state.count());
+        try std.testing.expectEqual(@as(usize, 0), opened.state.get(1).?);
+    }
+
+    var reopened = try VersionSet.open(allocator, tmp.dir);
+    defer reopened.state.deinit();
+    defer reopened.set.close();
+    try std.testing.expectEqual(@as(usize, 2), reopened.state.count());
+    try std.testing.expectEqual(@as(usize, 0), reopened.state.get(2).?);
+}
+
+test "logAndApply replaces a deleted file with new outputs (compaction pattern)" {
+    const allocator = std.testing.allocator;
+    var tmp = std.testing.tmpDir(.{});
+    defer tmp.cleanup();
+
+    var opened = try VersionSet.open(allocator, tmp.dir);
+    defer opened.state.deinit();
+    defer opened.set.close();
+
+    var e1 = VersionEdit.init(allocator);
+    defer e1.deinit();
+    try e1.addFile(1, 0);
+    try e1.addFile(2, 0);
+    try opened.set.logAndApply(&e1, &opened.state);
+
+    var e2 = VersionEdit.init(allocator);
+    defer e2.deinit();
+    try e2.deleteFile(1);
+    try e2.deleteFile(2);
+    try e2.addFile(3, 1);
+    try opened.set.logAndApply(&e2, &opened.state);
+
+    try std.testing.expectEqual(@as(usize, 1), opened.state.count());
+    try std.testing.expectEqual(@as(usize, 1), opened.state.get(3).?);
+    try std.testing.expectEqual(@as(?usize, null), opened.state.get(1));
+}
+
+test "manifest rotates once pending records outgrow live file count" {
+    const allocator = std.testing.allocator;
+    var tmp = std.testing.tmpDir(.{});
+    defer tmp.cleanup();
+
+    var opened = try VersionSet.open(allocator, tmp.dir);
+    defer opened.state.deinit();
+    defer opened.set.close();
+
+    // Simulate steady-state leveled compaction: the live file count stays
+    // fixed at 1 (each round replaces the previous output), but the edit
+    // log keeps growing until it is worth rotating to a fresh snapshot.
+    var fileId: u64 = 1;
+    {
+        var edit = VersionEdit.init(allocator);
+        defer edit.deinit();
+        try edit.addFile(fileId, 0);
+        try opened.set.logAndApply(&edit, &opened.state);
+    }
+