Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401//! The single durable metadata database for Stream.//!//! Segment files own event bytes. Everything structured that cannot be//! recovered cheaply from a segment scan belongs here. A synced metadata//! batch may acknowledge segment bytes only after those bytes have been//! fsynced; callers use `Batch.commitDurable` for that boundary.
const std = @import("std");const rocksdb = @import("rocksdb");const metrics = @import("../runtime/metrics.zig");
const Allocator = std.mem.Allocator;const Io = std.Io;const log = std.log.scoped(.stream_meta);
/// Deterministic, process-wide metadata fault seam used only by the offline/// restart oracle. Production leaves this unarmed. The operation still runs/// through the real Store and RocksDB schema; the selected durable mutation/// returns an I/O failure immediately before it reaches RocksDB so callers'/// fail-loud and restart behavior can be exercised without corrupting bytes.const Fault = struct { prefix: ?[]const u8 = null, ordinal: u64 = 0, seen: std.atomic.Value(u64) = .init(0),
fn check(self: *Fault, key: []const u8) !void { const prefix = self.prefix orelse return; if (!std.mem.startsWith(u8, key, prefix)) return; const seen = self.seen.fetchAdd(1, .monotonic) + 1; if (seen != self.ordinal) return; log.warn("store fault: injected I/O failure for prefix {s} at ordinal {d}", .{ prefix, seen }); return error.InjectedStoreFault; }};
var fault: Fault = .{};var read_fault: Fault = .{};
/// Arm one matching durable metadata operation. Must be called during process/// startup, before worker threads exist. Returns false for an empty prefix or/// zero ordinal so an invalid oracle configuration fails closed.pub fn armFault(prefix: []const u8, ordinal: u64) bool { if (prefix.len == 0 or ordinal == 0) return false; fault = .{ .prefix = prefix, .ordinal = ordinal }; return true;}
pub fn disarmFault() void { fault = .{};}
/// Independent read-side fault seam. Keeping it separate from durable mutation/// faults preserves the existing crash schedules while allowing startup code/// to prove that a storage error is not misclassified as a missing key.pub fn armReadFault(prefix: []const u8, ordinal: u64) bool { if (prefix.len == 0 or ordinal == 0) return false; read_fault = .{ .prefix = prefix, .ordinal = ordinal }; return true;}
pub fn disarmReadFault() void { read_fault = .{};}
pub const subdir = "meta.rocksdb";
pub const Store = struct { allocator: Allocator, db: rocksdb.DB, default_cf: rocksdb.ColumnFamilyHandle, io: ?Io = null, stats: ?*metrics.Stats = null,
pub fn open(allocator: Allocator, data_dir: []const u8) !Store { return openInner(allocator, data_dir, null, null); }
pub fn openWithStats(allocator: Allocator, io: Io, data_dir: []const u8, stats: *metrics.Stats) !Store { return openInner(allocator, data_dir, io, stats); }
fn openInner(allocator: Allocator, data_dir: []const u8, io: ?Io, stats: ?*metrics.Stats) !Store { const path = try std.fs.path.join(allocator, &.{ data_dir, subdir }); defer allocator.free(path);
var err_data: ?rocksdb.Data = null; defer if (err_data) |e| e.deinit(); const db, const families = rocksdb.DB.open( allocator, path, .{ .create_if_missing = true, .create_missing_column_families = true, }, &.{.{ .name = "default", .options = .{ .bloom_bits_per_key = 10 } }}, false, &err_data, ) catch |err| { if (err_data) |e| log.err("open {s}: {s}", .{ path, e.data }); return err; }; defer allocator.free(families); if (families.len != 1) return error.InvalidMetadataStore;
log.info("metadata store opened at {s}", .{path}); return .{ .allocator = allocator, .db = db, .default_cf = families[0].handle, .io = io, .stats = stats, }; }
pub fn deinit(self: *Store) void { self.db.deinit(); }
/// One-way import for canaries created before the unified metadata DB. /// The legacy files remain untouched as rollback material; existing DB /// keys always win, making this safe on every restart. pub fn migrateLegacy(self: *Store, io: std.Io, data_dir: []const u8) !void { if (try self.getAlloc(self.allocator, "relay/cursor")) |existing| { defer self.allocator.free(existing); // Stream previously stored a bare signed little-endian value. // Convert that known schema exactly once; all other malformed or // unknown-version values remain untouched so the strict loader // fails closed instead of laundering corruption. if (existing.len == 8) { const seq = std.mem.readInt(i64, existing[0..8], .little); if (seq < 0) return error.NegativeRelayCursor; var versioned: [9]u8 = undefined; versioned[0] = 1; std.mem.writeInt(u64, versioned[1..9], @intCast(seq), .little); try self.putDurable("relay/cursor", &versioned); log.info("migrated relay cursor to versioned encoding", .{}); } } else { var path_buf: [std.fs.max_path_bytes]u8 = undefined; const path = try std.fmt.bufPrint(&path_buf, "{s}/cursor", .{data_dir}); var text_buf: [64]u8 = undefined; if (std.Io.Dir.cwd().readFile(io, path, &text_buf)) |text| { if (std.fmt.parseInt(i64, std.mem.trim(u8, text, " \n\t"), 10)) |seq| { if (seq < 0) return error.NegativeRelayCursor; var bytes: [9]u8 = undefined; bytes[0] = 1; std.mem.writeInt(u64, bytes[1..9], @intCast(seq), .little); try self.putDurable("relay/cursor", &bytes); log.info("imported legacy relay cursor {d}", .{seq}); } else |_| {} } else |_| {} }
if (try self.getAlloc(self.allocator, "phase")) |existing| { self.allocator.free(existing); } else { var path_buf: [std.fs.max_path_bytes]u8 = undefined; const path = try std.fmt.bufPrint(&path_buf, "{s}/phase", .{data_dir}); var text_buf: [32]u8 = undefined; if (std.Io.Dir.cwd().readFile(io, path, &text_buf)) |text| { const phase = std.mem.trim(u8, text, " \n\t"); if (phase.len > 0) try self.putDurable("phase", phase); } else |_| {} }
if (try self.getAlloc(self.allocator, "compaction/seq")) |existing| { self.allocator.free(existing); } else { var path_buf: [std.fs.max_path_bytes]u8 = undefined; const path = try std.fmt.bufPrint(&path_buf, "{s}/compaction_watermark", .{data_dir}); var bytes: [9]u8 = undefined; if (std.Io.Dir.cwd().readFile(io, path, &bytes)) |value| { if (value.len == bytes.len) try self.putDurable("compaction/seq", value); } else |_| {} }
if (try self.getAlloc(self.allocator, "merge/next_source_idx")) |existing| { defer self.allocator.free(existing); if (existing.len == 8) { var versioned: [9]u8 = undefined; versioned[0] = 1; @memcpy(versioned[1..9], existing); try self.putDurable("merge/next_source_idx", &versioned); log.info("migrated merge cursor to versioned encoding", .{}); } } }
pub fn getAlloc(self: *Store, allocator: Allocator, key: []const u8) !?[]u8 { const started_at = self.started(); read_fault.check(key) catch |err| { self.observe(.get, .error_result, started_at); return err; }; var err_data: ?rocksdb.Data = null; defer if (err_data) |e| e.deinit(); const value = self.db.get(self.default_cf, key, &err_data) catch |err| { self.observe(.get, .error_result, started_at); if (err_data) |e| log.err("get {s}: {s}", .{ key, e.data }); return err; } orelse { self.observe(.get, .notfound, started_at); return null; }; self.observe(.get, .ok, started_at); defer value.deinit(); return try allocator.dupe(u8, value.data); }
pub fn putDurable(self: *Store, key: []const u8, value: []const u8) !void { const started_at = self.started(); fault.check(key) catch |err| { self.observe(.set, .error_result, started_at); return err; }; var err_data: ?rocksdb.Data = null; defer if (err_data) |e| e.deinit(); self.db.putWithOptions(self.default_cf, key, value, .{ .sync = true }, &err_data) catch |err| { self.observe(.set, .error_result, started_at); if (err_data) |e| log.err("durable put {s}: {s}", .{ key, e.data }); return err; }; self.observe(.set, .ok, started_at); }
pub fn deleteDurable(self: *Store, key: []const u8) !void { const started_at = self.started(); fault.check(key) catch |err| { self.observe(.delete, .error_result, started_at); return err; }; var err_data: ?rocksdb.Data = null; defer if (err_data) |e| e.deinit(); self.db.deleteWithOptions(self.default_cf, key, .{ .sync = true }, &err_data) catch |err| { self.observe(.delete, .error_result, started_at); if (err_data) |e| log.err("durable delete {s}: {s}", .{ key, e.data }); return err; }; self.observe(.delete, .ok, started_at); }
pub fn iterator(self: *Store, start: ?[]const u8) rocksdb.Iterator { return self.db.iterator(self.default_cf, .forward, start); }
pub fn batch(self: *Store) Batch { return .{ .store = self, .inner = rocksdb.WriteBatch.init() }; }
fn started(self: *const Store) ?Io.Timestamp { const io = self.io orelse return null; return Io.Timestamp.now(io, .awake); }
fn observe(self: *Store, op: metrics.StoreOp, status: metrics.StoreStatus, started_at: ?Io.Timestamp) void { const stats = self.stats orelse return; const io = self.io orelse return; const started_value = started_at orelse return; const elapsed = started_value.durationTo(Io.Timestamp.now(io, .awake)).toMicroseconds(); stats.observeStore(op, status, @intCast(@max(elapsed, 0))); }};
pub const Batch = struct { store: *Store, inner: rocksdb.WriteBatch, fault_key: ?[]const u8 = null,
pub fn deinit(self: *Batch) void { self.inner.deinit(); }
pub fn put(self: *Batch, key: []const u8, value: []const u8) void { if (fault.prefix) |prefix| { if (self.fault_key == null and std.mem.startsWith(u8, key, prefix)) self.fault_key = key; } self.inner.put(self.store.default_cf, key, value); }
pub fn delete(self: *Batch, key: []const u8) void { if (fault.prefix) |prefix| { if (self.fault_key == null and std.mem.startsWith(u8, key, prefix)) self.fault_key = key; } self.inner.delete(self.store.default_cf, key); }
pub fn commitDurable(self: *Batch) !void { const started_at = self.store.started(); if (self.fault_key) |key| fault.check(key) catch |err| { self.store.observe(.batch_commit, .error_result, started_at); return err; }; var err_data: ?rocksdb.Data = null; defer if (err_data) |e| e.deinit(); self.store.db.writeWithOptions(self.inner, .{ .sync = true }, &err_data) catch |err| { self.store.observe(.batch_commit, .error_result, started_at); if (err_data) |e| log.err("durable batch commit: {s}", .{e.data}); return err; }; self.store.observe(.batch_commit, .ok, started_at); }};
test "durable point reads and atomic batches survive reopen" { 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 store = try Store.open(testing.allocator, path); defer store.deinit(); var batch = store.batch(); defer batch.deinit(); batch.put("relay/cursor", "41"); batch.put("repo/did:plc:alice", "complete\x003lrev"); try batch.commitDurable(); } { var store = try Store.open(testing.allocator, path); defer store.deinit(); const cursor = (try store.getAlloc(testing.allocator, "relay/cursor")).?; defer testing.allocator.free(cursor); try testing.expectEqualStrings("41", cursor); const repo = (try store.getAlloc(testing.allocator, "repo/did:plc:alice")).?; defer testing.allocator.free(repo); try testing.expectEqualStrings("complete\x003lrev", repo); }}
test "real RocksDB operations populate canonical store metric outcomes" { 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 stats: metrics.Stats = .{}; var store = try Store.openWithStats(testing.allocator, testing.io, path_buf[0..path_len], &stats); defer store.deinit();
try store.putDurable("point", "value"); const found = (try store.getAlloc(testing.allocator, "point")).?; testing.allocator.free(found); try testing.expectEqual(null, try store.getAlloc(testing.allocator, "missing")); try store.deleteDurable("point"); var writes = store.batch(); defer writes.deinit(); writes.put("batch/a", "1"); writes.put("batch/b", "2"); try writes.commitDurable();
try testing.expectEqual(1, stats.store_op_duration_count[storeMetricIndex(.get, .ok)].load(.monotonic)); try testing.expectEqual(1, stats.store_op_duration_count[storeMetricIndex(.get, .notfound)].load(.monotonic)); try testing.expectEqual(1, stats.store_op_duration_count[storeMetricIndex(.set, .ok)].load(.monotonic)); try testing.expectEqual(1, stats.store_op_duration_count[storeMetricIndex(.delete, .ok)].load(.monotonic)); try testing.expectEqual(1, stats.store_op_duration_count[storeMetricIndex(.batch_commit, .ok)].load(.monotonic));}
test "armed durable store fault fires once at matching batch ordinal" { const testing = std.testing; disarmFault(); defer disarmFault(); 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 store = try Store.open(testing.allocator, path_buf[0..path_len]); defer store.deinit();
try testing.expect(armFault("merge/next_source_idx", 2)); for (0..2) |i| { var writes = store.batch(); defer writes.deinit(); writes.put("repo/did:plc:alice", "complete"); writes.put("merge/next_source_idx", "1"); if (i == 0) { try writes.commitDurable(); } else { try testing.expectError(error.InjectedStoreFault, writes.commitDurable()); } }
// The rejected batch never reached RocksDB. Disarming and retrying the // exact real batch succeeds, which is the recovery contract used by the // process oracle. disarmFault(); var retry = store.batch(); defer retry.deinit(); retry.put("merge/next_source_idx", "2"); try retry.commitDurable(); const cursor = (try store.getAlloc(testing.allocator, "merge/next_source_idx")).?; defer testing.allocator.free(cursor); try testing.expectEqualStrings("2", cursor);}
fn storeMetricIndex(op: metrics.StoreOp, status: metrics.StoreStatus) usize { return @backingInt(op) * @typeInfo(metrics.StoreStatus).@"enum".field_names.len + @backingInt(status);}