Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589//! sealed-segment rewrite — the physical side of compaction//!//! upstream segment/rewrite.go: read a sealed jss file, apply a keep//! predicate per event, and produce a new sealed file. block topology is//! PRESERVED — a fully-dropped block remains as an event_count=0 block,//! and every block keeps its HISTORICAL seq/witnessed-at envelope (a//! superset of the surviving rows), so block numbering, cursor//! translation, and seq-bound pruning stay stable across compaction//! generations. rewritten files get a new checksum.//!//! crash safety: write sibling `<name>.tmp` → fsync → rename over the//! original → fsync the parent dir. failure at or before the rename//! leaves the original untouched; stale tmps are reclaimed at boot and//! pass start (cleanStaleTmp). when no row drops the file is untouched//! ("clean" — rewritten=false).
const std = @import("std");const crashpoint = @import("../runtime/crashpoint.zig");const footer_mod = @import("segment_footer.zig");const gloom = @import("gloom.zig");const segment = @import("segment.zig");const writer_mod = @import("segment_writer.zig");const segment_io = @import("segment_io.zig");const zstd = @import("zstd.zig");
const Io = std.Io;const Allocator = std.mem.Allocator;const log = std.log.scoped(.stream);
pub const Result = struct { rewritten: bool = false, kept: u64 = 0, dropped: u64 = 0, bytes_rewritten: u64 = 0, /// Blocks that actually lost a row and so had to be re-encoded and /// recompressed. Blocks that dropped nothing reuse their source frame, so /// this stays far below block_count on a typical compaction rewrite. blocks_recompressed: u64 = 0,};
pub const Options = struct { /// null means full scan; a non-null slice enables the segment-level /// candidate-DID bloom prefilter used by upstream compaction. candidate_dids: ?[]const []const u8 = null,};
/// rewrite sealed bytes in memory. returns the new sealed file bytes/// (caller frees) when at least one row drops, null when the segment is/// clean. `keep(ctx, ev)` returns true to retain the row.pub fn rewrite( allocator: Allocator, old_bytes: []const u8, result: *Result, opts: Options, ctx: anytype, comptime keep: fn (@TypeOf(ctx), segment.Event) bool,) !?[]u8 { var sealed = try segment.Sealed.parse(allocator, old_bytes); defer sealed.deinit(allocator);
if (opts.candidate_dids) |candidates| { if (sealed.header.block_did_bloom_offset < sealed.header.did_bloom_offset) return error.BadFooter; var filter = try gloom.unmarshal(allocator, old_bytes[sealed.header.did_bloom_offset..sealed.header.block_did_bloom_offset]); defer filter.deinit(allocator); var possible = false; for (candidates) |did| { if (filter.contains(did)) { possible = true; break; } } if (!possible) return null; }
var arena = std.heap.ArenaAllocator.init(allocator); defer arena.deinit(); const scratch = arena.allocator();
var out: std.ArrayList(u8) = .empty; defer out.deinit(allocator); try out.appendSlice(allocator, segment.magic); try out.appendNTimes(allocator, 0, segment.header_size - segment.magic.len);
// the footer builder sizes per-block blooms from the surviving rows' // actual unique-DID cardinality (tighter than the source segment's // event-count bound it previously carried forward) var builder = footer_mod.Builder.init(scratch);
var kept_events: std.ArrayList(segment.Event) = .empty; defer kept_events.deinit(allocator);
for (sealed.block_index, 0..) |old_entry, i| { var block = try sealed.readBlock(allocator, i); defer block.deinit(allocator);
kept_events.clearRetainingCapacity(); var dropped_in_block = false; for (block.events) |ev| { if (keep(ctx, ev)) { try kept_events.append(allocator, ev); } else { result.dropped += 1; dropped_in_block = true; } } result.kept += kept_events.items.len;
// A block that lost nothing re-encodes to the same rows, so reuse its // original compressed frame verbatim rather than paying zstd to // reproduce bytes we already have (upstream segment/rewrite.go does the // same). Compaction rewrites the whole archive to apply the tombstones // of a much smaller window, so most blocks in most segments drop // nothing; recompressing them unconditionally was the bulk of the work, // and on a clean segment every byte of it is discarded below. var encoded: ?[]u8 = null; defer if (encoded) |e| allocator.free(e); var compressed: ?[]u8 = null; defer if (compressed) |c| allocator.free(c);
var frame: []const u8 = undefined; var uncompressed_size: u32 = old_entry.uncompressed_size; if (dropped_in_block) { result.blocks_recompressed += 1; const e = try writer_mod.encodeBlock(allocator, kept_events.items); encoded = e; const c = zstd.compress(allocator, e) catch |err| switch (err) { error.OutOfMemory => return error.OutOfMemory, else => return error.CompressFailed, }; compressed = c; frame = c; uncompressed_size = @intCast(e.len); } else { frame = old_bytes[old_entry.offset + 8 ..][0..old_entry.compressed_size]; }
var info = old_entry; // historical envelopes preserved (supersets) info.offset = out.items.len; info.compressed_size = @intCast(frame.len); info.uncompressed_size = uncompressed_size; info.event_count = @intCast(kept_events.items.len); try builder.addBlock(info, kept_events.items);
var len_prefix: [8]u8 = undefined; std.mem.writeInt(u64, &len_prefix, frame.len, .little); try out.appendSlice(allocator, &len_prefix); try out.appendSlice(allocator, frame); }
if (result.dropped == 0) return null; result.rewritten = true;
// header envelopes stay historical too; only the event count shrinks try builder.finish(allocator, &out, .{ .min_seq = sealed.header.min_seq, .max_seq = sealed.header.max_seq, .min_witnessed_at = sealed.header.min_witnessed_at, .max_witnessed_at = sealed.header.max_witnessed_at, .event_count = result.kept, }); const owned = try out.toOwnedSlice(allocator); result.bytes_rewritten = owned.len; return owned;}
/// rewrite one sealed segment file in place (tmp + fsync + rename +/// dir fsync). clean segments are untouched. errors are the caller's to/// surface loudly — a failed rewrite must fail the pass, never advance/// the watermark.pub fn rewriteFile( allocator: Allocator, io: Io, dir: Io.Dir, name: []const u8, opts: Options, ctx: anytype, comptime keep: fn (@TypeOf(ctx), segment.Event) bool,) !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 rewrite(allocator, old_bytes, &result, opts, ctx, keep)) orelse return result; defer allocator.free(new_bytes);
try swapFile(io, dir, name, new_bytes); return result;}
/// crash-safe replacement of a sealed file: sibling tmp → fsync → rename/// over the original → fsync the parent dir. failure at or before the/// rename leaves the original untouched.fn swapFile(io: Io, dir: Io.Dir, name: []const u8, new_bytes: []const u8) !void { var tmp_buf: [128]u8 = undefined; const tmp_name = try std.fmt.bufPrint(&tmp_buf, "{s}.tmp", .{name}); var f = try dir.createFile(io, tmp_name, .{ .truncate = true }); var file_open = true; errdefer { if (file_open) f.close(io); dir.deleteFile(io, tmp_name) catch {}; } try segment_io.writeStreamingAll(&f, io, new_bytes); crashpoint.hit(.after_segment_rewrite_temp_written); try segment_io.sync(&f, io); crashpoint.hit(.after_segment_rewrite_temp_synced); f.close(io); file_open = false; try segment_io.rename(dir, tmp_name, dir, name, io); crashpoint.hit(.after_segment_rewrite_renamed); // make the rename durable (portable directory fsync; see segment_io.syncDir) try segment_io.syncDir(dir, io); crashpoint.hit(.after_segment_rewrite_dir_synced);}
/// Rebuild only the footer with right-sized per-block DID blooms,/// reusing the original compressed frames verbatim (the data region is/// byte-identical; blocks are decompressed solely to read their DID/// sets). Returns null when the segment's bloom region is already at/// the right-sized width. Used by the legacy-archive rebloom sweep:/// segments sealed before the cardinality sizing fix carry blooms sized/// for the 4096-event block cap, ~1000x over the measured per-block/// unique-DID median, and resident bloom regions dominate server heap.pub fn rebloom(allocator: Allocator, old_bytes: []const u8, result: *Result) !?[]u8 { var sealed = try segment.Sealed.parse(allocator, old_bytes); defer sealed.deinit(allocator);
var arena = std.heap.ArenaAllocator.init(allocator); defer arena.deinit(); const scratch = arena.allocator();
var builder = footer_mod.Builder.init(scratch); for (sealed.block_index, 0..) |old_entry, i| { var block = try sealed.readBlock(allocator, i); defer block.deinit(allocator); result.kept += block.events.len; // offsets/sizes unchanged: the data region is copied verbatim try builder.addBlock(old_entry, block.events); }
var out: std.ArrayList(u8) = .empty; defer out.deinit(allocator); // header placeholder + the original data region, byte for byte try out.appendSlice(allocator, old_bytes[0..sealed.header.block_index_offset]); @memset(out.items[segment.magic.len..segment.header_size], 0);
try builder.finish(allocator, &out, .{ .min_seq = sealed.header.min_seq, .max_seq = sealed.header.max_seq, .min_witnessed_at = sealed.header.min_witnessed_at, .max_witnessed_at = sealed.header.max_witnessed_at, .event_count = sealed.header.event_count, });
if (out.items.len >= old_bytes.len) return null; // already right-sized result.rewritten = true; const owned = try out.toOwnedSlice(allocator); result.bytes_rewritten = owned.len; return owned;}
/// rebloom one sealed segment file in place with rewriteFile's exact/// crash-safety dance. Returns rewritten=false when already right-sized.pub fn rebloomFile(allocator: Allocator, io: Io, dir: Io.Dir, name: []const u8) !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 rebloom(allocator, old_bytes, &result)) orelse return result; defer allocator.free(new_bytes);
try swapFile(io, dir, name, new_bytes); return result;}
/// reclaim tmps left by a crashed rewrite (boot and pass start; the/// single-compactor invariant means any tmp seen here is stale)pub fn cleanStaleTmp(allocator: Allocator, io: Io, dir: Io.Dir) !void { var stale: std.ArrayList([]const u8) = .empty; defer { for (stale.items) |n| allocator.free(n); stale.deinit(allocator); } var it = dir.iterate(); while (try it.next(io)) |entry| { if (std.mem.endsWith(u8, entry.name, ".jss.tmp")) try stale.append(allocator, try allocator.dupe(u8, entry.name)); } for (stale.items) |n| { dir.deleteFile(io, n) catch |err| { log.warn("compaction: could not remove stale {s}: {s}", .{ n, @errorName(err) }); continue; }; log.info("compaction: reclaimed stale {s}", .{n}); }}
// === tests ===
const testing = std.testing;
fn sampleEvent(seq: u64, kind: segment.Kind, rkey: []const u8) segment.Event { return .{ .seq = seq, .witnessed_at = @intCast(1_700_000_000_000_000 + seq), .indexed_at = 0, .kind = kind, .did = "did:plc:rewritetestaccount", .collection = "app.bsky.feed.post", .rkey = rkey, .rev = "3l3qo2vutsw2b", .payload = "\xa1\x64\x74\x65\x78\x74\x62\x68\x69", };}
fn buildSealed(allocator: Allocator) ![]u8 { var w = try writer_mod.ActiveWriter.init(allocator); defer w.deinit(); w.max_events_per_block = 10; var seq: u64 = 1; while (seq <= 30) : (seq += 1) try w.append(sampleEvent(seq, .create, "3l3qo2vuowo2b")); try w.seal(); return allocator.dupe(u8, w.bytes());}
const DropBelow = struct { cutoff: u64 };fn keepAbove(ctx: DropBelow, ev: segment.Event) bool { return ev.seq >= ctx.cutoff;}
fn unreachableKeep(_: void, _: segment.Event) bool { @panic("segment bloom false negative: predicate must not run");}
test "rewrite: clean segment returns null, file layout preserved on drops" { const old = try buildSealed(testing.allocator); defer testing.allocator.free(old);
// clean: nothing drops var clean_res: Result = .{}; try testing.expectEqual(@as(?[]u8, null), try rewrite(testing.allocator, old, &clean_res, .{}, DropBelow{ .cutoff = 0 }, keepAbove)); try testing.expect(!clean_res.rewritten); try testing.expectEqual(@as(u64, 30), clean_res.kept);
// drop the whole first block (seqs 1..10) plus half the second var res: Result = .{}; const new_bytes = (try rewrite(testing.allocator, old, &res, .{}, DropBelow{ .cutoff = 16 }, keepAbove)).?; defer testing.allocator.free(new_bytes); try testing.expect(res.rewritten); try testing.expectEqual(@as(u64, 15), res.dropped); try testing.expectEqual(@as(u64, 15), res.kept);
var sealed = try segment.Sealed.parse(testing.allocator, new_bytes); defer sealed.deinit(testing.allocator); // block topology preserved: 3 blocks, first now empty try testing.expectEqual(@as(u32, 3), sealed.header.block_count); try testing.expectEqual(@as(u32, 15), sealed.header.event_count); // historical envelopes survive as supersets try testing.expectEqual(@as(u64, 1), sealed.header.min_seq); try testing.expectEqual(@as(u64, 30), sealed.header.max_seq); try testing.expectEqual(@as(u32, 0), sealed.block_index[0].event_count); try testing.expectEqual(@as(u64, 1), sealed.block_index[0].min_seq); try testing.expectEqual(@as(u64, 10), sealed.block_index[0].max_seq); try testing.expectEqual(@as(u32, 5), sealed.block_index[1].event_count); try testing.expectEqual(@as(u64, 11), sealed.block_index[1].min_seq);
var total: usize = 0; var expect_seq: u64 = 16; for (0..sealed.block_index.len) |i| { var block = try sealed.readBlock(testing.allocator, i); defer block.deinit(testing.allocator); for (block.events) |ev| { try testing.expectEqual(expect_seq, ev.seq); expect_seq += 1; } total += block.events.len; } try testing.expectEqual(@as(usize, 15), total);}
test "rewrite: blocks that drop nothing reuse their compressed frame verbatim" { const old = try buildSealed(testing.allocator); defer testing.allocator.free(old); var source = try segment.Sealed.parse(testing.allocator, old); defer source.deinit(testing.allocator);
// drop one row from block 0 only; blocks 1 and 2 are untouched var res: Result = .{}; const new_bytes = (try rewrite(testing.allocator, old, &res, .{}, DropBelow{ .cutoff = 2 }, keepAbove)).?; defer testing.allocator.free(new_bytes); try testing.expectEqual(@as(u64, 1), res.dropped);
// The load-bearing assertion. zstd is deterministic, so recompressing an // untouched block reproduces its bytes exactly -- comparing frames alone // would pass whether or not the reuse path exists. Counting the // recompressions is what actually distinguishes the two. try testing.expectEqual(@as(u64, 1), res.blocks_recompressed);
var sealed = try segment.Sealed.parse(testing.allocator, new_bytes); defer sealed.deinit(testing.allocator); try testing.expectEqual(@as(u32, 3), sealed.header.block_count);
// untouched blocks carry the ORIGINAL compressed bytes, not a // re-compression that merely happens to decode the same for (1..3) |i| { const src_entry = source.block_index[i]; const new_entry = sealed.block_index[i]; const src_frame = old[src_entry.offset + 8 ..][0..src_entry.compressed_size]; const new_frame = new_bytes[new_entry.offset + 8 ..][0..new_entry.compressed_size]; try testing.expectEqualSlices(u8, src_frame, new_frame); try testing.expectEqual(src_entry.uncompressed_size, new_entry.uncompressed_size); } // the touched block genuinely shrank try testing.expectEqual(@as(u32, 9), sealed.block_index[0].event_count);
// and every surviving row still decodes in order var expect_seq: u64 = 2; for (0..sealed.block_index.len) |i| { var block = try sealed.readBlock(testing.allocator, i); defer block.deinit(testing.allocator); for (block.events) |ev| { try testing.expectEqual(expect_seq, ev.seq); expect_seq += 1; } } try testing.expectEqual(@as(u64, 31), expect_seq);}
test "rewrite file exhaustively propagates segment I/O faults without silent corruption" { segment_io.disarm(); defer segment_io.disarm(); const old = try buildSealed(testing.allocator); defer testing.allocator.free(old);
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 rewriteFile(testing.allocator, testing.io, reference_tmp.dir, "seg.jss", .{}, DropBelow{ .cutoff = 12 }, keepAbove); 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, }, rewriteFile(testing.allocator, testing.io, tmp.dir, "seg.jss", .{}, DropBelow{ .cutoff = 12 }, keepAbove)); 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 "rewrite: unrelated candidate DIDs skip the real segment decode" { const old = try buildSealed(testing.allocator); defer testing.allocator.free(old); var result: Result = .{}; const candidates = [_][]const u8{"did:plc:not-in-this-segment"}; try testing.expectEqual(@as(?[]u8, null), try rewrite( testing.allocator, old, &result, .{ .candidate_dids = &candidates }, {}, unreachableKeep, )); try testing.expect(!result.rewritten);}
test "rewriteFile: atomic in-place rewrite; stale tmp reclaim" { var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); var dir = try tmp.dir.createDirPathOpen(io, "segments", .{ .open_options = .{ .iterate = true } }); defer dir.close(io);
const old = try buildSealed(testing.allocator); defer testing.allocator.free(old); { var f = try dir.createFile(io, "seg_0000000000.jss", .{}); defer f.close(io); try f.writeStreamingAll(io, old); } // a crashed rewrite left a stale tmp { var f = try dir.createFile(io, "seg_0000000000.jss.tmp", .{}); f.close(io); } try cleanStaleTmp(testing.allocator, io, dir); try testing.expectError(error.FileNotFound, dir.access(io, "seg_0000000000.jss.tmp", .{}));
const res = try rewriteFile(testing.allocator, io, dir, "seg_0000000000.jss", .{}, DropBelow{ .cutoff = 21 }, keepAbove); try testing.expect(res.rewritten); try testing.expectEqual(@as(u64, 20), res.dropped);
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(@as(u32, 10), sealed.header.event_count); try testing.expectEqual(@as(u32, 3), sealed.header.block_count);
// second pass over the compacted file is clean const res2 = try rewriteFile(testing.allocator, io, dir, "seg_0000000000.jss", .{}, DropBelow{ .cutoff = 21 }, keepAbove); try testing.expect(!res2.rewritten);}
test "rebloom shrinks a pre-fix upstream fixture and is idempotent" { var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); // committed upstream-produced segment sealed before the bloom // cardinality-sizing fix (2026-07-08 < upstream f02919c): its per-block // blooms carry the fixed 4096-event capacity. skipped when absent. const old_bytes = Io.Dir.cwd().readFileAlloc( io, "tests/fixtures/seg_0000000000.jss", testing.allocator, .limited(4 << 20), ) catch return; defer testing.allocator.free(old_bytes);
var result: Result = .{}; const new_bytes = (try rebloom(testing.allocator, old_bytes, &result)) orelse return error.FixtureAlreadyRightSized; defer testing.allocator.free(new_bytes); try testing.expect(result.rewritten); try testing.expect(new_bytes.len < old_bytes.len);
// the rewritten file must parse, checksum, and serve identical rows var old_sealed = try segment.Sealed.parse(testing.allocator, old_bytes); defer old_sealed.deinit(testing.allocator); var new_sealed = try segment.Sealed.parse(testing.allocator, new_bytes); defer new_sealed.deinit(testing.allocator); try testing.expect(segment.checksumValid(new_bytes, new_sealed.header)); try testing.expectEqual(old_sealed.header.event_count, new_sealed.header.event_count); try testing.expectEqual(old_sealed.block_index.len, new_sealed.block_index.len); for (0..new_sealed.block_index.len) |i| { var old_block = try old_sealed.readBlock(testing.allocator, i); defer old_block.deinit(testing.allocator); var new_block = try new_sealed.readBlock(testing.allocator, i); defer new_block.deinit(testing.allocator); try testing.expectEqual(old_block.events.len, new_block.events.len); for (old_block.events, new_block.events) |a, b| { try testing.expectEqual(a.seq, b.seq); try testing.expectEqualStrings(a.did, b.did); try testing.expectEqualStrings(a.payload, b.payload); } }
// right-sized blooms still answer membership for every resident DID const blobs = try new_sealed.blockBloomBlobs(testing.allocator); defer testing.allocator.free(blobs); for (0..new_sealed.block_index.len) |i| { var block = try new_sealed.readBlock(testing.allocator, i); defer block.deinit(testing.allocator); for (block.events) |ev| { if (ev.did.len > 0) try testing.expect(try gloom.containsMarshaled(blobs[i], ev.did)); } }
// second pass: nothing left to shrink var again: Result = .{}; try testing.expectEqual(@as(?[]u8, null), try rebloom(testing.allocator, new_bytes, &again));}