Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161//! differential test: parse a sealed segment produced by upstream jetstream//! (bootstrapped from the simulator) and check against the ground truth in//! tests/fixtures/seg_0000000000.inspect.txt (values hardcoded below).//!//! regenerate fixtures: build upstream's `jetstream` binary, run the//! simulator, `jetstream serve --relay-url=http://localhost:7777//! --plc-url=http://localhost:7777 --max-backfill-repos=50`, wait for//! steady state, copy <data-dir>/segments/seg_0000000000.jss +//! `jetstream inspect-segment` output. skipped when the fixture is absent.
const std = @import("std");const segment = @import("segment.zig");const zat = @import("zat");
const testing = std.testing;const fixture_path = "tests/fixtures/seg_0000000000.jss";
fn loadFixture(allocator: std.mem.Allocator) !?[]u8 { var threaded: std.Io.Threaded = .init(allocator, .{}); defer threaded.deinit(); const io = threaded.io(); return std.Io.Dir.cwd().readFileAlloc(io, fixture_path, allocator, .limited(4 << 20)) catch null;}
test "fixture: header + checksum match upstream inspect output" { const bytes = (try loadFixture(testing.allocator)) orelse return error.SkipZigTest; defer testing.allocator.free(bytes);
var sealed = try segment.Sealed.parse(testing.allocator, bytes); defer sealed.deinit(testing.allocator);
try testing.expectEqual(@as(u64, 0x356b4de698e8df94), sealed.header.checksum); try testing.expectEqual(@as(u32, 3), sealed.header.block_count); try testing.expectEqual(@as(u32, 7018), sealed.header.event_count); try testing.expectEqual(@as(u32, 100), sealed.header.unique_did_count); try testing.expectEqual(@as(u64, 1), sealed.header.min_seq); try testing.expectEqual(@as(u64, 8575), sealed.header.max_seq); try testing.expectEqual(@as(u64, 0x78199), sealed.header.footer_offset);
// per-block stats from inspect --blocks table const expect_blocks = [_]struct { offset: u64, comp: u32, uncomp: u32, events: u32, min_seq: u64, max_seq: u64 }{ .{ .offset = 0x100, .comp = 203396, .uncomp = 781999, .events = 2799, .min_seq = 1, .max_seq = 4096 }, .{ .offset = 0x31b8c, .comp = 117330, .uncomp = 426425, .events = 1500, .min_seq = 4097, .max_seq = 5596 }, .{ .offset = 0x4e5e6, .comp = 170923, .uncomp = 698431, .events = 2719, .min_seq = 5597, .max_seq = 8575 }, }; for (expect_blocks, sealed.block_index) |want, got| { try testing.expectEqual(want.offset, got.offset); try testing.expectEqual(want.comp, got.compressed_size); try testing.expectEqual(want.uncomp, got.uncompressed_size); try testing.expectEqual(want.events, got.event_count); try testing.expectEqual(want.min_seq, got.min_seq); try testing.expectEqual(want.max_seq, got.max_seq); }}
test "fixture: all blocks decode; events are well-formed and ordered" { const bytes = (try loadFixture(testing.allocator)) orelse return error.SkipZigTest; defer testing.allocator.free(bytes);
var sealed = try segment.Sealed.parse(testing.allocator, bytes); defer sealed.deinit(testing.allocator);
var total: usize = 0; var last_seq: u64 = 0; var commit_payloads: usize = 0; for (0..sealed.block_index.len) |i| { var block = try sealed.readBlock(testing.allocator, i); defer block.deinit(testing.allocator); const entry = sealed.block_index[i]; try testing.expectEqual(@as(usize, entry.event_count), block.events.len); for (block.events) |ev| { try testing.expect(ev.seq > last_seq); // strictly increasing globally last_seq = ev.seq; try testing.expect(ev.seq >= entry.min_seq and ev.seq <= entry.max_seq); try testing.expect(ev.witnessed_at >= entry.min_witnessed_at and ev.witnessed_at <= entry.max_witnessed_at); try testing.expect(ev.did.len > 0); switch (ev.kind) { .create, .update => { try testing.expect(ev.collection.len > 0 and ev.rkey.len > 0); try testing.expect(ev.payload.len > 0); // payload must be intact DAG-CBOR var arena = std.heap.ArenaAllocator.init(testing.allocator); defer arena.deinit(); _ = try zat.cbor.decodeAll(arena.allocator(), ev.payload); commit_payloads += 1; }, .delete => try testing.expectEqual(@as(usize, 0), ev.payload.len), else => {}, } } total += block.events.len; } try testing.expectEqual(@as(usize, 7018), total); try testing.expect(commit_payloads > 0);}
test "fixture: block checksum corruption is rejected" { const bytes = (try loadFixture(testing.allocator)) orelse return error.SkipZigTest; defer testing.allocator.free(bytes);
var sealed = try segment.Sealed.parse(testing.allocator, bytes); const first = sealed.block_index[0]; sealed.deinit(testing.allocator);
// The JSS checksum authenticates header + footer metadata. The zstd frame // checksum independently authenticates the compressed block body, so this // mutation deliberately leaves Sealed.parse valid and must fail only when // the block is decompressed. const checksum_byte = first.offset + 8 + first.compressed_size - 1; bytes[checksum_byte] ^= 0x01;
var corrupted = try segment.Sealed.parse(testing.allocator, bytes); defer corrupted.deinit(testing.allocator); try testing.expectError(error.DecompressFailed, corrupted.readBlock(testing.allocator, 0));}
test "fixture: collection index matches upstream inspect output" { const bytes = (try loadFixture(testing.allocator)) orelse return error.SkipZigTest; defer testing.allocator.free(bytes);
var sealed = try segment.Sealed.parse(testing.allocator, bytes); defer sealed.deinit(testing.allocator); var ci = try sealed.readCollectionIndex(testing.allocator); defer ci.deinit(testing.allocator);
try testing.expectEqual(@as(usize, 7), ci.entries.len); // [id] nsid events (from inspect; ids are table positions) const expected = [_]struct { id: usize, nsid: []const u8, count: u32 }{ .{ .id = 0, .nsid = "app.bsky.actor.profile", .count = 327 }, .{ .id = 1, .nsid = "app.bsky.feed.like", .count = 1288 }, .{ .id = 2, .nsid = "app.bsky.feed.post", .count = 4105 }, .{ .id = 3, .nsid = "app.bsky.feed.repost", .count = 323 }, .{ .id = 4, .nsid = "app.bsky.graph.follow", .count = 724 }, .{ .id = 5, .nsid = "$identity", .count = 0 }, .{ .id = 6, .nsid = "$sync", .count = 0 }, }; for (expected) |want| { try testing.expectEqualStrings(want.nsid, ci.entries[want.id].nsid); try testing.expectEqual(want.count, ci.entries[want.id].count); }
// bitmask cross-check: recount per-collection events by walking blocks var counts: [7]u32 = @splat(0); for (0..sealed.block_index.len) |i| { var block = try sealed.readBlock(testing.allocator, i); defer block.deinit(testing.allocator); for (block.events) |ev| { for (ci.entries, 0..) |entry, id| { if (std.mem.eql(u8, entry.nsid, ev.collection)) { counts[id] += 1; // no false negatives: the block bitmask must claim this id try testing.expect(ci.blockHasCollection(i, id)); } } } } for (expected) |want| { if (want.nsid[0] != '$') try testing.expectEqual(want.count, counts[want.id]); }}