Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751//! pass — one chunked delete-compaction pass over sealed segments//!//! upstream runDeleteCompaction (docs/upstream-compaction-spec.md)://! reclaim stale tmps, sweep sealed segment headers for the target//! watermark (max sealed max_seq; active segments are skipped — W never//! advances past them), then chunk from the committed watermark to the//! target. each chunk FOLDS ITS TOMBSTONES FROM DISK in//! (current, chunkEnd] — deliberately not the in-memory live set, which//! collapses each key to the global max seq and would let superseded//! rows below the committed W survive forever (issue #100 dual). block//! index seq bounds prune the fold (bounds survive rewrites as//! historical supersets, so pruning stays safe). after rewriting every//! overlapping segment the chunk watermark commits, then the live set//! evicts. a chunk cap bounds snapshot memory.//!//! candidate-DID segment blooms avoid unrelated decodes while the set is//! reasonably small; rewrites fan out across min(CPU, 8) workers.
const std = @import("std");const archive_mod = @import("../storage/archive.zig");const crashpoint = @import("../runtime/crashpoint.zig");const meta_store = @import("../storage/meta_store.zig");const metrics = @import("../runtime/metrics.zig");const rewrite_mod = @import("../storage/segment_rewrite.zig");const segment = @import("../storage/segment.zig");const segment_io = @import("../storage/segment_io.zig");const tombstone = @import("../storage/tombstone.zig");const watermark = @import("watermark.zig");
const Io = std.Io;const Allocator = std.mem.Allocator;const log = std.log.scoped(.stream);
/// upstream CompactionTombstoneCap is 32M entries sized for go maps;/// our per-entry cost is similar (hashmap slot + cloned key in the/// arena), so keep the constant until measured otherwisepub const default_tombstone_cap: usize = 32_000_000;
/// Segments per watermark commit. A chunk's tombstones are folded once, but/// the rewrite commits in ascending batches, because the watermark is the only/// thing a restart preserves. Production reached neither the tombstone cap nor/// the end of its single whole-archive chunk: 6,584 segments at ~4 segments/min/// is ~27h of rewriting behind one commit, and the process restarts sooner than/// that, so the watermark never moved and every restart refolded from scratch.pub const default_commit_segments: usize = 64;
pub const Options = struct { /// snapshot entry cap per chunk; reaching it closes the chunk at the /// current segment's max_seq tombstone_cap: usize = default_tombstone_cap, /// segments rewritten per watermark commit; 0 commits once per chunk commit_segments: usize = default_commit_segments, bloom_narrow_max_dids: usize = 100_000, rewrite_workers: usize = 0, instrumentation: ?*metrics.Stats = null, measure_pass: bool = true, on_rewritten_ctx: ?*anyopaque = null, on_rewritten: ?*const fn (?*anyopaque, u64) anyerror!void = null, rewrite_lock_ctx: ?*anyopaque = null, rewrite_lock: ?*const fn (?*anyopaque) void = null, rewrite_unlock: ?*const fn (?*anyopaque) void = null, on_committed_ctx: ?*anyopaque = null, on_committed: ?*const fn (?*anyopaque, u64) void = null,};
pub const Stats = struct { target_watermark: u64 = 0, chunks: u64 = 0, segments_examined: u64 = 0, segments_rewritten: u64 = 0, rows_dropped: u64 = 0, rows_dropped_by_reason: tombstone.Counts = .{}, segments_skipped_clean: u64 = 0, bytes_rewritten: u64 = 0, tombstones_folded: u64 = 0, /// Largest real rewrite worker group spawned by any chunk in this pass. max_rewrite_workers: u64 = 0,};
const SegMeta = struct { idx: u64, min_seq: u64, max_seq: u64, event_count: u32, min_witnessed_at: i64, max_witnessed_at: i64,};
/// run one pass. `seg_dir` is <data>/segments (iterable). `live` is the/// steady-state in-memory tombstone set to evict per committed chunk/// (null for merge-tail passes). errors are crash-loud to the caller —/// the watermark never advances past a failed rewrite.pub fn run( allocator: Allocator, io: Io, seg_dir: Io.Dir, wm: *watermark.Store, live: ?*tombstone.Snapshot, opts: Options,) !Stats { const started = Io.Timestamp.now(io, .awake); const result = runInner(allocator, io, seg_dir, wm, live, opts) catch |err| { if (opts.measure_pass) { if (opts.instrumentation) |s| s.observeCompactionPass(.error_result, elapsedUs(io, started)); } return err; }; if (opts.measure_pass) { if (opts.instrumentation) |s| s.observeCompactionPass(.ok, elapsedUs(io, started)); } return result;}
fn runInner( allocator: Allocator, io: Io, seg_dir: Io.Dir, wm: *watermark.Store, live: ?*tombstone.Snapshot, opts: Options,) !Stats { { if (opts.rewrite_lock) |lock| lock(opts.rewrite_lock_ctx); defer if (opts.rewrite_unlock) |unlock| unlock(opts.rewrite_lock_ctx); try rewrite_mod.cleanStaleTmp(allocator, io, seg_dir); }
var metas: std.ArrayList(SegMeta) = .empty; defer metas.deinit(allocator); try sweep(allocator, io, seg_dir, &metas);
var stats: Stats = .{}; var target: u64 = 0; for (metas.items) |m| if (m.event_count > 0) { target = @max(target, m.max_seq); }; stats.target_watermark = target;
var current = (try wm.load(io)) orelse 0; if (opts.instrumentation) |s| { s.compaction_watermark.store(current, .monotonic); s.compaction_watermark_lag_us.store(watermarkLagUs(metas.items, current), .monotonic); } if (metas.items.len == 0 or target <= current) { if (opts.instrumentation) |s| s.compaction_watermark_lag_us.store(0, .monotonic); return stats; }
while (current < target) { // fold this chunk's tombstones from disk in (current, chunk_end] var snap = tombstone.Snapshot.init(allocator); defer snap.deinit(); var chunk_end = target; var committed_seq = current; var folded: u64 = 0; for (metas.items) |m| { if (m.max_seq <= current) continue; try foldSegment(allocator, io, seg_dir, m.idx, current, &snap); folded += 1; if (opts.instrumentation) |s| _ = s.compaction_fold_segments_total.fetchAdd(1, .monotonic); if (folded % 200 == 0) log.info("compaction fold {d} segments into chunk: {d} tombstones", .{ folded, snap.len() }); if (opts.tombstone_cap > 0 and snap.len() >= opts.tombstone_cap and m.max_seq < target) { chunk_end = m.max_seq; break; } } const collected = snap.countsByReason(); stats.tombstones_folded += snap.len();
if (snap.len() > 0) { // Match upstream's lock scope: one mutating chunk. Timestamp // import may safely interleave between chunks, never within one. if (opts.rewrite_lock) |lock| lock(opts.rewrite_lock_ctx); defer if (opts.rewrite_unlock) |unlock| unlock(opts.rewrite_lock_ctx); const candidates = try snap.candidateDids(allocator, opts.bloom_narrow_max_dids); defer if (candidates) |items| allocator.free(items); const outcomes = try allocator.alloc(RewriteOutcome, metas.items.len); defer allocator.free(outcomes); @memset(outcomes, .{});
var tip_witness: i64 = 0; for (metas.items) |m| { if (m.event_count > 0) tip_witness = @max(tip_witness, m.max_witnessed_at); }
// Rewrite ascending in batches, saving the watermark after each. // Segments are still rewritten exactly once per chunk; only the // commit granularity changes, so a restart resumes near where it // stopped instead of refolding the whole archive. const batch = if (opts.commit_segments == 0) metas.items.len else opts.commit_segments; var lo: usize = 0; while (lo < metas.items.len) : (lo += batch) { const hi = @min(lo + batch, metas.items.len); const slice = metas.items[lo..hi]; var work: RewriteWork = .{ .allocator = allocator, .io = io, .seg_dir = seg_dir, .metas = slice, .snap = &snap, .chunk_end = chunk_end, .candidate_dids = candidates, .outcomes = outcomes[lo..hi], .instrumentation = opts.instrumentation, }; var group: Io.Group = .init; const cpu = std.Thread.getCpuCount() catch 1; const configured = if (opts.rewrite_workers == 0) @min(cpu, 8) else opts.rewrite_workers; const worker_count = @min(@max(configured, 1), slice.len); stats.max_rewrite_workers = @max(stats.max_rewrite_workers, worker_count); for (0..worker_count) |_| try group.concurrent(io, RewriteWork.runWorker, .{&work}); group.await(io) catch |err| return err; if (work.fatal_error) |err| return err;
for (slice, outcomes[lo..hi]) |m, outcome| { stats.segments_examined += 1; const r = outcome.result; if (r.rewritten) { if (opts.on_rewritten) |refresh| try refresh(opts.on_rewritten_ctx, m.idx); stats.segments_rewritten += 1; stats.rows_dropped += r.dropped; stats.rows_dropped_by_reason.record += outcome.drops.record; stats.rows_dropped_by_reason.sync += outcome.drops.sync; stats.rows_dropped_by_reason.account += outcome.drops.account; stats.bytes_rewritten += r.bytes_rewritten; if (opts.instrumentation) |s| { _ = s.compaction_segments_rewritten_total.fetchAdd(1, .monotonic); _ = s.compaction_rows_dropped.getPtr(.record).fetchAdd(outcome.drops.record, .monotonic); _ = s.compaction_rows_dropped.getPtr(.sync).fetchAdd(outcome.drops.sync, .monotonic); _ = s.compaction_rows_dropped.getPtr(.account).fetchAdd(outcome.drops.account, .monotonic); _ = s.compaction_bytes_rewritten_total.fetchAdd(r.bytes_rewritten, .monotonic); const age_us = tip_witness -| m.max_witnessed_at; s.observeCompactionRewriteAge(@intCast(@divTrunc(@max(0, age_us), std.time.us_per_s))); } log.info("compaction: rewrote segment {d} ({d} rows dropped)", .{ m.idx, r.dropped }); } else { stats.segments_skipped_clean += 1; if (opts.instrumentation) |s| _ = s.compaction_segments_skipped_clean_total.fetchAdd(1, .monotonic); } }
// Every segment at or below this point has had the chunk's // whole tombstone set applied, so the watermark may advance // here even though later batches are still pending: a // tombstone at seq j <= batch_end can only target rows // written before j, and those segments are already done. const batch_end = @min(chunk_end, slice[slice.len - 1].max_seq); if (batch_end > committed_seq and hi < metas.items.len) { crashpoint.hit(.after_compaction_rewrite_before_watermark); try wm.save(io, batch_end); committed_seq = batch_end; if (opts.instrumentation) |s| { s.compaction_watermark.store(batch_end, .monotonic); s.compaction_watermark_lag_us.store(watermarkLagUs(metas.items, batch_end), .monotonic); } if (opts.on_committed) |c| c(opts.on_committed_ctx, batch_end) else if (live) |l| l.evict(batch_end); log.info("compaction: watermark {d} ({d}/{d} segments this chunk)", .{ batch_end, hi, metas.items.len }); } } }
if (opts.instrumentation) |s| { _ = s.compaction_tombstones_collected.getPtr(.record).fetchAdd(collected.record, .monotonic); _ = s.compaction_tombstones_collected.getPtr(.sync).fetchAdd(collected.sync, .monotonic); _ = s.compaction_tombstones_collected.getPtr(.account).fetchAdd(collected.account, .monotonic); }
crashpoint.hit(.after_compaction_rewrite_before_watermark);
// durable commit, then eviction (rebuild-from-disk makes a crash // between the two immaterial) try wm.save(io, chunk_end); if (opts.instrumentation) |s| { s.compaction_watermark.store(chunk_end, .monotonic); s.compaction_watermark_lag_us.store(watermarkLagUs(metas.items, chunk_end), .monotonic); } if (opts.on_committed) |committed| committed(opts.on_committed_ctx, chunk_end) else if (live) |l| l.evict(chunk_end); crashpoint.hit(.after_compaction_chunk_watermark); stats.chunks += 1; current = chunk_end; } return stats;}
/// collect sealed segments with events, ascending index (== seq order)./// active segments are skipped; any other unreadable header fails the/// pass — silently skipping a sealed file would advance W past/// uncompacted rows.fn sweep(allocator: Allocator, io: Io, seg_dir: Io.Dir, metas: *std.ArrayList(SegMeta)) !void { var it = seg_dir.iterate(); while (try it.next(io)) |entry| { const idx = archive_mod.parseSegmentIndex(entry.name) orelse continue; var header_buf: [segment.header_size]u8 = undefined; var f = try seg_dir.openFile(io, entry.name, .{}); defer f.close(io); const n = try f.readPositionalAll(io, &header_buf, 0); const header = segment.Header.decode(header_buf[0..n]) catch |err| switch (err) { error.ActiveSegment => continue, else => return err, }; try metas.append(allocator, .{ .idx = idx, .min_seq = header.min_seq, .max_seq = header.max_seq, .event_count = header.event_count, .min_witnessed_at = header.min_witnessed_at, .max_witnessed_at = header.max_witnessed_at, }); } std.mem.sort(SegMeta, metas.items, {}, struct { fn lt(_: void, a: SegMeta, b: SegMeta) bool { return a.idx < b.idx; } }.lt);}
/// observe every marker row with seq > low_exclusive into the snapshot;/// block index seq bounds prune the read. pub for the steady compactor's/// startup rebuild (fold everything above the committed watermark).pub fn foldSegment( allocator: Allocator, io: Io, seg_dir: Io.Dir, idx: u64, low_exclusive: u64, snap: *tombstone.Snapshot,) !void { var name_buf: [64]u8 = undefined; const name = archive_mod.formatSegmentName(&name_buf, idx); // Header-level watermark skip (upstream compact_deletes.go rebuildLive- // Tombstones): a segment wholly at or below the watermark is already // physically compacted and can contribute nothing. Deciding that from // the header alone keeps rebuild cost scaled to the watermark backlog, // not the archive — without it, startup re-reads every sealed segment // in full (~1.7s each, hours on the production archive). if (low_exclusive > 0) { var file = try seg_dir.openFile(io, name, .{}); defer file.close(io); const header = try segment.readHeaderFile(io, &file); if (header.max_seq <= low_exclusive) return; } const bytes = try seg_dir.readFileAlloc(io, name, allocator, .limited(1 << 31)); defer allocator.free(bytes); var sealed = try segment.Sealed.parse(allocator, bytes); defer sealed.deinit(allocator);
for (sealed.block_index, 0..) |entry, i| { if (entry.max_seq <= low_exclusive or entry.event_count == 0) continue; var block = try sealed.readBlock(allocator, i); defer block.deinit(allocator); for (block.events) |ev| { if (ev.seq > low_exclusive) try snap.observe(ev); } }}
const Decide = struct { chunk_end: u64, snap: *const tombstone.Snapshot, drops: *tombstone.Counts,};
/// keep rows above the chunk (a later chunk's fold covers them), drop/// covered materialization rows below itfn decide(ctx: *Decide, ev: segment.Event) bool { if (ev.seq > ctx.chunk_end) return true; const reason = ctx.snap.dropReason(ev) orelse return true; switch (reason) { .record => ctx.drops.record += 1, .sync => ctx.drops.sync += 1, .account => ctx.drops.account += 1, } return false;}
const RewriteOutcome = struct { result: rewrite_mod.Result = .{}, drops: tombstone.Counts = .{},};
const RewriteWork = struct { allocator: Allocator, io: Io, seg_dir: Io.Dir, metas: []const SegMeta, snap: *const tombstone.Snapshot, chunk_end: u64, candidate_dids: ?[]const []const u8, outcomes: []RewriteOutcome, instrumentation: ?*metrics.Stats, next: std.atomic.Value(usize) = .init(0), fatal_mu: Io.Mutex = .init, fatal_error: ?anyerror = null, failed: std.atomic.Value(bool) = .init(false),
fn runWorker(self: *RewriteWork) void { while (true) { if (self.failed.load(.acquire)) return; const index = self.next.fetchAdd(1, .monotonic); if (index >= self.metas.len) return; self.process(index) catch |err| { self.fatal_mu.lockUncancelable(self.io); if (self.fatal_error == null) self.fatal_error = err; self.fatal_mu.unlock(self.io); self.failed.store(true, .release); return; }; } }
fn process(self: *RewriteWork, index: usize) !void { if (self.instrumentation) |s| _ = s.compaction_segments_examined_total.fetchAdd(1, .monotonic); const m = self.metas[index]; var name_buf: [64]u8 = undefined; const name = archive_mod.formatSegmentName(&name_buf, m.idx); var outcome: RewriteOutcome = .{}; var ctx: Decide = .{ .chunk_end = self.chunk_end, .snap = self.snap, .drops = &outcome.drops }; outcome.result = try rewrite_mod.rewriteFile(self.allocator, self.io, self.seg_dir, name, .{ .candidate_dids = self.candidate_dids, }, &ctx, decide); self.outcomes[index] = outcome; }};
fn watermarkLagUs(metas: []const SegMeta, watermark_seq: u64) u64 { if (metas.len == 0) return 0; var tip_witness: i64 = 0; var oldest_witness: i64 = std.math.maxInt(i64); var watermark_witness: ?i64 = null; for (metas) |m| { if (m.event_count == 0) continue; tip_witness = @max(tip_witness, m.max_witnessed_at); oldest_witness = @min(oldest_witness, m.min_witnessed_at); if (m.max_seq <= watermark_seq) watermark_witness = @max(watermark_witness orelse m.max_witnessed_at, m.max_witnessed_at); } const base = watermark_witness orelse oldest_witness; if (tip_witness <= base) return 0; return @intCast(tip_witness - base);}
fn elapsedUs(io: Io, started: Io.Timestamp) u64 { return @intCast(@max(0, started.durationTo(Io.Timestamp.now(io, .awake)).toMicroseconds()));}
// === tests ===
const testing = std.testing;
fn row(kind: segment.Kind, did: []const u8, coll: []const u8, rkey: []const u8, payload: []const u8) segment.Event { return .{ .seq = 0, // archive assigns .witnessed_at = 1, .indexed_at = 0, .kind = kind, .did = did, .collection = coll, .rkey = rkey, .rev = "3l3qo2vutsw2b", .payload = payload, };}
// {"active": false, "status": "deleted"} as dag-cborconst account_deleted = "\xa2\x66\x61\x63\x74\x69\x76\x65\xf4\x66\x73\x74\x61\x74\x75\x73\x67\x64\x65\x6c\x65\x74\x65\x64";
test "pass: drops superseded rows, keeps markers, commits watermark, idempotent" { var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); var path_buf: [Io.Dir.max_path_bytes]u8 = undefined; const n = try tmp.dir.realPath(io, &path_buf); var data_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined; const data_dir = try std.fmt.bufPrint(&data_dir_buf, "{s}/data", .{path_buf[0..n]});
// seqs: 1 create a/k1, 2 create a/k2, 3 create b/k1, 4 delete a/k1, // 5 account-delete b, 6 create a/k3 (nothing covers it) { var a = try archive_mod.Archive.init(testing.allocator, io, data_dir); defer a.deinit(); a.max_events_per_block = 2; a.writer.max_events_per_block = 2; const post = "app.bsky.feed.post"; _ = try a.append(row(.create, "did:plc:a", post, "k1", "x"), 1); _ = try a.append(row(.create, "did:plc:a", post, "k2", "x"), 2); _ = try a.append(row(.create, "did:plc:b", post, "k1", "x"), 3); _ = try a.append(row(.delete, "did:plc:a", post, "k1", ""), 4); _ = try a.append(row(.account, "did:plc:b", "", "", account_deleted), 5); _ = try a.append(row(.create, "did:plc:a", post, "k3", "x"), 6); try a.rotate(); // seal; leaves a fresh empty active behind }
var meta = try meta_store.Store.open(testing.allocator, data_dir); defer meta.deinit(); var wm = watermark.Store.init(&meta); defer wm.deinit(io); var seg_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined; const seg_path = try std.fmt.bufPrint(&seg_dir_buf, "{s}/segments", .{data_dir}); var seg_dir = try Io.Dir.cwd().openDir(io, seg_path, .{ .iterate = true }); defer seg_dir.close(io);
var instrumentation: metrics.Stats = .{}; const stats = try run(testing.allocator, io, seg_dir, &wm, null, .{ .instrumentation = &instrumentation }); try testing.expectEqual(@as(u64, 6), stats.target_watermark); try testing.expectEqual(@as(u64, 1), stats.chunks); try testing.expectEqual(@as(u64, 1), stats.segments_rewritten); // dropped: create a/k1 (seq 1 < delete seq 4), create b/k1 (seq 3 < account seq 5) try testing.expectEqual(@as(u64, 2), stats.rows_dropped); try testing.expectEqual(@as(u64, 1), stats.rows_dropped_by_reason.record); try testing.expectEqual(@as(u64, 1), stats.rows_dropped_by_reason.account); try testing.expect(stats.bytes_rewritten > 0); try testing.expectEqual(@as(?u64, 6), try wm.load(io)); try testing.expectEqual(@as(u64, 1), instrumentation.compaction_passes.getPtrConst(.ok).load(.monotonic)); try testing.expectEqual(@as(u64, 1), instrumentation.compaction_tombstones_collected.getPtrConst(.record).load(.monotonic)); try testing.expectEqual(@as(u64, 1), instrumentation.compaction_tombstones_collected.getPtrConst(.account).load(.monotonic)); try testing.expectEqual(@as(u64, 1), instrumentation.compaction_rows_dropped.getPtrConst(.record).load(.monotonic)); try testing.expectEqual(@as(u64, 1), instrumentation.compaction_rows_dropped.getPtrConst(.account).load(.monotonic)); try testing.expectEqual(@as(u64, 1), instrumentation.compaction_segments_examined_total.load(.monotonic)); try testing.expectEqual(@as(u64, 1), instrumentation.compaction_segments_rewritten_total.load(.monotonic)); try testing.expect(instrumentation.compaction_bytes_rewritten_total.load(.monotonic) > 0); try testing.expectEqual(@as(u64, 1), instrumentation.compaction_rewritten_age_count.load(.monotonic)); try testing.expect(instrumentation.compaction_fold_segments_total.load(.monotonic) >= 1); try testing.expectEqual(@as(u64, 6), instrumentation.compaction_watermark.load(.monotonic)); try testing.expectEqual(@as(u64, 0), instrumentation.compaction_watermark_lag_us.load(.monotonic));
// surviving rows: creates k2/k3 + both markers; block topology intact const bytes = try seg_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, 3), sealed.header.block_count); try testing.expectEqual(@as(u32, 4), sealed.header.event_count); var seqs: std.ArrayList(u64) = .empty; defer seqs.deinit(testing.allocator); 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 seqs.append(testing.allocator, ev.seq); } try testing.expectEqualSlices(u64, &.{ 2, 4, 5, 6 }, seqs.items);
// second pass: nothing above the watermark → no-op const again = try run(testing.allocator, io, seg_dir, &wm, null, .{ .instrumentation = &instrumentation }); try testing.expectEqual(@as(u64, 0), again.chunks); try testing.expectEqual(@as(u64, 0), again.segments_rewritten); try testing.expectEqual(@as(u64, 2), instrumentation.compaction_passes.getPtrConst(.ok).load(.monotonic));}
test "pass: tombstone cap splits the pass into chunks across segments" { var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); var path_buf: [Io.Dir.max_path_bytes]u8 = undefined; const n = try tmp.dir.realPath(io, &path_buf); var data_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined; const data_dir = try std.fmt.bufPrint(&data_dir_buf, "{s}/data", .{path_buf[0..n]});
// two sealed segments, each create-then-delete of a distinct key { var a = try archive_mod.Archive.init(testing.allocator, io, data_dir); defer a.deinit(); const post = "app.bsky.feed.post"; _ = try a.append(row(.create, "did:plc:a", post, "k1", "x"), 1); _ = try a.append(row(.delete, "did:plc:a", post, "k1", ""), 2); try a.rotate(); _ = try a.append(row(.create, "did:plc:a", post, "k2", "x"), 3); _ = try a.append(row(.delete, "did:plc:a", post, "k2", ""), 4); try a.rotate(); }
var meta = try meta_store.Store.open(testing.allocator, data_dir); defer meta.deinit(); var wm = watermark.Store.init(&meta); defer wm.deinit(io); var seg_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined; const seg_path = try std.fmt.bufPrint(&seg_dir_buf, "{s}/segments", .{data_dir}); var seg_dir = try Io.Dir.cwd().openDir(io, seg_path, .{ .iterate = true }); defer seg_dir.close(io);
const stats = try run(testing.allocator, io, seg_dir, &wm, null, .{ .tombstone_cap = 1, .rewrite_workers = 2 }); try testing.expectEqual(@as(u64, 2), stats.chunks); try testing.expectEqual(@as(u64, 2), stats.segments_rewritten); try testing.expectEqual(@as(u64, 2), stats.rows_dropped); try testing.expectEqual(@as(u64, 4), stats.segments_examined); try testing.expectEqual(@as(u64, 2), stats.segments_skipped_clean); try testing.expectEqual(@as(u64, 2), stats.max_rewrite_workers); try testing.expectEqual(@as(?u64, 4), try wm.load(io));}
test "pass: one chunk commits the watermark per batch, so progress survives a kill" { var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); var path_buf: [Io.Dir.max_path_bytes]u8 = undefined; const n = try tmp.dir.realPath(io, &path_buf); var data_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined; const data_dir = try std.fmt.bufPrint(&data_dir_buf, "{s}/data", .{path_buf[0..n]});
// four sealed segments, one create-then-delete pair each, all inside a // single chunk (the tombstone cap is nowhere near) { var a = try archive_mod.Archive.init(testing.allocator, io, data_dir); defer a.deinit(); const post = "app.bsky.feed.post"; for (0..4) |i| { var rkey_buf: [8]u8 = undefined; const rkey = try std.fmt.bufPrint(&rkey_buf, "k{d}", .{i}); const base: i64 = @intCast(i * 2); _ = try a.append(row(.create, "did:plc:a", post, rkey, "x"), base + 1); _ = try a.append(row(.delete, "did:plc:a", post, rkey, ""), base + 2); try a.rotate(); } }
var meta = try meta_store.Store.open(testing.allocator, data_dir); defer meta.deinit(); var wm = watermark.Store.init(&meta); defer wm.deinit(io); var seg_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined; const seg_path = try std.fmt.bufPrint(&seg_dir_buf, "{s}/segments", .{data_dir}); var seg_dir = try Io.Dir.cwd().openDir(io, seg_path, .{ .iterate = true }); defer seg_dir.close(io);
// Watermark saves observed as the pass runs. Before batching, the only // save was the final one, so a process killed mid-pass resumed from zero // no matter how many segments it had already rewritten. const Recorder = struct { var saves: std.ArrayList(u64) = .empty; fn onCommitted(_: ?*anyopaque, seq: u64) void { saves.append(testing.allocator, seq) catch @panic("oom"); } }; Recorder.saves = .empty; defer Recorder.saves.deinit(testing.allocator);
const stats = try run(testing.allocator, io, seg_dir, &wm, null, .{ .commit_segments = 1, .on_committed = Recorder.onCommitted, }); try testing.expectEqual(@as(u64, 1), stats.chunks); try testing.expectEqual(@as(u64, 4), stats.segments_rewritten); try testing.expectEqual(@as(u64, 4), stats.rows_dropped); try testing.expectEqual(@as(?u64, 8), try wm.load(io));
// one chunk, but the watermark advanced at every batch boundary along the // way rather than only at the end const batched = try Recorder.saves.toOwnedSlice(testing.allocator); defer testing.allocator.free(batched); try testing.expectEqualSlices(u64, &.{ 2, 4, 6, 8 }, batched);
// and a whole-chunk commit is still available for callers that want it var wm2 = watermark.Store.init(&meta); defer wm2.deinit(io); try wm2.save(io, 0); _ = try run(testing.allocator, io, seg_dir, &wm2, null, .{ .commit_segments = 0, .on_committed = Recorder.onCommitted, }); try testing.expectEqual(@as(usize, 1), Recorder.saves.items.len);}
test "pass: commit rename fault fails loud, preserves source, and retry converges" { segment_io.disarm(); defer segment_io.disarm(); var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); var tmp = testing.tmpDir(.{}); defer tmp.cleanup(); var path_buf: [Io.Dir.max_path_bytes]u8 = undefined; const n = try tmp.dir.realPath(io, &path_buf); var data_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined; const data_dir = try std.fmt.bufPrint(&data_dir_buf, "{s}/data", .{path_buf[0..n]}); { var archive = try archive_mod.Archive.init(testing.allocator, io, data_dir); defer archive.deinit(); const post = "app.bsky.feed.post"; _ = try archive.append(row(.create, "did:plc:a", post, "k", "x"), 1); _ = try archive.append(row(.delete, "did:plc:a", post, "k", ""), 2); try archive.rotate(); } var meta = try meta_store.Store.open(testing.allocator, data_dir); defer meta.deinit(); var wm = watermark.Store.init(&meta); defer wm.deinit(io); var seg_path_buf: [Io.Dir.max_path_bytes]u8 = undefined; const seg_path = try std.fmt.bufPrint(&seg_path_buf, "{s}/segments", .{data_dir}); var seg_dir = try Io.Dir.cwd().openDir(io, seg_path, .{ .iterate = true }); defer seg_dir.close(io); const original = try seg_dir.readFileAlloc(io, "seg_0000000000.jss", testing.allocator, .limited(1 << 20)); defer testing.allocator.free(original);
try testing.expect(segment_io.armSpec("rename:1:eio")); try testing.expectError(error.InputOutput, run(testing.allocator, io, seg_dir, &wm, null, .{ .rewrite_workers = 1 })); try testing.expectEqual(@as(?u64, null), try wm.load(io)); try testing.expectError(error.FileNotFound, seg_dir.openFile(io, "seg_0000000000.jss.tmp", .{})); const after_fault = try seg_dir.readFileAlloc(io, "seg_0000000000.jss", testing.allocator, .limited(1 << 20)); defer testing.allocator.free(after_fault); try testing.expectEqualSlices(u8, original, after_fault);
segment_io.disarm(); const recovered = try run(testing.allocator, io, seg_dir, &wm, null, .{ .rewrite_workers = 1 }); try testing.expectEqual(@as(u64, 1), recovered.segments_rewritten); try testing.expectEqual(@as(u64, 1), recovered.rows_dropped); try testing.expectEqual(@as(?u64, 2), try wm.load(io));}
test "pass: malformed account marker records failure without advancing watermark" { var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); var tmp = testing.tmpDir(.{}); defer tmp.cleanup(); var path_buf: [Io.Dir.max_path_bytes]u8 = undefined; const n = try tmp.dir.realPath(io, &path_buf); var data_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined; const data_dir = try std.fmt.bufPrint(&data_dir_buf, "{s}/data", .{path_buf[0..n]}); { var archive = try archive_mod.Archive.init(testing.allocator, io, data_dir); defer archive.deinit(); _ = try archive.append(row(.account, "did:plc:a", "", "", "\xa1"), 1); try archive.rotate(); } var meta = try meta_store.Store.open(testing.allocator, data_dir); defer meta.deinit(); var wm = watermark.Store.init(&meta); defer wm.deinit(io); var seg_path_buf: [Io.Dir.max_path_bytes]u8 = undefined; const seg_path = try std.fmt.bufPrint(&seg_path_buf, "{s}/segments", .{data_dir}); var seg_dir = try Io.Dir.cwd().openDir(io, seg_path, .{ .iterate = true }); defer seg_dir.close(io); var instrumentation: metrics.Stats = .{};
try testing.expectError(error.UnexpectedEof, run(testing.allocator, io, seg_dir, &wm, null, .{ .instrumentation = &instrumentation, })); try testing.expectEqual(@as(?u64, null), try wm.load(io)); try testing.expectEqual(@as(u64, 1), instrumentation.compaction_passes.getPtrConst(.error_result).load(.monotonic)); try testing.expectEqual(@as(u64, 0), instrumentation.compaction_passes.getPtrConst(.ok).load(.monotonic));}