diff --git a/README.md b/README.md index 053337d..5dfdc3d 100644 --- a/README.md +++ b/README.md @@ -61,6 +61,20 @@ just differential-oracle # exact pinned upstream semantics, fully offline just differential-oracle-multiseed # five upstream predicate-kill worlds, offline ``` +The binary also exposes upstream's offline operator commands. They do not +start the server or open RocksDB: + +``` +stream version +stream inspect-segment --blocks=table --blocks-truncate=100 +``` + +`inspect-segment` accepts `summary`, `table`, and `full` detail, handles both +sealed and active files, and reports a readable checksum-corrupt sealed file +as invalid without weakening the checksum enforcement used by archive reads. +The committed upstream-produced JSS fixture pins its rendered output byte for +byte (apart from the file path). + runtime flags are listed in `src/main.zig`; deploy notes in `deploy/README.md`. The Zig build pins and statically compiles its native dependencies, including RocksDB, zstd, xxHash, and websocket.zig's zlib 1.3.2; it does not depend on a diff --git a/build.zig b/build.zig index 77b4669..03f6072 100644 --- a/build.zig +++ b/build.zig @@ -90,6 +90,8 @@ pub fn build(b: *std.Build) void { // build provenance for the stream_build_info canary metric const build_options = b.addOptions(); + build_options.addOption([]const u8, "version", b.option([]const u8, "version", "Release version reported by the version command") orelse "dev"); + build_options.addOption([]const u8, "build_date", b.option([]const u8, "build-date", "RFC3339 build date reported by the version command") orelse "unknown"); build_options.addOption([]const u8, "git_sha", git_sha: { var code: u8 = 0; const result = b.runAllowFail(&.{ "git", "rev-parse", "--short", "HEAD" }, &code, .ignore); diff --git a/docs/configuration-parity.md b/docs/configuration-parity.md index 0a7fe65..617e505 100644 --- a/docs/configuration-parity.md +++ b/docs/configuration-parity.md @@ -28,7 +28,7 @@ the same runtime mechanism and an offline receipt observes that behavior. | public/debug bind addresses, shutdown/client drain timeouts | open | Stream has one port and fixed shutdown behavior | | log level/format and OpenTelemetry | open | canonical metrics exist; upstream process configuration and tracing surface do not | | `JETSTREAM_*` environment sources and unknown-variable rejection | open | Stream currently accepts CLI arguments only | -| serve/version/inspect command surface | open | operational CLI parity is not yet implemented or classified as a deliberate divergence | +| serve/version/inspect command surface | partial | `serve` is accepted explicitly while retaining Stream's historical flag-only invocation. `version` emits the upstream build-information shape, and `inspect-segment` matches the pinned renderer byte-for-byte on an upstream-produced sealed fixture while also walking active files, surfacing partial tails, and labelling readable checksum corruption. Database-wide `inspect-all` remains open. | This table tracks configuration semantics only. It does not override the behavioral gates in `semantic-parity.md` or `bootstrap-semantic-parity.md`. diff --git a/src/internal/operational.zig b/src/internal/operational.zig new file mode 100644 index 0000000..e405562 --- /dev/null +++ b/src/internal/operational.zig @@ -0,0 +1,492 @@ +//! 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("segment.zig"); +const segment_writer = @import("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); +} diff --git a/src/internal/segment.zig b/src/internal/segment.zig index 9a17ea6..d28f0e4 100644 --- a/src/internal/segment.zig +++ b/src/internal/segment.zig @@ -192,14 +192,29 @@ pub const Sealed = struct { /// parse and validate header + block index; verifies the file xxh3 /// (header[12..256] || footer bytes). pub fn parse(allocator: Allocator, bytes: []const u8) Error!Sealed { + return parseInternal(allocator, bytes, true); + } + + /// Parse the sealed structure without rejecting a checksum mismatch. + /// Operator inspection uses this path so it can report an otherwise + /// readable segment as `checksum: ... (invalid)`, matching upstream. + /// Subscriber/archive paths must continue to use `parse`. + pub fn parseUnchecked(allocator: Allocator, bytes: []const u8) Error!Sealed { + return parseInternal(allocator, bytes, false); + } + + fn parseInternal(allocator: Allocator, bytes: []const u8, verify_checksum: bool) Error!Sealed { const header = try Header.decode(bytes); if (header.footer_offset > bytes.len or header.footer_offset < header_size) return error.Truncated; + if (header.block_index_offset != header.footer_offset or + header.did_bloom_offset < header.block_index_offset or + header.block_did_bloom_offset < header.did_bloom_offset or + header.collection_index_offset < header.block_did_bloom_offset or + header.collection_index_offset > bytes.len) + return error.BadFooter; - var hasher = std.hash.XxHash3.init(0); - hasher.update(bytes[12..header_size]); - hasher.update(bytes[header.footer_offset..]); - if (hasher.final() != header.checksum) return error.BadChecksum; + if (verify_checksum and !checksumValid(bytes, header)) return error.BadChecksum; const index_bytes_len = @as(u64, header.block_count) * BlockIndexEntry.encoded_size; if (header.block_index_offset + index_bytes_len > bytes.len) return error.Truncated; @@ -259,6 +274,14 @@ pub const Sealed = struct { } }; +pub fn checksumValid(bytes: []const u8, header: Header) bool { + if (bytes.len < header_size or header.footer_offset > bytes.len) return false; + var hasher = std.hash.XxHash3.init(0); + hasher.update(bytes[12..header_size]); + hasher.update(bytes[header.footer_offset..]); + return hasher.final() == header.checksum; +} + /// Decode the collection footer from a standalone footer slice. The resident /// manifest uses this at startup so planBackfill never has to reopen segment /// data files on its request path. @@ -468,6 +491,30 @@ test "header rejects bad magic, active files, wrong version" { try testing.expectError(error.Truncated, Header.decode(buf[0..100])); } +test "unchecked sealed parse exposes checksum corruption only to inspection" { + const writer = @import("segment_writer.zig"); + 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:test", + .collection = "app.bsky.feed.post", + .rkey = "r", + .rev = "v", + .payload = "p", + }); + try active.seal(); + active.buf.items[active.buf.items.len - 1] ^= 0x01; + + try testing.expectError(error.BadChecksum, Sealed.parse(testing.allocator, active.bytes())); + var inspected = try Sealed.parseUnchecked(testing.allocator, active.bytes()); + defer inspected.deinit(testing.allocator); + try testing.expect(!checksumValid(active.bytes(), inspected.header)); +} + test "empty block decodes to zero events" { const buffer = try testing.allocator.dupe(u8, &[_]u8{ 0, 0, 0, 0 }); var block = try decodeBlock(testing.allocator, buffer); diff --git a/src/main.zig b/src/main.zig index 2336903..d23d489 100644 --- a/src/main.zig +++ b/src/main.zig @@ -27,6 +27,7 @@ const tail_mod = @import("internal/tail.zig"); const timestamp_rules = @import("internal/timestamp/rules.zig"); const timestamp_jobs = @import("internal/timestamp/jobs.zig"); const timestamp_manager = @import("internal/timestamp/manager.zig"); +const operational = @import("internal/operational.zig"); const Io = std.Io; const log = std.log.scoped(.stream); @@ -262,16 +263,90 @@ test "segment cache duration matches Go duration and ceil policy" { try testing.expectEqual(@as(u64, 1_500_000_000), try parseDurationNanoseconds("1.5s")); } +fn writeVersion(io: Io, args: []const []const u8) !void { + if (args.len != 0) return error.BadArgs; + var buffer: [1024]u8 = undefined; + var stdout = Io.File.stdout().writer(io, &buffer); + try operational.renderVersion(&stdout.interface); + try stdout.interface.flush(); +} + +fn runInspectSegment(allocator: std.mem.Allocator, io: Io, args: []const []const u8) !void { + var mode: operational.BlocksMode = .table; + var truncate: usize = 100; + var path: ?[]const u8 = null; + var i: usize = 0; + while (i < args.len) : (i += 1) { + const arg = args[i]; + if (std.mem.startsWith(u8, arg, "--blocks=")) { + mode = parseBlocksMode(arg["--blocks=".len..]) orelse return error.BadArgs; + } else if (std.mem.eql(u8, arg, "--blocks")) { + i += 1; + if (i >= args.len) return error.BadArgs; + mode = parseBlocksMode(args[i]) orelse return error.BadArgs; + } else if (std.mem.startsWith(u8, arg, "--blocks-truncate=")) { + truncate = try parseNonNegativeUsize(arg["--blocks-truncate=".len..]); + } else if (std.mem.eql(u8, arg, "--blocks-truncate")) { + i += 1; + if (i >= args.len) return error.BadArgs; + truncate = try parseNonNegativeUsize(args[i]); + } else if (std.mem.startsWith(u8, arg, "--")) { + return error.BadArgs; + } else if (path != null) { + return error.BadArgs; + } else { + path = arg; + } + } + var inspection = try operational.inspectSegment(allocator, io, path orelse return error.BadArgs); + defer inspection.deinit(); + var buffer: [16 * 1024]u8 = undefined; + var stdout = Io.File.stdout().writer(io, &buffer); + try operational.renderInspection(&stdout.interface, &inspection, mode, truncate); + try stdout.interface.flush(); +} + +fn parseBlocksMode(raw: []const u8) ?operational.BlocksMode { + if (std.mem.eql(u8, raw, "summary")) return .summary; + if (std.mem.eql(u8, raw, "table")) return .table; + if (std.mem.eql(u8, raw, "full")) return .full; + return null; +} + +fn parseNonNegativeUsize(raw: []const u8) !usize { + if (raw.len > 0 and raw[0] == '-') return error.BadArgs; + return std.fmt.parseInt(usize, raw, 10); +} + pub fn main(init: std.process.Init.Minimal) !void { const allocator = std.heap.c_allocator; - shutdown_requested.store(false, .release); - installSignalHandlers(); - var threaded: Io.Threaded = .init(allocator, .{ .stack_size = default_stack_size }); defer threaded.deinit(); const io = threaded.io(); + var arg_it = init.args.iterate(); + _ = arg_it.next(); // program name + var first_arg = arg_it.next(); + if (first_arg) |command| { + if (std.mem.eql(u8, command, "version") or std.mem.eql(u8, command, "inspect-segment")) { + var command_args: std.ArrayList([]const u8) = .empty; + defer command_args.deinit(allocator); + while (arg_it.next()) |arg| try command_args.append(allocator, arg); + if (std.mem.eql(u8, command, "version")) + try writeVersion(io, command_args.items) + else + try runInspectSegment(allocator, io, command_args.items); + return; + } + // Like upstream's explicit `serve` subcommand, while retaining the + // historical flag-only invocation for existing Stream deployments. + if (std.mem.eql(u8, command, "serve")) first_arg = arg_it.next(); + } + + shutdown_requested.store(false, .release); + installSignalHandlers(); + var upstream: []const u8 = "ws://localhost:7777"; var canonical_relay: ?RelayUrls = null; defer if (canonical_relay) |owned| owned.deinit(allocator); @@ -314,9 +389,8 @@ pub fn main(init: std.process.Init.Minimal) !void { var subscribe_read_batch: usize = server_mod.default_subscribe_read_batch; var plan_config: xrpcapi.PlanConfig = .{}; - var arg_it = init.args.iterate(); - _ = arg_it.next(); // program name - while (arg_it.next()) |arg| { + var next_arg = first_arg; + while (next_arg) |arg| : (next_arg = arg_it.next()) { if (std.mem.startsWith(u8, arg, "--upstream=")) { upstream = arg["--upstream=".len..]; } else if (std.mem.startsWith(u8, arg, "--relay-url=")) { diff --git a/src/tests.zig b/src/tests.zig index de3903d..9c0af31 100644 --- a/src/tests.zig +++ b/src/tests.zig @@ -35,6 +35,7 @@ comptime { _ = @import("internal/compact/pass.zig"); _ = @import("internal/compact/steady.zig"); _ = @import("internal/segment_fixture_test.zig"); + _ = @import("internal/operational.zig"); _ = @import("internal/verify.zig"); _ = @import("internal/wire.zig"); _ = @import("internal/convert.zig");