Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980//! Durable physical-compaction watermark in the shared metadata database.
const std = @import("std");const meta_store = @import("../storage/meta_store.zig");
const log = std.log.scoped(.stream);pub const version: u8 = 1;const key = "compaction/seq";
pub const Store = struct { meta: *meta_store.Store,
pub fn init(meta: *meta_store.Store) Store { return .{ .meta = meta }; } pub fn deinit(_: *Store, _: std.Io) void {}
pub fn load(self: *Store, _: std.Io) !?u64 { const maybe_bytes = try self.meta.getAlloc(self.meta.allocator, key); const bytes = maybe_bytes orelse return null; defer self.meta.allocator.free(bytes); if (bytes.len != 9) return error.InvalidCompactionWatermarkLength; if (bytes[0] != version) return error.UnsupportedCompactionWatermarkVersion; return std.mem.readInt(u64, bytes[1..9], .little); }
pub fn save(self: *Store, _: std.Io, watermark: u64) !void { var payload: [9]u8 = undefined; payload[0] = version; std.mem.writeInt(u64, payload[1..9], watermark, .little); try self.meta.putDurable(key, &payload); }
pub fn initFloor(self: *Store, io: std.Io, next_seq: u64) !u64 { if (try self.load(io)) |w| return w; const w = next_seq -| 1; try self.save(io, w); log.info("compaction watermark floored at {d}", .{w}); return w; }};
test "watermark floor and updates" { const testing = std.testing; var tmp = testing.tmpDir(.{}); defer tmp.cleanup(); var path_buf: [std.fs.max_path_bytes]u8 = undefined; const path_len = try tmp.dir.realPath(testing.io, &path_buf); var meta = try meta_store.Store.open(testing.allocator, path_buf[0..path_len]); defer meta.deinit(); var store = Store.init(&meta); try testing.expectEqual(@as(?u64, null), try store.load(testing.io)); try testing.expectEqual(@as(u64, 0), try store.initFloor(testing.io, 0)); try store.save(testing.io, 99); try testing.expectEqual(@as(?u64, 99), try store.load(testing.io));}
test "watermark corruption and storage errors fail closed without initialization" { const testing = std.testing; var tmp = testing.tmpDir(.{}); defer tmp.cleanup(); var path_buf: [std.fs.max_path_bytes]u8 = undefined; const path_len = try tmp.dir.realPath(testing.io, &path_buf); var meta = try meta_store.Store.open(testing.allocator, path_buf[0..path_len]); defer meta.deinit(); var store = Store.init(&meta);
try meta.putDurable(key, &.{ 1, 2, 3 }); try testing.expectError(error.InvalidCompactionWatermarkLength, store.initFloor(testing.io, 100)); const malformed = (try meta.getAlloc(testing.allocator, key)).?; defer testing.allocator.free(malformed); try testing.expectEqualSlices(u8, &.{ 1, 2, 3 }, malformed);
try meta.putDurable(key, &.{ 2, 0, 0, 0, 0, 0, 0, 0, 0 }); try testing.expectError(error.UnsupportedCompactionWatermarkVersion, store.load(testing.io)); try testing.expect(meta_store.armReadFault(key, 1)); defer meta_store.disarmReadFault(); try testing.expectError(error.InjectedStoreFault, store.initFloor(testing.io, 100));}