Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493//! Offline operator commands shared by the executable and unit receipts.//!//! `inspect-segment` intentionally reads only local JSS bytes. It follows the//! upstream inspection contract for sealed, checksum-corrupt, and active//! files; it never opens RocksDB or starts the network service.
const std = @import("std");const build_options = @import("build_options");const segment = @import("../storage/segment.zig");const segment_writer = @import("../storage/segment_writer.zig");
const Allocator = std.mem.Allocator;const Io = std.Io;
pub const BlocksMode = enum { summary, table, full };
const Collection = struct { nsid: []const u8, event_count: u32,};
pub const Inspection = struct { arena: std.heap.ArenaAllocator, path: []const u8, file_size: usize, sealed: bool, header: ?segment.Header, checksum_valid: bool = false, blocks: []const segment.BlockIndexEntry, collections: []const Collection, block_collections: []const []const u32, total_events: u64, unique_did_count: u32, min_seq: u64, max_seq: u64, min_witnessed_at: i64, max_witnessed_at: i64, block_index_bytes: u64 = 0, segment_bloom_bytes: u64 = 0, block_blooms_bytes: u64 = 0, collection_index_bytes: u64 = 0, per_block_bloom_bytes: u32 = 0, partial_error: ?anyerror = null,
pub fn deinit(self: *Inspection) void { self.arena.deinit(); }};
pub fn renderVersion(writer: *Io.Writer) !void { try writer.print("jetstream version {s} (commit {s}, built {s})\n", .{ build_options.version, build_options.git_sha, build_options.build_date, });}
pub fn inspectSegment(allocator: Allocator, io: Io, path: []const u8) !Inspection { const bytes = try Io.Dir.cwd().readFileAlloc(io, path, allocator, .limited(1 << 31)); defer allocator.free(bytes); if (bytes.len < segment.header_size) return error.Truncated; if (!std.mem.eql(u8, bytes[0..4], segment.magic)) return error.BadMagic;
var arena = std.heap.ArenaAllocator.init(allocator); errdefer arena.deinit(); const alloc = arena.allocator(); const owned_path = try alloc.dupe(u8, path); const checksum = std.mem.readInt(u64, bytes[4..12], .little); if (checksum == 0) return inspectActive(&arena, owned_path, bytes); return inspectSealed(&arena, owned_path, bytes);}
fn inspectSealed( arena: *std.heap.ArenaAllocator, path: []const u8, bytes: []const u8,) !Inspection { const alloc = arena.allocator(); var reader = try segment.Sealed.parseUnchecked(alloc, bytes); defer reader.deinit(alloc); var collections = try reader.readCollectionIndex(alloc); defer collections.deinit(alloc);
const blocks = try alloc.dupe(segment.BlockIndexEntry, reader.block_index); const collection_rows = try alloc.alloc(Collection, collections.entries.len); for (collections.entries, collection_rows) |entry, *row| { row.* = .{ .nsid = try alloc.dupe(u8, entry.nsid), .event_count = entry.count, }; } const block_collections = try alloc.alloc([]const u32, blocks.len); for (block_collections, 0..) |*ids, block_index| { var found: std.ArrayList(u32) = .empty; for (collections.entries, 0..) |_, collection_id| { if (collections.blockHasCollection(block_index, collection_id)) try found.append(alloc, @intCast(collection_id)); } ids.* = try found.toOwnedSlice(alloc); }
const header = reader.header; const block_index_bytes = header.did_bloom_offset - header.block_index_offset; const segment_bloom_bytes = header.block_did_bloom_offset - header.did_bloom_offset; const block_blooms_bytes = header.collection_index_offset - header.block_did_bloom_offset; const collection_index_bytes = bytes.len - header.collection_index_offset; var per_block_bloom_bytes: u32 = 0; if (header.block_count > 0 and block_blooms_bytes >= 8) per_block_bloom_bytes = @intCast((block_blooms_bytes - 8) / header.block_count);
return .{ .arena = arena.*, .path = path, .file_size = bytes.len, .sealed = true, .header = header, .checksum_valid = segment.checksumValid(bytes, header), .blocks = blocks, .collections = collection_rows, .block_collections = block_collections, .total_events = header.event_count, .unique_did_count = header.unique_did_count, .min_seq = header.min_seq, .max_seq = header.max_seq, .min_witnessed_at = header.min_witnessed_at, .max_witnessed_at = header.max_witnessed_at, .block_index_bytes = block_index_bytes, .segment_bloom_bytes = segment_bloom_bytes, .block_blooms_bytes = block_blooms_bytes, .collection_index_bytes = collection_index_bytes, .per_block_bloom_bytes = per_block_bloom_bytes, };}
fn inspectActive( arena: *std.heap.ArenaAllocator, path: []const u8, bytes: []const u8,) !Inspection { const alloc = arena.allocator(); var blocks: std.ArrayList(segment.BlockIndexEntry) = .empty; var collection_rows: std.ArrayList(Collection) = .empty; var block_collections: std.ArrayList([]const u32) = .empty; var collection_ids: std.StringHashMapUnmanaged(u32) = .empty; var unique_dids: std.StringHashMapUnmanaged(void) = .empty; var total_events: u64 = 0; var min_seq: u64 = 0; var max_seq: u64 = 0; var min_witnessed_at: i64 = 0; var max_witnessed_at: i64 = 0; var saw_any = false; var partial_error: ?anyerror = null;
var it: segment_writer.ActiveIterator = .{ .bytes = bytes }; while (true) { const before = it.offset; var block = (it.next(alloc) catch |err| { partial_error = err; break; }) orelse break; defer block.deinit(alloc); if (block.events.len == 0) break;
var info: segment.BlockIndexEntry = .{ .offset = before, .compressed_size = @intCast(it.offset - before - 8), .uncompressed_size = @intCast(block.buffer.len), .event_count = @intCast(block.events.len), .min_seq = block.events[0].seq, .max_seq = block.events[0].seq, .min_witnessed_at = block.events[0].witnessed_at, .max_witnessed_at = block.events[0].witnessed_at, }; var ids_in_block: std.AutoHashMapUnmanaged(u32, void) = .empty; for (block.events) |event| { info.min_seq = @min(info.min_seq, event.seq); info.max_seq = @max(info.max_seq, event.seq); info.min_witnessed_at = @min(info.min_witnessed_at, event.witnessed_at); info.max_witnessed_at = @max(info.max_witnessed_at, event.witnessed_at); if (!saw_any) { min_seq = event.seq; max_seq = event.seq; min_witnessed_at = event.witnessed_at; max_witnessed_at = event.witnessed_at; saw_any = true; } else { min_seq = @min(min_seq, event.seq); max_seq = @max(max_seq, event.seq); min_witnessed_at = @min(min_witnessed_at, event.witnessed_at); max_witnessed_at = @max(max_witnessed_at, event.witnessed_at); } if (event.did.len > 0 and !unique_dids.contains(event.did)) try unique_dids.put(alloc, try alloc.dupe(u8, event.did), {}); if (event.collection.len > 0) try internCollection(alloc, &collection_rows, &collection_ids, &ids_in_block, event.collection, true); if (sentinelFor(event.kind)) |sentinel| try internCollection(alloc, &collection_rows, &collection_ids, &ids_in_block, sentinel, false); }
const ids = try alloc.alloc(u32, ids_in_block.count()); var id_it = ids_in_block.keyIterator(); var id_at: usize = 0; while (id_it.next()) |id| : (id_at += 1) ids[id_at] = id.*; std.mem.sort(u32, ids, {}, std.sort.asc(u32)); try block_collections.append(alloc, ids); try blocks.append(alloc, info); total_events += block.events.len; }
return .{ .arena = arena.*, .path = path, .file_size = bytes.len, .sealed = false, .header = null, .blocks = try blocks.toOwnedSlice(alloc), .collections = try collection_rows.toOwnedSlice(alloc), .block_collections = try block_collections.toOwnedSlice(alloc), .total_events = total_events, .unique_did_count = @intCast(unique_dids.count()), .min_seq = min_seq, .max_seq = max_seq, .min_witnessed_at = min_witnessed_at, .max_witnessed_at = max_witnessed_at, .partial_error = partial_error, };}
fn sentinelFor(kind: segment.Kind) ?[]const u8 { return switch (kind) { .account => "$account", .identity => "$identity", .sync => "$sync", else => null, };}
fn internCollection( allocator: Allocator, rows: *std.ArrayList(Collection), ids: *std.StringHashMapUnmanaged(u32), ids_in_block: *std.AutoHashMapUnmanaged(u32, void), nsid: []const u8, count_event: bool,) !void { const id = ids.get(nsid) orelse id: { const owned = try allocator.dupe(u8, nsid); const next: u32 = @intCast(rows.items.len); try rows.append(allocator, .{ .nsid = owned, .event_count = 0 }); try ids.put(allocator, owned, next); break :id next; }; if (count_event) rows.items[id].event_count += 1; try ids_in_block.put(allocator, id, {});}
pub fn renderInspection(writer: *Io.Writer, ins: *Inspection, mode: BlocksMode, truncate: usize) !void { try writer.print("file: {s}\n", .{ins.path}); try writer.print("size: {d} bytes\n", .{ins.file_size}); try writer.writeAll(if (ins.sealed) "state: sealed\n" else "state: active (unsealed; block walk)\n"); try writer.writeAll("magic: jss0\n"); if (ins.header) |header| { try writer.print("version: {d}\n", .{segment.version}); try writer.print("checksum: 0x{x:0>16} ({s})\n", .{ header.checksum, if (ins.checksum_valid) "valid" else "invalid" }); } else { try writer.writeAll("version: -\nchecksum: 0x0 (active)\n"); }
if (ins.partial_error) |err| try writer.print("\nWARNING: partial inspection — {s}\n", .{@errorName(err)});
try writer.writeAll("\nheader summary:\n"); if (ins.sealed) { try writer.print(" block_count: {d}\n", .{ins.blocks.len}); try writer.print(" event_count: {d}\n", .{ins.total_events}); try writer.print(" unique_did_count: {d}\n", .{ins.unique_did_count}); } else { try writer.print(" block_count: {d} (discovered via block walk)\n", .{ins.blocks.len}); try writer.print(" event_count: {d} (from walk)\n", .{ins.total_events}); try writer.print(" unique_did_count: {d} (from walk; not durable until seal)\n", .{ins.unique_did_count}); } try writer.print(" seq range: [{d}, {d}]\n", .{ ins.min_seq, ins.max_seq }); var min_buf: [40]u8 = undefined; var max_buf: [40]u8 = undefined; try writer.print(" witnessed_at range: [{s}, {s}]\n", .{ formatMicros(&min_buf, ins.min_witnessed_at), formatMicros(&max_buf, ins.max_witnessed_at), });
if (ins.header) |header| { try writer.writeAll("\nfooter layout (all offsets absolute; block_index_offset is also the footer start):\n"); try writer.print(" block_index_offset: 0x{x:0>16} block_index_size: {d} bytes\n", .{ header.block_index_offset, ins.block_index_bytes }); try writer.print(" did_bloom_offset: 0x{x:0>16} segment_bloom_size: {d} bytes\n", .{ header.did_bloom_offset, ins.segment_bloom_bytes }); try writer.print(" block_did_bloom_offset: 0x{x:0>16} per_block_blooms: {d} x {d} bytes (incl. 8B region header)\n", .{ header.block_did_bloom_offset, ins.blocks.len, ins.per_block_bloom_bytes }); try writer.print(" collection_index_offset: 0x{x:0>16} collection_index_size: {d} bytes\n", .{ header.collection_index_offset, ins.collection_index_bytes }); try writer.print(" end_of_file: 0x{x:0>16}\n", .{ins.file_size}); } else { try writer.writeAll("\nfooter layout: not present (active file)\n"); }
try renderCollections(writer, ins); if (mode == .summary) return; try renderBlocks(writer, ins, mode, truncate);}
fn renderCollections(writer: *Io.Writer, ins: *Inspection) !void { try writer.print("\ncollections ({d} distinct NSIDs):\n", .{ins.collections.len}); if (ins.collections.len == 0) return writer.writeAll(" (none)\n");
const allocator = ins.arena.allocator(); const block_counts = try allocator.alloc(usize, ins.collections.len); @memset(block_counts, 0); for (ins.block_collections) |ids| for (ids) |id| { if (id < block_counts.len) block_counts[id] += 1; }; const order = try allocator.alloc(usize, ins.collections.len); for (order, 0..) |*id, i| id.* = i; std.mem.sort(usize, order, ins.collections, struct { fn lessThan(collections: []const Collection, a: usize, b: usize) bool { const ac = collections[a].event_count; const bc = collections[b].event_count; return ac > bc or (ac == bc and a < b); } }.lessThan);
var nsid_width: usize = 0; var events_width: usize = "events".len; var blocks_width: usize = "blocks".len; for (ins.collections, block_counts) |collection, count| { nsid_width = @max(nsid_width, collection.nsid.len); events_width = @max(events_width, decimalWidth(collection.event_count)); blocks_width = @max(blocks_width, decimalWidth(count)); } for (order) |id| { const collection = ins.collections[id]; try writer.print(" [{d:>3}] {s:<[4]} events: {d:>[5]} blocks: {d:>[6]}\n", .{ id, collection.nsid, collection.event_count, block_counts[id], nsid_width, events_width, blocks_width, }); }}
fn renderBlocks(writer: *Io.Writer, ins: *const Inspection, mode: BlocksMode, truncate: usize) !void { try writer.print("\nblocks ({d} total):\n", .{ins.blocks.len}); try writer.writeAll(" idx offset comp_size uncomp_size events min_seq max_seq min_witnessed_at max_witnessed_at cols\n");
const n = ins.blocks.len; if (truncate == 0 or mode == .full or n <= truncate) { for (0..n) |i| try renderBlock(writer, ins, mode, i); return; } const half = truncate / 2; for (0..half) |i| try renderBlock(writer, ins, mode, i); try writer.print(" ... ({d} rows omitted) ...\n", .{n - 2 * half}); for (n - half..n) |i| try renderBlock(writer, ins, mode, i);}
fn renderBlock(writer: *Io.Writer, ins: *const Inspection, mode: BlocksMode, i: usize) !void { const block = ins.blocks[i]; const ids = if (i < ins.block_collections.len) ins.block_collections[i] else &.{}; var min_buf: [40]u8 = undefined; var max_buf: [40]u8 = undefined; try writer.print(" {d:>3} 0x{x:0>10} {d:>9} {d:>11} {d:>6} {d:>10} {d:>10} {s:>30} {s:>30} {d:>4}\n", .{ i, block.offset, block.compressed_size, block.uncompressed_size, block.event_count, block.min_seq, block.max_seq, formatMicros(&min_buf, block.min_witnessed_at), formatMicros(&max_buf, block.max_witnessed_at), ids.len, }); if (mode == .full and ids.len > 0) { try writer.writeAll(" collections: "); for (ids, 0..) |id, at| { if (id >= ins.collections.len) continue; if (at > 0) try writer.writeAll(", "); try writer.writeAll(ins.collections[id].nsid); } try writer.writeByte('\n'); }}
fn decimalWidth(value: anytype) usize { var n = value; var width: usize = 1; while (n >= 10) : (n /= 10) width += 1; return width;}
fn formatMicros(buffer: []u8, micros: i64) []const u8 { if (micros == 0) return "0"; const secs: u64 = @intCast(@divFloor(micros, std.time.us_per_s)); const fraction: u64 = @intCast(@mod(micros, std.time.us_per_s)); const epoch_seconds: std.time.epoch.EpochSeconds = .{ .secs = secs }; const year_day = epoch_seconds.getEpochDay().calculateYearDay(); const month_day = year_day.calculateMonthDay(); const day_seconds = epoch_seconds.getDaySeconds(); return std.fmt.bufPrint(buffer, "{d:0>4}-{d:0>2}-{d:0>2}T{d:0>2}:{d:0>2}:{d:0>2}.{d:0>6}Z", .{ year_day.year, month_day.month.numeric(), month_day.day_index + 1, day_seconds.getHoursIntoDay(), day_seconds.getMinutesIntoHour(), day_seconds.getSecondsIntoMinute(), fraction, }) catch unreachable;}
test "version command follows upstream text contract" { var out: Io.Writer.Allocating = .init(std.testing.allocator); defer out.deinit(); try renderVersion(&out.writer); try std.testing.expect(std.mem.startsWith(u8, out.written(), "jetstream version ")); try std.testing.expect(std.mem.indexOf(u8, out.written(), " (commit ") != null); try std.testing.expect(std.mem.endsWith(u8, out.written(), ", built unknown)\n"));}
test "sealed fixture inspection matches upstream output byte for byte after path" { var threaded: Io.Threaded = .init(std.testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); const fixture = "tests/fixtures/seg_0000000000.jss"; var inspection = try inspectSegment(std.testing.allocator, io, fixture); defer inspection.deinit(); var out: Io.Writer.Allocating = .init(std.testing.allocator); defer out.deinit(); try renderInspection(&out.writer, &inspection, .table, 100);
const expected = try Io.Dir.cwd().readFileAlloc(io, "tests/fixtures/seg_0000000000.inspect.txt", std.testing.allocator, .limited(1 << 20)); defer std.testing.allocator.free(expected); const expected_body = expected[std.mem.indexOfScalar(u8, expected, '\n').? + 1 ..]; const actual_body = out.written()[std.mem.indexOfScalar(u8, out.written(), '\n').? + 1 ..]; try std.testing.expectEqualStrings(expected_body, actual_body);}
test "active inspection walks real frames and labels torn tails as partial" { var active = try segment_writer.ActiveWriter.init(std.testing.allocator); defer active.deinit(); active.max_events_per_block = 1; try active.append(.{ .seq = 7, .witnessed_at = 1_700_000_000_000_007, .indexed_at = 0, .kind = .identity, .did = "did:plc:test", .collection = "", .rkey = "", .rev = "", .payload = "{}", }); try active.buf.appendSlice(std.testing.allocator, "torn");
var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); try tmp.dir.writeFile(std.testing.io, .{ .sub_path = "active.jss", .data = active.bytes() }); const path = try tmp.dir.realPathFileAlloc(std.testing.io, "active.jss", std.testing.allocator); defer std.testing.allocator.free(path); var inspection = try inspectSegment(std.testing.allocator, std.testing.io, path); defer inspection.deinit(); try std.testing.expect(!inspection.sealed); try std.testing.expectEqual(@as(usize, 1), inspection.blocks.len); try std.testing.expectEqual(@as(u64, 1), inspection.total_events); try std.testing.expectEqual(@as(u32, 1), inspection.unique_did_count); try std.testing.expectEqualStrings("$identity", inspection.collections[0].nsid); try std.testing.expect(inspection.partial_error != null);}
test "inspection reports checksum mismatch without weakening normal parsing" { const fixture = "tests/fixtures/seg_0000000000.jss"; const original = try Io.Dir.cwd().readFileAlloc(std.testing.io, fixture, std.testing.allocator, .limited(1 << 20)); defer std.testing.allocator.free(original); const header = try segment.Header.decode(original); const corrupt_at = header.block_did_bloom_offset + 8 + (header.collection_index_offset - header.block_did_bloom_offset - 8) / 2; original[corrupt_at] ^= 1; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); try tmp.dir.writeFile(std.testing.io, .{ .sub_path = "corrupt.jss", .data = original }); const path = try tmp.dir.realPathFileAlloc(std.testing.io, "corrupt.jss", std.testing.allocator); defer std.testing.allocator.free(path); var inspection = try inspectSegment(std.testing.allocator, std.testing.io, path); defer inspection.deinit(); try std.testing.expect(!inspection.checksum_valid);}