Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383//! Phase B timestamp-import bucketing.//!//! A concrete sealed-segment catalog loads the resident DID blooms without//! reading segment bodies. Valid CSV rows are routed to per-segment packed//! offset files through bounded least-recently-used DID and file caches. The//! import lifecycle holds the archive/compaction topology lock while this//! point-in-time catalog is in use; rows born after the preceding force-seal//! are stamped by the durable append rule store.
const std = @import("std");const gloom = @import("../storage/gloom.zig");const segment = @import("../storage/segment.zig");const segment_io = @import("../storage/segment_io.zig");const parse = @import("parse.zig");
const Allocator = std.mem.Allocator;const Io = std.Io;
pub const default_did_cache_size: usize = 100_000;pub const default_open_file_limit: usize = 256;pub const offset_record_size = 8;
pub const Stats = struct { rows_routed: u64 = 0, rows_no_candidate: u64 = 0, offsets_written: u64 = 0, did_cache_hits: u64 = 0, did_cache_misses: u64 = 0, segments_touched: usize = 0, selector_calls: u64 = 0,};
pub const SegmentRef = struct { index: u64, name: []u8, bloom: gloom.Filter,};
/// Immutable point-in-time view of sealed segments. Opening fails on an/// unreadable resident bloom: treating corruption as a miss could silently/// strand timestamp intent.pub const Catalog = struct { allocator: Allocator, io: Io, segments: std.ArrayList(SegmentRef) = .empty,
pub fn open(allocator: Allocator, io: Io, segments_dir: Io.Dir) !Catalog { var self: Catalog = .{ .allocator = allocator, .io = io }; errdefer self.deinit(); var iterator = segments_dir.iterate(); while (try iterator.next(io)) |entry| { const index = parseSegmentName(entry.name) orelse continue; var file = try segments_dir.openFile(io, entry.name, .{}); defer file.close(io); var header_bytes: [segment.header_size]u8 = undefined; if (try file.readPositionalAll(io, &header_bytes, 0) != header_bytes.len) return error.TruncatedSegmentHeader; const header = segment.Header.decode(&header_bytes) catch |err| switch (err) { error.ActiveSegment => continue, else => return err, }; if (header.block_did_bloom_offset < header.did_bloom_offset) return error.BadSegmentBloomOffsets; const bloom_len = header.block_did_bloom_offset - header.did_bloom_offset; if (bloom_len > 1 << 30) return error.BadSegmentBloomOffsets; const bloom_bytes = try allocator.alloc(u8, @intCast(bloom_len)); defer allocator.free(bloom_bytes); if (try file.readPositionalAll(io, bloom_bytes, header.did_bloom_offset) != bloom_bytes.len) return error.TruncatedSegmentBloom; const filter = try gloom.unmarshal(allocator, bloom_bytes); errdefer { var owned = filter; owned.deinit(allocator); } const name = try allocator.dupe(u8, entry.name); errdefer allocator.free(name); try self.segments.append(allocator, .{ .index = index, .name = name, .bloom = filter, }); } std.mem.sort(SegmentRef, self.segments.items, {}, struct { fn less(_: void, a: SegmentRef, b: SegmentRef) bool { return a.index < b.index; } }.less); return self; }
pub fn deinit(self: *Catalog) void { for (self.segments.items) |*item| { self.allocator.free(item.name); item.bloom.deinit(self.allocator); } self.segments.deinit(self.allocator); self.* = undefined; }
fn select(self: *const Catalog, allocator: Allocator, did: []const u8) ![]usize { var candidates: std.ArrayList(usize) = .empty; errdefer candidates.deinit(allocator); for (self.segments.items, 0..) |*item, index| { if (item.bloom.contains(did)) try candidates.append(allocator, index); } return candidates.toOwnedSlice(allocator); }};
const CacheEntry = struct { candidates: []usize, used: u64,};
const OpenOffset = struct { segment_index: u64, file: Io.File, size: u64, used: u64,};
pub const Config = struct { did_cache_size: usize = 0, open_file_limit: usize = 0,};
pub const Bucketer = struct { allocator: Allocator, io: Io, catalog: *const Catalog, job_path: []u8, job_dir: Io.Dir, did_cache_size: usize, open_file_limit: usize, cache: std.StringHashMapUnmanaged(CacheEntry) = .empty, files: std.ArrayList(OpenOffset) = .empty, touched: std.AutoHashMapUnmanaged(u64, void) = .empty, clock: u64 = 0, stats: Stats = .{}, closed: bool = false,
pub fn init( allocator: Allocator, io: Io, catalog: *const Catalog, job_path: []const u8, config: Config, ) !Bucketer { if (job_path.len == 0) return error.MissingJobPath; try Io.Dir.cwd().createDirPath(io, job_path); const dir = try Io.Dir.cwd().openDir(io, job_path, .{ .iterate = true }); errdefer dir.close(io); return .{ .allocator = allocator, .io = io, .catalog = catalog, .job_path = try allocator.dupe(u8, job_path), .job_dir = dir, .did_cache_size = if (config.did_cache_size == 0) default_did_cache_size else config.did_cache_size, .open_file_limit = if (config.open_file_limit == 0) default_open_file_limit else config.open_file_limit, }; }
pub fn deinit(self: *Bucketer) void { if (!self.closed) self.close() catch {}; var iterator = self.cache.iterator(); while (iterator.next()) |entry| { self.allocator.free(entry.key_ptr.*); self.allocator.free(entry.value_ptr.candidates); } self.cache.deinit(self.allocator); self.files.deinit(self.allocator); self.touched.deinit(self.allocator); self.job_dir.close(self.io); self.allocator.free(self.job_path); self.* = undefined; }
pub fn routeOpaque(raw: ?*anyopaque, row: parse.Row) anyerror!void { const self: *Bucketer = @ptrCast(@alignCast(raw.?)); try self.route(row); }
pub fn route(self: *Bucketer, row: parse.Row) !void { if (self.closed) return error.BucketerClosed; const candidates = try self.select(row.did); if (candidates.len == 0) { self.stats.rows_no_candidate += 1; return; } for (candidates) |candidate| { try self.appendOffset(self.catalog.segments.items[candidate].index, row.offset); self.stats.offsets_written += 1; } self.stats.rows_routed += 1; }
fn select(self: *Bucketer, did: []const u8) ![]const usize { self.clock += 1; if (self.cache.getPtr(did)) |entry| { entry.used = self.clock; self.stats.did_cache_hits += 1; return entry.candidates; } self.stats.did_cache_misses += 1; self.stats.selector_calls += 1; if (self.cache.count() >= self.did_cache_size) self.evictDID(); const owned_did = try self.allocator.dupe(u8, did); errdefer self.allocator.free(owned_did); const candidates = try self.catalog.select(self.allocator, did); errdefer self.allocator.free(candidates); try self.cache.put(self.allocator, owned_did, .{ .candidates = candidates, .used = self.clock }); return self.cache.get(did).?.candidates; }
fn evictDID(self: *Bucketer) void { var oldest_key: ?[]const u8 = null; var oldest_tick: u64 = std.math.maxInt(u64); var iterator = self.cache.iterator(); while (iterator.next()) |entry| { if (entry.value_ptr.used < oldest_tick) { oldest_tick = entry.value_ptr.used; oldest_key = entry.key_ptr.*; } } if (oldest_key) |key| { const removed = self.cache.fetchRemove(key).?; self.allocator.free(removed.key); self.allocator.free(removed.value.candidates); } }
fn appendOffset(self: *Bucketer, segment_index: u64, offset: u64) !void { self.clock += 1; var slot: ?usize = null; for (self.files.items, 0..) |item, index| { if (item.segment_index == segment_index) { slot = index; break; } } if (slot == null) { if (self.files.items.len >= self.open_file_limit) try self.evictFile(); var name_buffer: [64]u8 = undefined; const name = try offsetFileName(&name_buffer, segment_index); const file = try self.job_dir.createFile(self.io, name, .{ .read = true, .truncate = false }); errdefer file.close(self.io); const stat = try file.stat(self.io); try self.files.append(self.allocator, .{ .segment_index = segment_index, .file = file, .size = stat.size, .used = self.clock, }); slot = self.files.items.len - 1; } const open = &self.files.items[slot.?]; open.used = self.clock; var encoded: [offset_record_size]u8 = undefined; std.mem.writeInt(u64, &encoded, offset, .little); try open.file.writePositionalAll(self.io, &encoded, open.size); open.size += encoded.len; try self.touched.put(self.allocator, segment_index, {}); }
fn evictFile(self: *Bucketer) !void { var oldest: usize = 0; for (self.files.items, 0..) |item, index| { if (item.used < self.files.items[oldest].used) oldest = index; } const item = self.files.items[oldest]; try item.file.sync(self.io); item.file.close(self.io); _ = self.files.orderedRemove(oldest); }
/// Sync all offset contents, close them, then sync the job directory and /// its parent before the lifecycle may persist `bucketed=true`. pub fn close(self: *Bucketer) !void { if (self.closed) return; var first_error: ?anyerror = null; for (self.files.items) |item| { item.file.sync(self.io) catch |err| if (first_error == null) { first_error = err; }; item.file.close(self.io); } self.files.clearRetainingCapacity(); if (first_error) |err| return err; try syncDir(self.io, self.job_dir); const parent_path = std.fs.path.dirname(self.job_path) orelse "."; // On Linux Zig's default directory handle may be O_PATH, which cannot // be fsynced (EBADF/Unexpected). Request an iterable directory so the // parent-dir durability barrier uses a real readable fd. var parent = try Io.Dir.cwd().openDir(self.io, parent_path, .{ .iterate = true }); defer parent.close(self.io); try syncDir(self.io, parent); self.closed = true; }
pub fn statsSnapshot(self: *const Bucketer) Stats { var result = self.stats; result.segments_touched = self.touched.count(); return result; }};
pub fn offsetFileName(buffer: *[64]u8, index: u64) ![]const u8 { return std.fmt.bufPrint(buffer, "offsets_{d:0>10}.bin", .{index});}
pub fn parseOffsetFileName(name: []const u8) ?u64 { if (!std.mem.startsWith(u8, name, "offsets_") or !std.mem.endsWith(u8, name, ".bin")) return null; const digits = name[8 .. name.len - 4]; if (digits.len != 10) return null; const index = std.fmt.parseInt(u64, digits, 10) catch return null; var buffer: [64]u8 = undefined; const canonical = offsetFileName(&buffer, index) catch return null; return if (std.mem.eql(u8, name, canonical)) index else null;}
fn parseSegmentName(name: []const u8) ?u64 { if (!std.mem.startsWith(u8, name, "seg_") or !std.mem.endsWith(u8, name, ".jss")) return null; const digits = name[4 .. name.len - 4]; if (digits.len != 10) return null; // Archive segment ordinals are fixed-width lowercase base-36. Decimal // parsing happens to work for the first ten files, then silently drops // every filename containing a-z from the timestamp-import catalog. const index = std.fmt.parseInt(u64, digits, 36) catch return null; var canonical_digits: [10]u8 = @splat('0'); var value = index; var position: usize = canonical_digits.len; while (value > 0) : (value /= 36) { if (position == 0) return null; position -= 1; canonical_digits[position] = std.fmt.digitToChar(@intCast(value % 36), .lower); } return if (std.mem.eql(u8, digits, &canonical_digits)) index else null;}
fn syncDir(io: Io, dir: Io.Dir) !void { try segment_io.syncDir(dir, io);}
test "catalog discovers real base-36 segment names beyond nine" { const testing = std.testing; const writer = @import("../storage/segment_writer.zig"); var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); var tmp = testing.tmpDir(.{}); defer tmp.cleanup(); var segments_dir = try tmp.dir.createDirPathOpen(io, "segments", .{ .open_options = .{ .iterate = true }, }); defer segments_dir.close(io);
var active = try writer.ActiveWriter.init(testing.allocator); defer active.deinit(); try active.append(.{ .seq = 1, .witnessed_at = 1, .indexed_at = 0, .kind = .create, .did = "did:plc:base36", .collection = "app.bsky.feed.post", .rkey = "one", .rev = "3lbase36", .payload = "payload", }); try active.seal(); var file = try segments_dir.createFile(io, "seg_000000000a.jss", .{}); defer file.close(io); try file.writeStreamingAll(io, active.bytes()); try file.sync(io);
var catalog = try Catalog.open(testing.allocator, io, segments_dir); defer catalog.deinit(); try testing.expectEqual(@as(usize, 1), catalog.segments.items.len); try testing.expectEqual(@as(u64, 10), catalog.segments.items[0].index); try testing.expect(catalog.segments.items[0].bloom.contains("did:plc:base36"));}