Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473//! Topology-preserving sealed-segment display timestamp patching.//!//! This is the Zig equivalent of upstream `segment.Patch`: rows, blocks,//! witness/sequence envelopes, blooms, and collection indexes are invariant.//! Only dirty blocks are re-encoded, while clean block frames and the complete//! bloom/collection footer tail are copied byte-for-byte. A zero-mutation pass//! returns no replacement bytes and an in-place patch uses tmp + fsync + rename//! + parent-directory fsync.
const std = @import("std");const crashpoint = @import("../runtime/crashpoint.zig");const gloom = @import("gloom.zig");const segment = @import("segment.zig");const writer = @import("segment_writer.zig");const segment_io = @import("segment_io.zig");const zstd = @import("zstd.zig");
const Allocator = std.mem.Allocator;const Io = std.Io;
pub const Result = struct { patched: bool = false, rows_mutated: u64 = 0, blocks_touched: u32 = 0, new_checksum: u64 = 0,};
const Guard = struct { seq: u64, witnessed_at: i64, kind: segment.Kind, did_len: usize, did_hash: u64, collection_len: usize, collection_hash: u64, rkey_len: usize, rkey_hash: u64, rev_len: usize, rev_hash: u64, payload_len: usize, payload_hash: u64,
fn init(ev: segment.Event) Guard { return .{ .seq = ev.seq, .witnessed_at = ev.witnessed_at, .kind = ev.kind, .did_len = ev.did.len, .did_hash = hash(ev.did), .collection_len = ev.collection.len, .collection_hash = hash(ev.collection), .rkey_len = ev.rkey.len, .rkey_hash = hash(ev.rkey), .rev_len = ev.rev.len, .rev_hash = hash(ev.rev), .payload_len = ev.payload.len, .payload_hash = hash(ev.payload), }; }
fn verify(self: Guard, ev: segment.Event) !void { if (ev.seq != self.seq) return error.PatchChangedImmutableField; if (ev.witnessed_at != self.witnessed_at) return error.PatchChangedImmutableField; if (ev.kind != self.kind) return error.PatchChangedImmutableField; if (ev.did.len != self.did_len or hash(ev.did) != self.did_hash) return error.PatchChangedImmutableField; if (ev.collection.len != self.collection_len or hash(ev.collection) != self.collection_hash) return error.PatchChangedImmutableField; if (ev.rkey.len != self.rkey_len or hash(ev.rkey) != self.rkey_hash) return error.PatchChangedImmutableField; if (ev.rev.len != self.rev_len or hash(ev.rev) != self.rev_hash) return error.PatchChangedImmutableField; if (ev.payload.len != self.payload_len or hash(ev.payload) != self.payload_hash) return error.PatchChangedImmutableField; }};
fn hash(bytes: []const u8) u64 { return std.hash.XxHash3.hash(0, bytes);}
/// Return replacement sealed bytes when at least one indexed_at value changes./// `mutate` may change only `indexed_at`; the post-callback guard checks every/// row after every callback, including cross-row retained-pointer mutations.pub fn patch( allocator: Allocator, old_bytes: []const u8, result: *Result, candidate_dids: []const []const u8, ctx: anytype, comptime mutate: fn (@TypeOf(ctx), *segment.Event) void,) !?[]u8 { result.* = .{}; var sealed = try segment.Sealed.parse(allocator, old_bytes); defer sealed.deinit(allocator); const header = sealed.header;
const block_index_len = @as(usize, header.block_count) * segment.BlockIndexEntry.encoded_size; if (header.did_bloom_offset != header.footer_offset + block_index_len) return error.NonContiguousFooter; if (header.did_bloom_offset > header.block_did_bloom_offset or header.block_did_bloom_offset > header.collection_index_offset or header.collection_index_offset > old_bytes.len) return error.BadFooter;
if (candidate_dids.len > 0) { var filter = try gloom.unmarshal(allocator, old_bytes[header.did_bloom_offset..header.block_did_bloom_offset]); defer filter.deinit(allocator); var possible = false; for (candidate_dids) |did| { if (did.len > 0 and filter.contains(did)) { possible = true; break; } } if (!possible) return null; }
var out: std.ArrayList(u8) = .empty; defer out.deinit(allocator); try out.appendSlice(allocator, old_bytes[0..segment.header_size]);
var infos = try allocator.alloc(segment.BlockIndexEntry, sealed.block_index.len); defer allocator.free(infos);
for (sealed.block_index, 0..) |old_info, block_index| { var block = try sealed.readBlock(allocator, block_index); defer block.deinit(allocator); if (block.events.len != old_info.event_count) return error.BlockEventCountMismatch;
const guards = try allocator.alloc(Guard, block.events.len); defer allocator.free(guards); const old_indexed = try allocator.alloc(i64, block.events.len); defer allocator.free(old_indexed); for (block.events, guards, old_indexed) |ev, *guard, *indexed| { guard.* = .init(ev); indexed.* = ev.indexed_at; } for (block.events) |*ev| mutate(ctx, ev);
var dirty = false; for (block.events, guards, old_indexed) |ev, guard, indexed| { try guard.verify(ev); if (ev.indexed_at != indexed) { dirty = true; result.rows_mutated += 1; } }
var info = old_info; info.offset = out.items.len; if (dirty) { const encoded = try writer.encodeBlock(allocator, block.events); defer allocator.free(encoded); if (encoded.len != old_info.uncompressed_size) return error.PatchChangedBlockSize; const frame = zstd.compress(allocator, encoded) catch |err| switch (err) { error.OutOfMemory => return error.OutOfMemory, else => return error.CompressFailed, }; defer allocator.free(frame); var prefix: [8]u8 = undefined; std.mem.writeInt(u64, &prefix, frame.len, .little); try out.appendSlice(allocator, &prefix); try out.appendSlice(allocator, frame); info.compressed_size = @intCast(frame.len); result.blocks_touched += 1; } else { const frame_end = old_info.offset + 8 + old_info.compressed_size; try out.appendSlice(allocator, old_bytes[old_info.offset..frame_end]); } infos[block_index] = info; }
if (result.rows_mutated == 0) return null;
const new_footer_offset = out.items.len; for (infos) |info| try appendBlockIndex(allocator, &out, info); // DID bloom, per-block blooms, and collection index do not describe // indexed_at and therefore remain byte-identical. try out.appendSlice(allocator, old_bytes[header.did_bloom_offset..]);
const did_bloom_len = header.block_did_bloom_offset - header.did_bloom_offset; const block_bloom_len = header.collection_index_offset - header.block_did_bloom_offset; const h = out.items[0..segment.header_size]; std.mem.writeInt(u64, h[4..12], 0, .little); std.mem.writeInt(u64, h[58..66], new_footer_offset, .little); std.mem.writeInt(u64, h[66..74], new_footer_offset + block_index_len, .little); std.mem.writeInt(u64, h[74..82], new_footer_offset + block_index_len + did_bloom_len, .little); std.mem.writeInt(u64, h[82..90], new_footer_offset + block_index_len + did_bloom_len + block_bloom_len, .little); std.mem.writeInt(u64, h[90..98], new_footer_offset, .little); var hasher = std.hash.XxHash3.init(0); hasher.update(h[12..segment.header_size]); hasher.update(out.items[new_footer_offset..]); result.new_checksum = hasher.final(); std.mem.writeInt(u64, h[4..12], result.new_checksum, .little); result.patched = true; return try out.toOwnedSlice(allocator);}
fn appendBlockIndex(allocator: Allocator, out: *std.ArrayList(u8), info: segment.BlockIndexEntry) !void { 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 out.appendSlice(allocator, &entry);}
pub fn patchFile( allocator: Allocator, io: Io, dir: Io.Dir, name: []const u8, candidate_dids: []const []const u8, ctx: anytype, comptime mutate: fn (@TypeOf(ctx), *segment.Event) void,) !Result { const old_bytes = try dir.readFileAlloc(io, name, allocator, .limited(1 << 31)); defer allocator.free(old_bytes); var result: Result = .{}; const new_bytes = (try patch(allocator, old_bytes, &result, candidate_dids, ctx, mutate)) orelse return result; defer allocator.free(new_bytes);
var tmp_buf: [128]u8 = undefined; const tmp_name = try std.fmt.bufPrint(&tmp_buf, "{s}.tmp", .{name}); dir.deleteFile(io, tmp_name) catch |err| if (err != error.FileNotFound) return err; var file = try dir.createFile(io, tmp_name, .{ .truncate = true }); var file_open = true; errdefer { if (file_open) file.close(io); dir.deleteFile(io, tmp_name) catch {}; } try segment_io.writeStreamingAll(&file, io, new_bytes); crashpoint.hit(.after_segment_patch_temp_written); try segment_io.sync(&file, io); crashpoint.hit(.after_segment_patch_temp_synced); file.close(io); file_open = false; try segment_io.rename(dir, tmp_name, dir, name, io); crashpoint.hit(.after_segment_patch_renamed); try segment_io.syncDir(dir, io); crashpoint.hit(.after_segment_patch_dir_synced); return result;}
// === offline production-format tests ===
const testing = std.testing;
fn fixtureEvent(seq: u64, did: []const u8, collection: []const u8) segment.Event { return .{ .seq = seq, .witnessed_at = @intCast(1000 + seq), .indexed_at = 0, .kind = if (collection.len == 0) .identity else .create, .did = did, .collection = collection, .rkey = if (collection.len == 0) "" else "r1", .rev = if (collection.len == 0) "" else "v1", .payload = if (collection.len == 0) "" else "\xa1\x61\x78\x01", };}
fn fixture(allocator: Allocator) ![]u8 { var active = try writer.ActiveWriter.init(allocator); defer active.deinit(); active.max_events_per_block = 2; try active.append(fixtureEvent(1, "did:plc:a", "app.bsky.feed.post")); try active.append(fixtureEvent(2, "did:plc:b", "app.bsky.feed.like")); try active.append(fixtureEvent(3, "did:plc:a", "")); try active.append(fixtureEvent(4, "did:plc:c", "app.bsky.feed.post")); try active.append(fixtureEvent(5, "did:plc:b", "app.bsky.feed.post")); try active.seal(); return allocator.dupe(u8, active.bytes());}
const Stamp = struct { did: []const u8, value: i64 };fn stamp(ctx: Stamp, ev: *segment.Event) void { if (std.mem.eql(u8, ev.did, ctx.did)) ev.indexed_at = ctx.value;}
test "patch changes only indexed_at and preserves clean frames plus footer tail" { const old = try fixture(testing.allocator); defer testing.allocator.free(old); var before = try segment.Sealed.parse(testing.allocator, old); defer before.deinit(testing.allocator); var result: Result = .{}; const updated = (try patch(testing.allocator, old, &result, &.{}, Stamp{ .did = "did:plc:a", .value = 99 }, stamp)).?; defer testing.allocator.free(updated); try testing.expect(result.patched); try testing.expectEqual(@as(u64, 2), result.rows_mutated); try testing.expectEqual(@as(u32, 2), result.blocks_touched);
var after = try segment.Sealed.parse(testing.allocator, updated); defer after.deinit(testing.allocator); try testing.expectEqual(before.header.block_count, after.header.block_count); try testing.expectEqual(before.header.event_count, after.header.event_count); try testing.expectEqual(before.header.min_seq, after.header.min_seq); try testing.expectEqual(before.header.max_seq, after.header.max_seq); try testing.expectEqual(before.header.min_witnessed_at, after.header.min_witnessed_at); try testing.expectEqual(before.header.max_witnessed_at, after.header.max_witnessed_at); try testing.expectEqualSlices(u8, old[before.header.did_bloom_offset..], updated[after.header.did_bloom_offset..]);
// The final block has no did:plc:a and is copied exactly, including its // original zstd frame bytes. const old_last = before.block_index[2]; const new_last = after.block_index[2]; try testing.expectEqualSlices( u8, old[old_last.offset .. old_last.offset + 8 + old_last.compressed_size], updated[new_last.offset .. new_last.offset + 8 + new_last.compressed_size], ); for (0..after.block_index.len) |bi| { var block = try after.readBlock(testing.allocator, bi); defer block.deinit(testing.allocator); for (block.events) |ev| { try testing.expectEqual(if (std.mem.eql(u8, ev.did, "did:plc:a")) @as(i64, 99) else 0, ev.indexed_at); } }}
test "patch file exhaustively propagates segment I/O faults without silent corruption" { segment_io.disarm(); defer segment_io.disarm(); const old = try fixture(testing.allocator); defer testing.allocator.free(old);
// Learn the deterministic committed output from the real implementation. var reference_tmp = testing.tmpDir(.{}); defer reference_tmp.cleanup(); { var file = try reference_tmp.dir.createFile(testing.io, "seg.jss", .{}); defer file.close(testing.io); try file.writeStreamingAll(testing.io, old); } _ = try patchFile(testing.allocator, testing.io, reference_tmp.dir, "seg.jss", &.{}, Stamp{ .did = "did:plc:a", .value = 99 }, stamp); const committed = try reference_tmp.dir.readFileAlloc(testing.io, "seg.jss", testing.allocator, .limited(1 << 20)); defer testing.allocator.free(committed); try testing.expect(!std.mem.eql(u8, old, committed));
const cases = [_]struct { spec: []const u8, committed_after_error: bool }{ .{ .spec = "write:1:enospc", .committed_after_error = false }, .{ .spec = "write:1:shortwrite", .committed_after_error = false }, .{ .spec = "sync:1:eio", .committed_after_error = false }, .{ .spec = "rename:1:eio", .committed_after_error = false }, .{ .spec = "sync:2:eio", .committed_after_error = true }, }; for (cases) |case| { var tmp = testing.tmpDir(.{}); defer tmp.cleanup(); { var file = try tmp.dir.createFile(testing.io, "seg.jss", .{}); defer file.close(testing.io); try file.writeStreamingAll(testing.io, old); } try testing.expect(segment_io.armSpec(case.spec)); try testing.expectError(switch (case.spec[0]) { 'w' => if (std.mem.endsWith(u8, case.spec, "enospc")) error.NoSpaceLeft else error.ShortWrite, else => error.InputOutput, }, patchFile(testing.allocator, testing.io, tmp.dir, "seg.jss", &.{}, Stamp{ .did = "did:plc:a", .value = 99 }, stamp)); segment_io.disarm(); try testing.expectError(error.FileNotFound, tmp.dir.openFile(testing.io, "seg.jss.tmp", .{})); const got = try tmp.dir.readFileAlloc(testing.io, "seg.jss", testing.allocator, .limited(1 << 20)); defer testing.allocator.free(got); try testing.expectEqualSlices(u8, if (case.committed_after_error) committed else old, got); }}
test "patch is idempotent and candidate bloom skip never calls mutate" { const old = try fixture(testing.allocator); defer testing.allocator.free(old); var first: Result = .{}; const updated = (try patch(testing.allocator, old, &first, &.{}, Stamp{ .did = "did:plc:a", .value = 99 }, stamp)).?; defer testing.allocator.free(updated); var second: Result = .{}; try testing.expectEqual(@as(?[]u8, null), try patch(testing.allocator, updated, &second, &.{}, Stamp{ .did = "did:plc:a", .value = 99 }, stamp)); try testing.expectEqual(@as(u64, 0), second.rows_mutated);
const Never = struct { fn run(_: void, _: *segment.Event) void { @panic("candidate bloom false-negative"); } }; var skipped: Result = .{}; try testing.expectEqual(@as(?[]u8, null), try patch(testing.allocator, old, &skipped, &.{"did:plc:not-present"}, {}, Never.run));}
test "patch rejects immutable and cross-row mutation" { const old = try fixture(testing.allocator); defer testing.allocator.free(old); const Evil = struct { fn run(_: void, ev: *segment.Event) void { ev.seq += 1; } }; var result: Result = .{}; try testing.expectError(error.PatchChangedImmutableField, patch(testing.allocator, old, &result, &.{}, {}, Evil.run));
const Cross = struct { first: ?*segment.Event = null, fn run(self: *@This(), ev: *segment.Event) void { if (self.first) |first| { first.seq += 1; } else self.first = ev; } }; var cross: Cross = .{}; result = .{}; try testing.expectError(error.PatchChangedImmutableField, patch(testing.allocator, old, &result, &.{}, &cross, Cross.run));}
test "patchFile atomically installs a checksummed segment" { var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); var tmp = testing.tmpDir(.{}); defer tmp.cleanup(); var dir = try tmp.dir.createDirPathOpen(io, "segments", .{ .open_options = .{ .iterate = true } }); defer dir.close(io); const old = try fixture(testing.allocator); defer testing.allocator.free(old); var file = try dir.createFile(io, "seg_0000000000.jss", .{}); try file.writeStreamingAll(io, old); file.close(io);
const result = try patchFile(testing.allocator, io, dir, "seg_0000000000.jss", &.{}, Stamp{ .did = "did:plc:a", .value = 99 }, stamp); try testing.expect(result.patched); const bytes = try dir.readFileAlloc(io, "seg_0000000000.jss", testing.allocator, .limited(1 << 20)); defer testing.allocator.free(bytes); var sealed = try segment.Sealed.parse(testing.allocator, bytes); defer sealed.deinit(testing.allocator); try testing.expectEqual(result.new_checksum, sealed.header.checksum); try testing.expectError(error.FileNotFound, dir.access(io, "seg_0000000000.jss.tmp", .{}));}
test "patch accepts upstream-produced sealed JSS and preserves its opaque indexes" { var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); const old = try Io.Dir.cwd().readFileAlloc( io, "tests/fixtures/seg_0000000000.jss", testing.allocator, .limited(4 << 20), ); defer testing.allocator.free(old); var before = try segment.Sealed.parse(testing.allocator, old); defer before.deinit(testing.allocator);
const One = struct { seq: u64, value: i64, fn run(self: @This(), ev: *segment.Event) void { if (ev.seq == self.seq) ev.indexed_at = self.value; } }; var first_block = try before.readBlock(testing.allocator, 0); defer first_block.deinit(testing.allocator); const target = One{ .seq = first_block.events[0].seq, .value = first_block.events[0].witnessed_at - 1234 }; var result: Result = .{}; const updated = (try patch(testing.allocator, old, &result, &.{first_block.events[0].did}, target, One.run)).?; defer testing.allocator.free(updated); try testing.expectEqual(@as(u64, 1), result.rows_mutated); try testing.expectEqual(@as(u32, 1), result.blocks_touched);
var after = try segment.Sealed.parse(testing.allocator, updated); defer after.deinit(testing.allocator); try testing.expectEqualSlices(u8, old[before.header.did_bloom_offset..], updated[after.header.did_bloom_offset..]); var patched_block = try after.readBlock(testing.allocator, 0); defer patched_block.deinit(testing.allocator); try testing.expectEqual(target.value, patched_block.events[0].indexed_at); try testing.expectEqual(target.seq, patched_block.events[0].seq);}