Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124//! Durable upstream relay cursor in the shared metadata database.
const std = @import("std");const meta_store = @import("../storage/meta_store.zig");
const log = std.log.scoped(.stream);const key = "relay/cursor";const encoding_version: u8 = 1;pub const flush_interval_us: i64 = 5 * std.time.us_per_s;
pub const Store = struct { meta: *meta_store.Store, last_flushed: i64 = 0, pending: ?i64 = null,
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) !?i64 { 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.InvalidRelayCursorLength; if (bytes[0] != encoding_version) return error.UnsupportedRelayCursorVersion; const raw = std.mem.readInt(u64, bytes[1..9], .little); if (raw > std.math.maxInt(i64)) return error.NegativeRelayCursor; return @intCast(raw); }
pub fn update(self: *Store, io: std.Io, seq: i64, now_us: i64) void { self.pending = seq; if (now_us - self.last_flushed < flush_interval_us) return; self.flush(io); self.last_flushed = now_us; }
pub fn flush(self: *Store, _: std.Io) void { const seq = self.pending orelse return; if (seq < 0) { log.err("refusing to flush negative cursor {d}", .{seq}); return; } var bytes: [9]u8 = undefined; bytes[0] = encoding_version; std.mem.writeInt(u64, bytes[1..9], @intCast(seq), .little); self.meta.putDurable(key, &bytes) catch |err| { log.warn("cursor flush failed: {s}", .{@errorName(err)}); return; }; self.pending = null; }
pub fn stage(writes: *meta_store.Batch, seq: i64) !void { if (seq < 0) return error.NegativeRelayCursor; var bytes: [9]u8 = undefined; bytes[0] = encoding_version; std.mem.writeInt(u64, bytes[1..9], @intCast(seq), .little); writes.put(key, &bytes); }};
test "cursor round-trip and interval flushing" { 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.expect(try store.load(testing.io) == null); store.update(testing.io, 41, 10 * std.time.us_per_s); try testing.expectEqual(@as(?i64, 41), try store.load(testing.io)); store.update(testing.io, 42, 11 * std.time.us_per_s); try testing.expectEqual(@as(?i64, 41), try store.load(testing.io)); store.flush(testing.io); try testing.expectEqual(@as(?i64, 42), try store.load(testing.io));}
test "cursor rejects malformed version width negative values and storage errors" { 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.InvalidRelayCursorLength, store.load(testing.io)); try meta.putDurable(key, &.{ 2, 0, 0, 0, 0, 0, 0, 0, 0 }); try testing.expectError(error.UnsupportedRelayCursorVersion, store.load(testing.io)); try meta.putDurable(key, &.{ 1, 0, 0, 0, 0, 0, 0, 0, 0x80 }); try testing.expectError(error.NegativeRelayCursor, store.load(testing.io)); var writes = meta.batch(); defer writes.deinit(); try testing.expectError(error.NegativeRelayCursor, Store.stage(&writes, -1)); try testing.expect(meta_store.armReadFault(key, 1)); defer meta_store.disarmReadFault(); try testing.expectError(error.InjectedStoreFault, store.load(testing.io));}
test "legacy bare cursor is migrated before strict decoding" { 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); const path = path_buf[0..path_len]; var meta = try meta_store.Store.open(testing.allocator, path); defer meta.deinit(); var legacy: [8]u8 = undefined; std.mem.writeInt(i64, &legacy, 73, .little); try meta.putDurable(key, &legacy); try meta.migrateLegacy(testing.io, path); var store = Store.init(&meta); try testing.expectEqual(@as(?i64, 73), try store.load(testing.io));}