//! 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));
}