Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635//! steady — the long-running compactor beside the live consumer//!//! upstream runSteadyCompactor (docs/upstream-compaction-spec.md): a//! pass every interval (default 4h; 0 disables compaction entirely),//! plus an early coalesced trigger when the live in-memory tombstone//! set reaches the cap, floored by 30s spacing so a failed pass can't//! re-fire per event. each pass force-rotates the active segment first —//! race-free because the live set is fed by the archive's on_append hook//! under the writer lock, so by rotate's return every sealed row's//! tombstone is observed and later rows land above the pass target. the//! live set is used ONLY for the cap trigger; passes fold their drop//! windows from disk. a failed pass logs and retries at the next scheduled//! pass — it must never tear down the daemon.//!//! every successful rewrite refreshes the resident header/block-index entry//! before its watermark may commit. In-flight readers retain the old//! immutable index, but obtain physical offsets from their freshly opened//! file generation; topology-preserving rewrites make that combination safe.
const std = @import("std");const archive_mod = @import("../storage/archive.zig");const metrics = @import("../runtime/metrics.zig");const meta_store = @import("../storage/meta_store.zig");const observability = @import("../runtime/observability.zig");const pass = @import("pass.zig");const segment = @import("../storage/segment.zig");const segment_rewrite = @import("../storage/segment_rewrite.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);
pub const Compactor = struct { allocator: Allocator, io: Io, archive: *archive_mod.Archive, meta: *meta_store.Store, data_dir: []const u8, /// nanoseconds between passes; 0 disables (run() returns immediately) interval_ns: u64 = 4 * std.time.ns_per_hour, tombstone_cap: usize = pass.default_tombstone_cap, rewrite_workers: usize = 0, min_trigger_spacing_ns: u64 = 30 * std.time.ns_per_s, /// guards `live` (observe on the writer thread, evict on ours) mu: Io.Mutex = .init, live: tombstone.Snapshot, stop_flag: std.atomic.Value(bool) = .init(false), /// coalesced cap trigger (upstream's 1-buffered channel) trigger: std.atomic.Value(bool) = .init(false), wake: Io.Event = .unset, stats: ?*metrics.Stats = null, trace_context: observability.Context = observability.root_context,
pub fn init(allocator: Allocator, io: Io, archive: *archive_mod.Archive, meta: *meta_store.Store, data_dir: []const u8) Compactor { return .{ .allocator = allocator, .io = io, .archive = archive, .meta = meta, .data_dir = data_dir, .live = tombstone.Snapshot.init(allocator), }; }
pub fn deinit(self: *Compactor) void { self.live.deinit(); }
pub fn requestStop(self: *Compactor) void { self.stop_flag.store(true, .release); self.wake.set(self.io); }
/// archive on_append hook; runs under the archive writer lock pub fn onAppend(ctx: *anyopaque, ev: segment.Event) !void { const self: *Compactor = @ptrCast(@alignCast(ctx)); self.mu.lockUncancelable(self.io); defer self.mu.unlock(self.io); try self.live.observe(ev); self.publishTombstoneGauges(); if (self.tombstone_cap > 0 and self.live.len() >= self.tombstone_cap) { self.trigger.store(true, .release); self.wake.set(self.io); } }
/// startup rebuild (upstream rebuildLiveTombstones): fold sealed rows /// above the committed watermark into the live set. cost scales with /// the watermark backlog — segments/blocks at or below it are pruned. /// the recovered-then-sealed active segment is covered; the fresh /// active is empty. pub fn rebuild(self: *Compactor) !void { var wm = watermark.Store.init(self.meta); defer wm.deinit(self.io); const w = (try wm.load(self.io)) orelse 0;
var seg_dir = try self.openSegDir(); defer seg_dir.close(self.io); self.mu.lockUncancelable(self.io); defer self.mu.unlock(self.io); // This scan runs before ingest dials (upstream ordering) and, with a // low watermark on a large archive, takes hours. Announce the work up // front and log progress so a long rebuild is a progress bar, never a // silent hang (the 2026-08-07 "3.5h restart pause" diagnosis chase). var total: u64 = 0; var count_it = seg_dir.iterate(); while (try count_it.next(self.io)) |entry| { if (archive_mod.parseSegmentIndex(entry.name) != null) total += 1; } log.info("tombstone rebuild starting above watermark {d}: {d} segments to fold (ingest dials after this completes)", .{ w, total }); var folded: u64 = 0; var it = seg_dir.iterate(); while (try it.next(self.io)) |entry| { const idx = archive_mod.parseSegmentIndex(entry.name) orelse continue; pass.foldSegment(self.allocator, self.io, seg_dir, idx, w, &self.live) catch |err| switch (err) { error.ActiveSegment => continue, else => return err, }; folded += 1; if (folded % 200 == 0) { log.info("tombstone rebuild {d}/{d}: {d} live entries", .{ folded, total, self.live.len() }); if (self.stats) |s| s.compaction_tombstone_set_entries.store(self.live.len(), .monotonic); } } if (self.stats) |s| s.compaction_watermark.store(w, .monotonic); self.publishTombstoneGauges(); log.info("live tombstones rebuilt above watermark {d}: {d} entries", .{ w, self.live.len() }); }
/// timer loop; spawn via io.concurrent, stop via requestStop pub fn run(self: *Compactor) void { if (self.interval_ns == 0) return; // boot time, so a compactor that never completes a pass shows as // growing age rather than a healthy-looking zero self.publishLastPassNow(); // the loop below has silent return paths (Canceled propagation); a // compactor that stops running must never do so invisibly — zero // passes over 4+ days went unexplained twice (2026-08-09) because // nothing distinguished "waiting" from "exited" log.info("compactor timer loop running (interval {d}s)", .{self.interval_ns / std.time.ns_per_s}); defer log.warn("compactor timer loop exited (no further passes will run)", .{}); const interval: Io.Clock.Duration = .{ .raw = Io.Duration.fromNanoseconds(@intCast(self.interval_ns)), .clock = .awake, }; var last_pass: ?Io.Clock.Timestamp = null; var deadline = Io.Clock.Timestamp.fromNow(self.io, interval); while (!self.stop_flag.load(.acquire)) { while (Io.Clock.Timestamp.now(self.io, .awake).compare(.lt, deadline)) { self.wake.waitTimeout(self.io, .{ .deadline = deadline }) catch |err| switch (err) { error.Timeout => continue, error.Canceled => return, }; if (self.stop_flag.load(.acquire)) return; break; } if (self.stop_flag.load(.acquire)) return; // Reset before consuming the atomic trigger. An append racing // after reset either joins this pass or leaves the event set for // the next loop; no cap notification can be lost. self.wake.reset(); const now = Io.Clock.Timestamp.now(self.io, .awake); const periodic_due = !now.compare(.lt, deadline); // Consume the coalesced trigger even when spacing suppresses it, // exactly like upstream's one-buffered channel. A later append // above the cap re-arms it; the first-ever trigger is immediate. const triggered = self.trigger.swap(false, .acq_rel); const cap_due = triggered and (last_pass == null or last_pass.?.durationTo(now).raw.toNanoseconds() >= self.min_trigger_spacing_ns); const due = periodic_due or cap_due; if (!due) continue; const early = cap_due and !periodic_due; const result = self.runOnce(); if (early) { if (self.stats) |s| _ = s.compaction_passes_early_total.fetchAdd(1, .monotonic); } const stats = result catch |err| { // failure policy: never tear down the daemon; retry later log.warn("compaction pass failed: {s} (retrying next interval)", .{@errorName(err)}); last_pass = Io.Clock.Timestamp.now(self.io, .awake); deadline = Io.Clock.Timestamp.fromNow(self.io, interval); continue; }; log.info("compaction pass: watermark {d}, {d} chunks, {d}/{d} segments rewritten, {d} rows dropped, {d} rewrite workers", .{ stats.target_watermark, stats.chunks, stats.segments_rewritten, stats.segments_examined, stats.rows_dropped, stats.max_rewrite_workers, }); last_pass = Io.Clock.Timestamp.now(self.io, .awake); deadline = Io.Clock.Timestamp.fromNow(self.io, interval); } }
/// one force-rotate + pass + eviction; also the test seam pub fn runOnce(self: *Compactor) !pass.Stats { var trace = observability.Span.start( self.allocator, "ingest/orchestrator", "runDeleteCompaction", self.trace_context, &.{}, ); defer trace.deinit(); defer trace.succeed(); return trace.recordError(self.runObservedPass(trace.context())); }
fn runObservedPass(self: *Compactor, trace_context: observability.Context) !pass.Stats { const started = Io.Timestamp.now(self.io, .awake); const result = self.runOnceInner(trace_context) catch |err| { if (self.stats) |s| s.observeCompactionPass(.error_result, elapsedUs(self.io, started)); self.publishLastPassNow(); return @errorCast(err); }; if (self.stats) |s| s.observeCompactionPass(.ok, elapsedUs(self.io, started)); self.publishLastPassNow(); return result; }
fn runOnceInner(self: *Compactor, trace_context: observability.Context) !pass.Stats { // rows deleted while their segment was active get compacted this // pass; on_append ordering makes this race-free (module doc) try self.archive.rotateTraced(trace_context);
var wm = watermark.Store.init(self.meta); defer wm.deinit(self.io); var seg_dir = try self.openSegDir(); defer seg_dir.close(self.io); const reconciled = try self.archive.reconcileManifest(); if (self.stats) |s| _ = s.compaction_manifest_reconciled_total.fetchAdd(reconciled, .monotonic);
// the pass gets no live-set handle: it folds from disk, and we // evict once after the final chunk commit (observe and evict then // never race; entries <= W only cover physically-compacted rows) const stats = try pass.run(self.allocator, self.io, seg_dir, &wm, null, .{ .tombstone_cap = self.tombstone_cap, .rewrite_workers = self.rewrite_workers, .instrumentation = self.stats, .measure_pass = false, .on_rewritten_ctx = self, .on_rewritten = refreshManifest, .rewrite_lock_ctx = self.archive, .rewrite_lock = lockRewrites, .rewrite_unlock = unlockRewrites, .on_committed_ctx = self, .on_committed = onCommitted, }); return stats; }
fn publishLastPassNow(self: *Compactor) void { if (self.stats) |s| s.compaction_last_pass_unix_s.store( @divTrunc(Io.Timestamp.now(self.io, .real).toMicroseconds(), std.time.us_per_s), .monotonic, ); }
fn publishTombstoneGauges(self: *Compactor) void { if (self.stats) |s| { s.compaction_tombstone_set_entries.store(self.live.len(), .monotonic); s.compaction_tombstone_set_bytes.store(self.live.approxBytes(), .monotonic); } }
fn openSegDir(self: *Compactor) !Io.Dir { var buf: [256]u8 = undefined; const seg_path = try std.fmt.bufPrint(&buf, "{s}/segments", .{self.data_dir}); return Io.Dir.cwd().openDir(self.io, seg_path, .{ .iterate = true }); }
fn refreshManifest(ctx: ?*anyopaque, idx: u64) !void { const self: *Compactor = @ptrCast(@alignCast(ctx.?)); try self.archive.manifest.refresh(self.archive.dir, idx, true); if (self.stats) |s| _ = s.compaction_manifest_reconciled_total.fetchAdd(1, .monotonic); }
fn lockRewrites(ctx: ?*anyopaque) void { const archive: *archive_mod.Archive = @ptrCast(@alignCast(ctx.?)); archive.lockRewrites(); }
fn unlockRewrites(ctx: ?*anyopaque) void { const archive: *archive_mod.Archive = @ptrCast(@alignCast(ctx.?)); archive.unlockRewrites(); }
fn onCommitted(ctx: ?*anyopaque, watermark_seq: u64) void { const self: *Compactor = @ptrCast(@alignCast(ctx.?)); self.mu.lockUncancelable(self.io); defer self.mu.unlock(self.io); self.live.evict(watermark_seq); self.publishTombstoneGauges(); }};
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, rkey: []const u8, payload: []const u8) segment.Event { return .{ .seq = 0, // archive assigns .witnessed_at = 1, .indexed_at = 0, .kind = kind, .did = did, .collection = "app.bsky.feed.post", .rkey = rkey, .rev = "3l3qo2vutsw2b", .payload = payload, };}
const KeepAtOrAbove = struct { cutoff: u64 };fn keepAtOrAbove(ctx: KeepAtOrAbove, event: segment.Event) bool { return event.seq >= ctx.cutoff;}
test "steady: hook feeds live set, runOnce force-rotates, compacts, evicts" { 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]});
var a = try archive_mod.Archive.init(testing.allocator, io, data_dir); defer a.deinit(); var meta = try meta_store.Store.open(testing.allocator, data_dir); defer meta.deinit();
var c = Compactor.init(testing.allocator, io, &a, &meta, data_dir); defer c.deinit(); var instrumentation: metrics.Stats = .{}; c.stats = &instrumentation; a.on_append_ctx = &c; a.on_append = Compactor.onAppend;
// create k1, then delete it — both land in the ACTIVE segment _ = try a.append(row(.create, "did:plc:a", "k1", "x"), 1); _ = try a.append(row(.create, "did:plc:a", "k2", "x"), 2); _ = try a.append(row(.delete, "did:plc:a", "k1", ""), 3); try testing.expectEqual(@as(usize, 1), c.live.len()); try testing.expectEqual(@as(usize, 1), instrumentation.compaction_tombstone_set_entries.load(.monotonic)); try testing.expect(instrumentation.compaction_tombstone_set_bytes.load(.monotonic) > 64);
// force-rotate seals them; the pass drops the superseded create const stats = try c.runOnce(); try testing.expectEqual(@as(u64, 3), stats.target_watermark); try testing.expectEqual(@as(u64, 1), stats.segments_rewritten); try testing.expectEqual(@as(u64, 1), stats.rows_dropped); // pass completion stamps the liveness gauge (alertable staleness signal) try testing.expect(instrumentation.compaction_last_pass_unix_s.load(.monotonic) > 0); const resident = a.manifest.segmentByIndex(0).?; try testing.expectEqual(@as(u32, 2), resident.header.event_count); var resident_blocks = try a.manifest.blockIndex(0); defer resident_blocks.deinit(); try testing.expectEqual(@as(u32, 2), resident_blocks.entries[0].event_count); // eviction cleared the covered entry (delete seq 3 <= watermark 3) try testing.expectEqual(@as(usize, 0), c.live.len()); try testing.expectEqual(@as(usize, 0), instrumentation.compaction_tombstone_set_entries.load(.monotonic)); try testing.expectEqual(@as(u64, 1), instrumentation.compaction_passes.getPtrConst(.ok).load(.monotonic)); try testing.expectEqual(@as(u64, 1), instrumentation.compaction_manifest_reconciled_total.load(.monotonic));
// Model rewrite-succeeded/refresh-failed: mutate the sealed file through // the real atomic rewriter without notifying the manifest. The next // steady pass must heal the checksum before its no-op watermark check. const old_checksum = resident.header.checksum; const direct = try segment_rewrite.rewriteFile( testing.allocator, io, a.dir, "seg_0000000000.jss", .{}, KeepAtOrAbove{ .cutoff = 3 }, keepAtOrAbove, ); try testing.expect(direct.rewritten); var rewritten_file = try a.dir.openFile(io, "seg_0000000000.jss", .{}); defer rewritten_file.close(io); const rewritten_header = try segment.readHeaderFile(io, &rewritten_file); try testing.expect(old_checksum != rewritten_header.checksum); const healed = try c.runOnce(); try testing.expectEqual(@as(u64, 0), healed.chunks); try testing.expectEqual(rewritten_header.checksum, a.manifest.segmentByIndex(0).?.header.checksum); try testing.expectEqual(@as(u64, 2), instrumentation.compaction_manifest_reconciled_total.load(.monotonic));
// rebuild from disk reproduces an empty live set above the watermark var c2 = Compactor.init(testing.allocator, io, &a, &meta, data_dir); defer c2.deinit(); try c2.rebuild(); try testing.expectEqual(@as(usize, 0), c2.live.len());}
test "steady: rebuild is bounded by the watermark (upstream spec 3.4)" { // mirrors upstream TestRebuildLiveTombstones_BoundedByWatermark: the // rebuilt set equals the fold over (compaction/seq, tip] — segments // entirely at or below the watermark contribute nothing, decided from // the header without reading their blocks. 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]});
var a = try archive_mod.Archive.init(testing.allocator, io, data_dir); defer a.deinit(); var meta = try meta_store.Store.open(testing.allocator, data_dir); defer meta.deinit();
// segment A (seqs 1-2): create + covering delete for did:plc:a _ = try a.append(row(.create, "did:plc:a", "r", "x"), 1); _ = try a.append(row(.delete, "did:plc:a", "r", ""), 2); try a.rotate(); // segment B (seq 3): delete marker for did:plc:b _ = try a.append(row(.delete, "did:plc:b", "r", ""), 3); try a.rotate();
// watermark covers all of segment A var wm = watermark.Store.init(&meta); defer wm.deinit(io); try wm.save(io, 2);
// truncate A to its bare header: the rebuild passes only if it decides // "wholly below watermark" from the header alone — any block read of A // now fails, so the block-pruning-after-full-read shape cannot sneak back { var seg_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined; const seg_path = try std.fmt.bufPrint(&seg_dir_buf, "{s}/segments/seg_0000000000.jss", .{data_dir}); var f = try Io.Dir.cwd().openFile(io, seg_path, .{ .mode = .read_write }); defer f.close(io); try f.setLength(io, segment.header_size); }
var c = Compactor.init(testing.allocator, io, &a, &meta, data_dir); defer c.deinit(); try c.rebuild();
// only segment B's marker survives; A's is already physically applied try testing.expectEqual(@as(usize, 1), c.live.len()); var probe_a = row(.create, "did:plc:a", "r", "x"); probe_a.seq = 1; try testing.expectEqual(@as(?tombstone.DropReason, null), c.live.dropReason(probe_a)); var probe_b = row(.create, "did:plc:b", "r", "x"); probe_b.seq = 1; try testing.expectEqual(@as(?tombstone.DropReason, .record), c.live.dropReason(probe_b));}
test "steady: cap sets the coalesced trigger" { 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]});
var a = try archive_mod.Archive.init(testing.allocator, io, data_dir); defer a.deinit(); var meta = try meta_store.Store.open(testing.allocator, data_dir); defer meta.deinit(); var c = Compactor.init(testing.allocator, io, &a, &meta, data_dir); defer c.deinit(); c.tombstone_cap = 2; a.on_append_ctx = &c; a.on_append = Compactor.onAppend;
_ = try a.append(row(.delete, "did:plc:a", "k1", ""), 1); try testing.expect(!c.trigger.load(.acquire)); _ = try a.append(row(.delete, "did:plc:a", "k2", ""), 2); try testing.expect(c.trigger.load(.acquire));}
test "steady: first cap trigger runs an early real pass and resets live gauges" { 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(); var meta = try meta_store.Store.open(testing.allocator, data_dir); defer meta.deinit(); var compactor = Compactor.init(testing.allocator, io, &archive, &meta, data_dir); defer compactor.deinit(); var instrumentation: metrics.Stats = .{}; compactor.stats = &instrumentation; compactor.interval_ns = std.time.ns_per_hour; compactor.tombstone_cap = 1; archive.on_append_ctx = &compactor; archive.on_append = Compactor.onAppend; var future = try io.concurrent(Compactor.run, .{&compactor}); defer { compactor.requestStop(); _ = future.cancel(io); }
const triggered_at = Io.Timestamp.now(io, .awake); _ = try archive.append(row(.delete, "did:plc:a", "k", ""), 1); for (0..3000) |_| { if (instrumentation.compaction_passes_early_total.load(.acquire) == 1) break; try io.sleep(.fromMilliseconds(1), .awake); } try testing.expectEqual(@as(u64, 1), instrumentation.compaction_passes_early_total.load(.acquire)); try testing.expectEqual(@as(u64, 1), instrumentation.compaction_passes.getPtrConst(.ok).load(.acquire)); try testing.expectEqual(@as(usize, 0), instrumentation.compaction_tombstone_set_entries.load(.acquire)); try testing.expect(triggered_at.durationTo(Io.Timestamp.now(io, .awake)).toMilliseconds() < 500);}
test "steady: malformed account marker fails the real archive append" { 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(); var meta = try meta_store.Store.open(testing.allocator, data_dir); defer meta.deinit(); var compactor = Compactor.init(testing.allocator, io, &archive, &meta, data_dir); defer compactor.deinit(); archive.on_append_ctx = &compactor; archive.on_append = Compactor.onAppend;
try testing.expectError(error.UnexpectedEof, archive.append(row(.account, "did:plc:a", "", "\xa1"), 1)); try testing.expectError(error.ArchivePoisoned, archive.append(row(.create, "did:plc:a", "k", "x"), 2));}
test "steady: timestamp rewrite lock excludes the real compaction pass without blocking append" { 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 path_len = 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..path_len]});
var archive = try archive_mod.Archive.init(testing.allocator, io, data_dir); defer archive.deinit(); var meta = try meta_store.Store.open(testing.allocator, data_dir); defer meta.deinit(); var compactor = Compactor.init(testing.allocator, io, &archive, &meta, data_dir); defer compactor.deinit(); archive.on_append_ctx = &compactor; archive.on_append = Compactor.onAppend; _ = try archive.append(row(.create, "did:plc:a", "k1", "x"), 1); _ = try archive.append(row(.delete, "did:plc:a", "k1", ""), 2);
const Task = struct { compactor: *Compactor, started: std.atomic.Value(bool) = .init(false), finished: std.atomic.Value(bool) = .init(false), result: ?pass.Stats = null, failure: ?anyerror = null,
fn run(self: *@This()) void { self.started.store(true, .release); self.result = self.compactor.runOnce() catch |err| { self.failure = err; self.finished.store(true, .release); return; }; self.finished.store(true, .release); } }; var task: Task = .{ .compactor = &compactor }; archive.lockRewrites(); var locked = true; defer if (locked) archive.unlockRewrites(); var future = try io.concurrent(Task.run, .{&task}); while (!task.started.load(.acquire)) try io.sleep(.fromMilliseconds(1), .awake);
// Wait until runOnce's pre-lock force-rotation is observable on disk. At // that point the task can only be waiting on rewrite_mu, making the // exclusion assertion deterministic rather than scheduler-timing based. var sealed_before_append = false; for (0..1000) |_| { var header_bytes: [segment.header_size]u8 = undefined; if (archive.dir.readFile(io, "seg_0000000000.jss", &header_bytes)) |read| { if (read.len == header_bytes.len) if (segment.Header.decode(read)) |_| { sealed_before_append = true; } else |_| {}; } else |_| {} if (sealed_before_append) break; try io.sleep(.fromMilliseconds(1), .awake); } try testing.expect(sealed_before_append);
// The compactor can force-rotate, but cannot enter its mutating pass. Live // append uses the independent writer mutex and remains available. _ = try archive.append(row(.create, "did:plc:a", "k2", "x"), 3); try io.sleep(.fromMilliseconds(20), .awake); try testing.expect(!task.finished.load(.acquire)); var wm = watermark.Store.init(&meta); defer wm.deinit(io); try testing.expectEqual(@as(?u64, null), try wm.load(io));
locked = false; archive.unlockRewrites(); _ = future.await(io); if (task.failure) |err| return err; try testing.expect(task.finished.load(.acquire)); try testing.expectEqual(@as(u64, 2), task.result.?.target_watermark); try testing.expectEqual(@as(?u64, 2), try wm.load(io));}