Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217//! jss segment format v1 — writer, part 2: footer + header finalization//!//! shared by the active-writer seal path and the compaction rewrite: takes//! per-block metadata (a populated index entry + the block's events) and//! appends the four footer sections (block index, segment DID bloom,//! per-block DID blooms, collection index), then finalizes the 256-byte//! header with the file xxh3. envelope values are the caller's: seal//! recomputes them from the walked frames; rewrite preserves the//! historical (superset) bounds so block seq-pruning stays safe across//! compaction generations.
const std = @import("std");const gloom = @import("gloom.zig");const segment = @import("segment.zig");const zstd = @import("zstd.zig");
const Allocator = std.mem.Allocator;
pub const Error = error{ OutOfMemory, CompressFailed };
/// segment-wide header stats; block envelopes live in the index entriespub const Envelope = struct { min_seq: u64 = 0, max_seq: u64 = 0, min_witnessed_at: i64 = 0, max_witnessed_at: i64 = 0, event_count: u64 = 0,};
pub const Builder = struct { /// arena; the caller owns its lifetime (must outlive finish) scratch: Allocator, infos: std.ArrayList(segment.BlockIndexEntry) = .empty, block_dids: std.ArrayList(std.StringHashMapUnmanaged(void)) = .empty, block_collections: std.ArrayList(std.AutoHashMapUnmanaged(u32, void)) = .empty, segment_dids: std.StringHashMapUnmanaged(void) = .empty, // collection table: ids in first-seen order; sentinels interleave table_names: std.ArrayList([]const u8) = .empty, table_counts: std.ArrayList(u32) = .empty, table_ids: std.StringHashMapUnmanaged(u32) = .empty,
pub fn init(scratch: Allocator) Builder { return .{ .scratch = scratch }; }
/// record one block. `info` arrives fully populated (offset, sizes, /// event_count, envelopes); `events` may be empty (a fully-compacted /// block keeps its index slot). retained strings are duped into /// scratch, so events may alias a block buffer freed after this call. pub fn addBlock(self: *Builder, info: segment.BlockIndexEntry, events: []const segment.Event) Error!void { var dids: std.StringHashMapUnmanaged(void) = .empty; var cols: std.AutoHashMapUnmanaged(u32, void) = .empty; for (events) |ev| { if (ev.did.len > 0) { const owned = try dupeIfNew(self.scratch, &self.segment_dids, ev.did); try dids.put(self.scratch, owned, {}); } if (ev.collection.len > 0) { const id = try self.intern(ev.collection, true); try cols.put(self.scratch, id, {}); } if (sentinelFor(ev.kind)) |name| { const id = try self.intern(name, false); try cols.put(self.scratch, id, {}); } } try self.infos.append(self.scratch, info); try self.block_dids.append(self.scratch, dids); try self.block_collections.append(self.scratch, cols); }
/// append the footer to `buf` (whose current length is the footer /// offset) and finalize the header in buf[0..256], including the file /// xxh3. after this, buf holds a sealed jss v1 file. pub fn finish(self: *Builder, allocator: Allocator, buf: *std.ArrayList(u8), env: Envelope) Error!void { const scratch = self.scratch; const footer_offset = buf.items.len;
// === footer section 1: block index === const block_index_offset = footer_offset; for (self.infos.items) |info| { var entry: [segment.BlockIndexEntry.encoded_size]u8 = undefined; std.mem.writeInt(u64, entry[0..8], info.offset, .little); std.mem.writeInt(u32, entry[8..12], info.compressed_size, .little); std.mem.writeInt(u32, entry[12..16], info.uncompressed_size, .little); std.mem.writeInt(u32, entry[16..20], info.event_count, .little); std.mem.writeInt(u64, entry[20..28], info.min_seq, .little); std.mem.writeInt(u64, entry[28..36], info.max_seq, .little); std.mem.writeInt(i64, entry[36..44], info.min_witnessed_at, .little); std.mem.writeInt(i64, entry[44..52], info.max_witnessed_at, .little); try buf.appendSlice(allocator, &entry); }
// === section 2: segment-level DID bloom === const did_bloom_offset = buf.items.len; { var filter = try gloom.Filter.init(scratch, self.segment_dids.count(), 0.001); var keys = self.segment_dids.keyIterator(); while (keys.next()) |k| filter.add(k.*); const blob = try filter.marshal(scratch); try buf.appendSlice(allocator, blob); }
// === section 3: per-block DID blooms === // Sized for the segment's actual max per-block unique-DID // cardinality, not the 4096-event block cap (upstream f02919c): // real blocks hold runs of a few repos (median 1-3 unique DIDs), // and fixed-capacity blooms made the resident region >90% of // server heap. The segment-wide max keeps every bloom identical // in size — the invariant that lets readers index the region by // multiplication — and the region header is self-describing, so // right-sized and legacy fixed-size segments coexist freely. const block_did_bloom_offset = buf.items.len; { var bloom_capacity: usize = 1; for (self.block_dids.items) |*dids| bloom_capacity = @max(bloom_capacity, dids.count()); var region_header: [8]u8 = undefined; std.mem.writeInt(u32, region_header[0..4], @intCast(self.infos.items.len), .little); var bloom_size: u32 = 0; var blobs: std.ArrayList([]const u8) = .empty; for (self.block_dids.items) |*dids| { var filter = try gloom.Filter.init(scratch, bloom_capacity, 0.001); var keys = dids.keyIterator(); while (keys.next()) |k| filter.add(k.*); const blob = try filter.marshal(scratch); bloom_size = @intCast(blob.len); try blobs.append(scratch, blob); } std.mem.writeInt(u32, region_header[4..8], bloom_size, .little); try buf.appendSlice(allocator, ®ion_header); for (blobs.items) |blob| try buf.appendSlice(allocator, blob); }
// === section 4: collection block index === const collection_index_offset = buf.items.len; { const bitmask_len: u32 = @intCast((self.table_names.items.len + 7) / 8); var body: std.ArrayList(u8) = .empty; for (self.table_names.items, self.table_counts.items) |name, count| { try body.append(scratch, @intCast(name.len)); var count_bytes: [4]u8 = undefined; std.mem.writeInt(u32, &count_bytes, count, .little); try body.appendSlice(scratch, &count_bytes); try body.appendSlice(scratch, name); } for (self.block_collections.items) |*cols| { const row_start = body.items.len; try body.appendNTimes(scratch, 0, bitmask_len); var ids = cols.keyIterator(); while (ids.next()) |id| { body.items[row_start + id.* / 8] |= @as(u8, 1) << @intCast(id.* % 8); } } const compressed = zstd.compress(scratch, body.items) catch return error.CompressFailed; var ci_header: [16]u8 = undefined; std.mem.writeInt(u32, ci_header[0..4], @intCast(self.table_names.items.len), .little); std.mem.writeInt(u32, ci_header[4..8], @intCast(self.infos.items.len), .little); std.mem.writeInt(u32, ci_header[8..12], bitmask_len, .little); std.mem.writeInt(u32, ci_header[12..16], @intCast(body.items.len), .little); try buf.appendSlice(allocator, &ci_header); try buf.appendSlice(allocator, compressed); }
// === finalized header + checksum === const h = buf.items[0..segment.header_size]; std.mem.writeInt(u64, h[4..12], 0, .little); // zero while hashing std.mem.writeInt(u16, h[12..14], segment.version, .little); std.mem.writeInt(u32, h[14..18], @intCast(self.infos.items.len), .little); std.mem.writeInt(u32, h[18..22], @intCast(env.event_count), .little); std.mem.writeInt(u32, h[22..26], self.segment_dids.count(), .little); std.mem.writeInt(u64, h[26..34], env.min_seq, .little); std.mem.writeInt(u64, h[34..42], env.max_seq, .little); std.mem.writeInt(i64, h[42..50], env.min_witnessed_at, .little); std.mem.writeInt(i64, h[50..58], env.max_witnessed_at, .little); std.mem.writeInt(u64, h[58..66], footer_offset, .little); std.mem.writeInt(u64, h[66..74], did_bloom_offset, .little); std.mem.writeInt(u64, h[74..82], block_did_bloom_offset, .little); std.mem.writeInt(u64, h[82..90], collection_index_offset, .little); std.mem.writeInt(u64, h[90..98], block_index_offset, .little);
var hasher = std.hash.XxHash3.init(0); hasher.update(buf.items[12..segment.header_size]); hasher.update(buf.items[footer_offset..]); std.mem.writeInt(u64, h[4..12], hasher.final(), .little); }
fn intern(self: *Builder, name: []const u8, count_event: bool) Error!u32 { const gop = try self.table_ids.getOrPut(self.scratch, name); if (!gop.found_existing) { const owned = try self.scratch.dupe(u8, name); gop.key_ptr.* = owned; gop.value_ptr.* = @intCast(self.table_names.items.len); try self.table_names.append(self.scratch, owned); try self.table_counts.append(self.scratch, 0); } if (count_event) self.table_counts.items[gop.value_ptr.*] += 1; return gop.value_ptr.*; }};
fn sentinelFor(kind: segment.Kind) ?[]const u8 { return switch (kind) { .account => "$account", .identity => "$identity", .sync => "$sync", else => null, };}
/// dedupe helper: returns the canonical copy of `key`, inserting if newfn dupeIfNew(allocator: Allocator, set: *std.StringHashMapUnmanaged(void), key: []const u8) ![]const u8 { const gop = try set.getOrPut(allocator, key); if (!gop.found_existing) gop.key_ptr.* = try allocator.dupe(u8, key); return gop.key_ptr.*;}