//! archive — disk-backed jss segments with rotation, sealing, and recovery
//!
//! owns <data-dir>/segments: assigns instance-local seqs (starting at 1;
//! 0 is the "before beginning" sentinel), appends events to an active
//! segment, flushes blocks to disk with fsync, seals + rotates at a byte
//! budget, and on startup recovers by truncating any torn tail and sealing
//! the leftover active file (upstream's recovery-scan lesson:
//! docs/lessons-from-zlay.md #3 — validate on recovery, truncate torn
//! tails, never trust an unvalidated record).
//!
//! durability watermark: committed_seq is the last local seq whose block
//! is fsynced. anything above it exists only in memory. sealed headers and
//! block indexes are held by the same refreshable resident manifest as
//! upstream; files remain the source of truth for frame bytes.

const std = @import("std");
const segment = @import("segment.zig");
const writer_mod = @import("segment_writer.zig");
const metrics = @import("../runtime/metrics.zig");
const observability = @import("../runtime/observability.zig");
const manifest_mod = @import("manifest.zig");
const segment_io = @import("segment_io.zig");

const Io = std.Io;
const Allocator = std.mem.Allocator;
const log = std.log.scoped(.stream);

pub const default_max_segment_bytes: usize = 256 * 1024 * 1024;

const AsyncFlushJob = struct {
    id: u64 = 0,
    prepared: *writer_mod.PreparedBlock,
    frame: ?[]u8 = null,
    failure: ?anyerror = null,
    compressed: Io.Event = .unset,
};

const AsyncFlushQueue = struct {
    buffer: []*AsyncFlushJob,
    head: usize = 0,
    len: usize = 0,
    closed: bool = false,
    mu: Io.Mutex = .init,
    not_empty: Io.Condition = .init,
    not_full: Io.Condition = .init,

    fn put(self: *AsyncFlushQueue, io: Io, job: *AsyncFlushJob) void {
        self.mu.lockUncancelable(io);
        defer self.mu.unlock(io);
        while (self.len == self.buffer.len and !self.closed)
            self.not_full.waitUncancelable(io, &self.mu);
        if (self.closed) unreachable;
        self.buffer[(self.head + self.len) % self.buffer.len] = job;
        self.len += 1;
        self.not_empty.signal(io);
    }

    fn pop(self: *AsyncFlushQueue, io: Io) !*AsyncFlushJob {
        self.mu.lockUncancelable(io);
        defer self.mu.unlock(io);
        while (self.len == 0 and !self.closed)
            try self.not_empty.wait(io, &self.mu);
        if (self.len == 0) return error.Closed;
        const job = self.buffer[self.head];
        self.head = (self.head + 1) % self.buffer.len;
        self.len -= 1;
        self.not_full.signal(io);
        return job;
    }

    fn close(self: *AsyncFlushQueue, io: Io) void {
        self.mu.lockUncancelable(io);
        defer self.mu.unlock(io);
        self.closed = true;
        self.not_empty.broadcast(io);
        self.not_full.broadcast(io);
    }
};

const AsyncFlush = struct {
    allocator: Allocator,
    io: Io,
    queue: AsyncFlushQueue,
    group: Io.Group = .init,
    next_job_id: u64 = 0,
    next_commit_id: u64 = 0,
    commit_mu: Io.Mutex = .init,
    commit_ready: Io.Condition = .init,
    admission_mu: Io.Mutex = .init,
    outstanding_mu: Io.Mutex = .init,
    outstanding_ready: Io.Condition = .init,
    outstanding: usize = 0,
    workers_active: std.atomic.Value(usize) = .init(0),
    workers_peak: std.atomic.Value(usize) = .init(0),

    fn init(allocator: Allocator, io: Io, worker_count: usize) !AsyncFlush {
        std.debug.assert(worker_count > 0);
        return .{
            .allocator = allocator,
            .io = io,
            .queue = .{ .buffer = try allocator.alloc(*AsyncFlushJob, worker_count) },
        };
    }

    fn start(self: *AsyncFlush, worker_count: usize) !void {
        errdefer {
            self.queue.close(self.io);
            self.group.cancel(self.io);
        }
        for (0..worker_count) |_| try self.group.concurrent(self.io, workerLoop, .{self});
    }

    fn deinit(self: *AsyncFlush) void {
        self.queue.close(self.io);
        self.group.await(self.io) catch {};
        self.allocator.free(self.queue.buffer);
    }

    fn assign(self: *AsyncFlush, job: *AsyncFlushJob) void {
        job.id = self.next_job_id;
        self.next_job_id += 1;
        self.outstanding_mu.lockUncancelable(self.io);
        self.outstanding += 1;
        self.outstanding_mu.unlock(self.io);
    }

    fn submit(self: *AsyncFlush, job: *AsyncFlushJob) void {
        self.queue.put(self.io, job);
    }

    fn workerLoop(self: *AsyncFlush) void {
        while (self.queue.pop(self.io)) |job| {
            const active = self.workers_active.fetchAdd(1, .acq_rel) + 1;
            var peak = self.workers_peak.load(.acquire);
            while (active > peak) {
                peak = self.workers_peak.cmpxchgWeak(peak, active, .acq_rel, .acquire) orelse break;
            }
            job.frame = writer_mod.compressPrepared(self.allocator, job.prepared) catch |err| blk: {
                job.failure = err;
                break :blk null;
            };
            _ = self.workers_active.fetchSub(1, .acq_rel);
            job.compressed.set(self.io);
        } else |_| {}
    }

    fn waitTurn(self: *AsyncFlush, id: u64) void {
        self.commit_mu.lockUncancelable(self.io);
        while (self.next_commit_id != id)
            self.commit_ready.waitUncancelable(self.io, &self.commit_mu);
    }

    fn finishTurn(self: *AsyncFlush) void {
        self.next_commit_id += 1;
        self.commit_ready.broadcast(self.io);
        self.commit_mu.unlock(self.io);
        self.outstanding_mu.lockUncancelable(self.io);
        self.outstanding -= 1;
        self.outstanding_ready.broadcast(self.io);
        self.outstanding_mu.unlock(self.io);
    }

    fn drain(self: *AsyncFlush) void {
        self.outstanding_mu.lockUncancelable(self.io);
        defer self.outstanding_mu.unlock(self.io);
        while (self.outstanding != 0)
            self.outstanding_ready.waitUncancelable(self.io, &self.outstanding_mu);
    }
};

pub const Archive = struct {
    allocator: Allocator,
    io: Io,
    dir: Io.Dir,
    manifest: manifest_mod.Manifest,
    writer: writer_mod.ActiveWriter,
    file: Io.File,
    /// bytes of writer.buf already written to disk
    file_written: usize = 0,
    /// Number of compressed blocks included in the last successful fsync +
    /// metadata-hook boundary. Metrics advance from this exact watermark.
    synced_block_count: u32 = 0,
    seg_index: u64 = 0,
    next_seq: u64 = 1,
    max_segment_bytes: usize = default_max_segment_bytes,
    max_events_per_block: usize = writer_mod.default_max_events_per_block,
    /// last local seq whose block is durable (fsynced)
    committed_seq: std.atomic.Value(u64) = .init(0),
    /// wall clock µs of the last disk sync; partial blocks flush after
    /// flush_interval_us so quiet periods still become durable (upstream:
    /// blocks flush at 4096 events OR 30s)
    last_sync_us: i64 = 0,
    flush_interval_us: i64 = 30 * std.time.us_per_s,
    /// serializes the writer against the steady compactor's force-rotate;
    /// uncontended in the single-writer steady state
    mu: Io.Mutex = .init,
    /// Serializes every in-place sealed-segment rewrite. Delete compaction and
    /// timestamp import both take this lock around their mutating pass so two
    /// tmp+fsync+rename transactions can never overwrite one another's result.
    /// It is deliberately separate from `mu`: live appends continue while a
    /// cold segment pass is running.
    rewrite_mu: Io.Mutex = .init,
    /// upstream's OnAppend hook: runs under `mu` with the stamped event,
    /// so a force-rotate returns only after every sealed row was observed
    /// (the live-tombstone ordering guarantee)
    on_append_ctx: ?*anyopaque = null,
    on_append: ?*const fn (*anyopaque, segment.Event) anyerror!void = null,
    /// Upstream's timestamp Stamper: called under `mu`, before sequence
    /// assignment and buffering. It mutates the caller's row so archive bytes
    /// and immediate subscriber delivery use the identical display timestamp.
    stamper_ctx: ?*anyopaque = null,
    stamper: ?*const fn (*anyopaque, *segment.Event) anyerror!void = null,
    /// Called after segment bytes are fsynced but before committed_seq is
    /// published. The hook must synchronously commit metadata describing
    /// those bytes; failure keeps the durability watermark behind them.
    on_durable_ctx: ?*anyopaque = null,
    on_durable: ?*const fn (*anyopaque, u64, u64) anyerror!void = null,
    stats: ?*metrics.Stats = null,
    /// Bootstrap-only ordered compression pipeline. Steady-state writes stay
    /// synchronous, matching upstream's BackfillAsyncFlushWorkers scope.
    async_flush: ?AsyncFlush = null,
    /// A failed append/flush may leave a partial in-memory transaction. Never
    /// attempt a graceful flush afterward; recovery trusts only complete
    /// frames already on disk and truncates a torn suffix.
    poisoned: bool = false,

    pub const ColdSnapshot = struct {
        allocator: Allocator,
        summaries: []manifest_mod.Summary,
        active_index: u64,
        active_length: usize,
        active_start: usize,

        pub fn deinit(self: *ColdSnapshot) void {
            self.allocator.free(self.summaries);
            self.* = undefined;
        }
    };

    /// Atomically pair the sealed manifest with the active segment's durable
    /// byte boundary. A later rotation is harmless: the old active filename
    /// becomes that same index's sealed filename, while it was not yet present
    /// in `summaries` and the captured prefix remains byte-stable.
    pub fn coldSnapshot(self: *Archive, allocator: Allocator, from_seq: u64) !ColdSnapshot {
        self.mu.lockUncancelable(self.io);
        defer self.mu.unlock(self.io);
        const summaries = try self.manifest.all(allocator);
        errdefer allocator.free(summaries);
        // Prepared async blocks have no file offsets yet; committed blocks
        // may still exceed the metadata-accepted fsync boundary.
        const blocks = self.writer.builder.infos.items[0..self.writer.block_count];
        var lo: usize = 0;
        var hi: usize = blocks.len;
        while (lo < hi) {
            const mid = lo + (hi - lo) / 2;
            if (blocks[mid].max_seq < from_seq) lo = mid + 1 else hi = mid;
        }
        const start = if (lo < blocks.len) @min(blocks[lo].offset, self.file_written) else self.file_written;
        return .{
            .allocator = allocator,
            .summaries = summaries,
            .active_index = self.seg_index,
            .active_length = self.file_written,
            .active_start = @intCast(start),
        };
    }

    pub fn init(allocator: Allocator, io: Io, data_dir: []const u8) !Archive {
        return initWithStats(allocator, io, data_dir, null);
    }

    pub fn initWithStats(allocator: Allocator, io: Io, data_dir: []const u8, stats: ?*metrics.Stats) !Archive {
        var buf: [256]u8 = undefined;
        const seg_path = try std.fmt.bufPrint(&buf, "{s}/segments", .{data_dir});
        var dir = try Io.Dir.cwd().createDirPathOpen(io, seg_path, .{ .open_options = .{ .iterate = true } });
        errdefer dir.close(io);

        var a: Archive = .{
            .allocator = allocator,
            .io = io,
            .dir = dir,
            .manifest = undefined,
            .writer = undefined,
            .file = undefined,
            .stats = stats,
        };
        const resumed_active = try a.recover();
        errdefer if (resumed_active) {
            a.writer.deinit();
            a.file.close(io);
        };
        // Recovery only advances next_seq from bytes that are already sealed
        // and fsynced. Publish that durable floor immediately; otherwise the
        // watermark (and /status) falsely reports zero until the first new
        // block happens to flush after a restart.
        a.committed_seq.store(a.next_seq - 1, .release);
        a.manifest = try manifest_mod.Manifest.init(allocator, io, dir, stats);
        errdefer a.manifest.deinit();
        if (!resumed_active) try a.openNextSegment();
        return a;
    }

    pub fn deinit(self: *Archive) void {
        if (self.async_flush) |*pipeline| {
            pipeline.admission_mu.lockUncancelable(self.io);
            pipeline.drain();
            pipeline.deinit();
            pipeline.admission_mu.unlock(self.io);
            self.async_flush = null;
        }
        self.writer.deinit();
        self.file.close(self.io);
        self.manifest.deinit();
        self.dir.close(self.io);
    }

    /// Enable upstream's bootstrap-only async compression path. A zero worker
    /// count is the explicit synchronous mode.
    pub fn enableAsyncFlush(self: *Archive, worker_count: usize) !void {
        if (worker_count == 0) return;
        if (self.async_flush != null) return error.AsyncFlushAlreadyEnabled;
        self.async_flush = try AsyncFlush.init(self.allocator, self.io, worker_count);
        errdefer {
            self.async_flush.?.deinit();
            self.async_flush = null;
        }
        try self.async_flush.?.start(worker_count);
    }

    /// Drain and remove the bootstrap pipeline before merge/steady state.
    pub fn disableAsyncFlush(self: *Archive) void {
        if (self.async_flush) |*pipeline| {
            pipeline.admission_mu.lockUncancelable(self.io);
            pipeline.drain();
            pipeline.deinit();
            pipeline.admission_mu.unlock(self.io);
            self.async_flush = null;
        }
    }

    /// Scan the directory, recover the durable seq floor, and resume the
    /// highest active segment in place. Returns true when `writer` and `file`
    /// were initialized from that active segment.
    fn recover(self: *Archive) !bool {
        var max_index: ?u64 = null;
        var it = self.dir.iterate();
        while (try it.next(self.io)) |entry| {
            const idx = parseSegmentIndex(entry.name) orelse continue;
            if (max_index == null or idx > max_index.?) max_index = idx;
        }
        const candidate = max_index orelse return false;
        var name_buf: [64]u8 = undefined;
        const name = formatSegmentName(&name_buf, candidate);
        const bytes = try self.dir.readFileAlloc(self.io, name, self.allocator, .limited(1 << 31));
        defer self.allocator.free(bytes);
        if (bytes.len < segment.header_size) return error.Truncated;

        if (segment.Header.decode(bytes)) |_| {
            const max_seq = try self.sealedMaxSeq(candidate, bytes);
            self.next_seq = @max(self.next_seq, max_seq + 1);
            self.seg_index = candidate + 1;
            return false;
        } else |err| switch (err) {
            error.ActiveSegment => {},
            else => return err,
        }

        var valid_end: usize = segment.header_size;
        var max_seq: u64 = 0;
        var event_count: u64 = 0;
        var block_count: u32 = 0;
        var torn = false;
        var walk: writer_mod.ActiveIterator = .{ .bytes = bytes };
        while (true) {
            var block = (walk.next(self.allocator) catch |err| switch (err) {
                error.Truncated => {
                    torn = true;
                    break;
                },
                else => return err,
            }) orelse break;
            defer block.deinit(self.allocator);
            if (block.events.len == 0) break;
            valid_end = walk.offset;
            event_count += block.events.len;
            block_count += 1;
            for (block.events) |event| max_seq = @max(max_seq, event.seq);
        }
        if (max_seq == 0 and candidate > 0)
            max_seq = try self.sealedFloorBefore(candidate);

        var writer = try writer_mod.ActiveWriter.init(self.allocator);
        errdefer writer.deinit();
        writer.buf.clearRetainingCapacity();
        try writer.buf.appendSlice(self.allocator, bytes[0..valid_end]);
        writer.event_count = event_count;
        writer.block_count = block_count;
        try writer.rebuildIndex();

        var file = try self.dir.openFile(self.io, name, .{ .mode = .read_write });
        errdefer file.close(self.io);
        if (torn or valid_end != bytes.len) {
            try file.setLength(self.io, valid_end);
            try segment_io.sync(&file, self.io);
        }

        self.writer = writer;
        self.file = file;
        self.file_written = valid_end;
        self.synced_block_count = block_count;
        self.seg_index = candidate;
        self.next_seq = @max(self.next_seq, max_seq + 1);
        if (self.stats) |stats|
            stats.ingest_active_segment_bytes.store(valid_end - segment.header_size, .monotonic);
        log.info("recovery: resumed active segment {s} at {d} valid bytes", .{ name, valid_end });
        return true;
    }

    fn sealedMaxSeq(self: *Archive, candidate: u64, candidate_bytes: []const u8) !u64 {
        var sealed = try segment.Sealed.parse(self.allocator, candidate_bytes);
        defer sealed.deinit(self.allocator);
        if (sealed.header.max_seq != 0) return sealed.header.max_seq;
        if (candidate == 0) return 0;
        return self.sealedFloorBefore(candidate);
    }

    fn sealedFloorBefore(self: *Archive, before: u64) !u64 {
        var candidate = before;
        while (candidate > 0) {
            candidate -= 1;
            var name_buf: [64]u8 = undefined;
            const name = formatSegmentName(&name_buf, candidate);
            const bytes = try self.dir.readFileAlloc(self.io, name, self.allocator, .limited(1 << 31));
            defer self.allocator.free(bytes);
            var sealed = try segment.Sealed.parse(self.allocator, bytes);
            defer sealed.deinit(self.allocator);
            if (sealed.header.max_seq != 0) return sealed.header.max_seq;
        }
        return 0;
    }

    fn openNextSegment(self: *Archive) !void {
        self.writer = try writer_mod.ActiveWriter.init(self.allocator);
        errdefer self.writer.deinit();
        // the writer owns block granularity (it also sizes per-block blooms
        // from this at seal); we sync to disk whenever the buffer grows
        self.writer.max_events_per_block = self.max_events_per_block;
        var name_buf: [64]u8 = undefined;
        const name = formatSegmentName(&name_buf, self.seg_index);
        self.file = try self.dir.createFile(self.io, name, .{ .truncate = true });
        var file_open = true;
        errdefer {
            if (file_open) self.file.close(self.io);
            self.dir.deleteFile(self.io, name) catch {};
        }
        try segment_io.writeStreamingAll(&self.file, self.io, self.writer.bytes());
        try segment_io.sync(&self.file, self.io);
        // The header is not durable if a power cut can still forget the new
        // directory entry. Upstream syncs the parent after writer creation.
        try segment_io.syncDir(self.dir, self.io);
        file_open = false;
        self.file_written = self.writer.bytes().len;
        self.synced_block_count = 0;
        if (self.stats) |stats| stats.ingest_active_segment_bytes.store(0, .monotonic);
    }

    /// append one event; assigns and returns its local seq. blocks flush to
    /// disk at the block threshold; rotation seals at the byte budget.
    pub fn append(self: *Archive, ev: segment.Event, now_us: i64) !u64 {
        return self.appendTraced(ev, now_us, observability.root_context);
    }

    pub fn appendTraced(
        self: *Archive,
        ev: segment.Event,
        now_us: i64,
        trace_context: observability.Context,
    ) !u64 {
        var rows = [_]segment.Event{ev};
        return (try self.appendBatchTraced(&rows, now_us, trace_context)).?;
    }

    /// Append every row derived from one upstream event under one writer
    /// lock. This is the transaction boundary relied on by frame seq
    /// assignment (`first_seq + op_index`) and verifier checkpoints.
    pub fn appendBatch(self: *Archive, events: []segment.Event, now_us: i64) !?u64 {
        return self.appendBatchTraced(events, now_us, observability.root_context);
    }

    pub fn appendBatchTraced(
        self: *Archive,
        events: []segment.Event,
        now_us: i64,
        trace_context: observability.Context,
    ) !?u64 {
        return self.appendBatchBeforeDurableTraced(events, now_us, null, null, trace_context);
    }

    /// Append rows and run `before_durable(first,last)` under the writer lock
    /// before this append is allowed to trigger an fsync. The ingest writer
    /// uses this to register cursor/verifier checkpoints against assigned
    /// sequence numbers, closing the rows-before-metadata crash window.
    pub fn appendBatchBeforeDurable(
        self: *Archive,
        events: []segment.Event,
        now_us: i64,
        hook_ctx: ?*anyopaque,
        before_durable: ?*const fn (*anyopaque, u64, u64) anyerror!void,
    ) !?u64 {
        return self.appendBatchBeforeDurableTraced(
            events,
            now_us,
            hook_ctx,
            before_durable,
            observability.root_context,
        );
    }

    pub fn appendBatchBeforeDurableTraced(
        self: *Archive,
        events: []segment.Event,
        now_us: i64,
        hook_ctx: ?*anyopaque,
        before_durable: ?*const fn (*anyopaque, u64, u64) anyerror!void,
        trace_context: observability.Context,
    ) !?u64 {
        if (events.len == 0) return null;
        if (self.async_flush != null)
            return @as(?u64, try self.appendBatchAsync(events, now_us, hook_ctx, before_durable, trace_context));
        return @as(?u64, try self.appendBatchSync(events, now_us, hook_ctx, before_durable, trace_context));
    }

    fn appendBatchSync(
        self: *Archive,
        events: []segment.Event,
        now_us: i64,
        hook_ctx: ?*anyopaque,
        before_durable: ?*const fn (*anyopaque, u64, u64) anyerror!void,
        trace_context: observability.Context,
    ) !u64 {
        var flush_trace: ?observability.Span = null;
        defer if (flush_trace) |*trace| trace.deinit();
        const result = self.appendBatchSyncLocked(events, now_us, hook_ctx, before_durable, trace_context, &flush_trace);
        if (flush_trace) |*trace| return trace.recordError(result);
        return result;
    }

    fn appendBatchSyncLocked(
        self: *Archive,
        events: []segment.Event,
        now_us: i64,
        hook_ctx: ?*anyopaque,
        before_durable: ?*const fn (*anyopaque, u64, u64) anyerror!void,
        trace_context: observability.Context,
        flush_trace: *?observability.Span,
    ) !u64 {
        self.mu.lockUncancelable(self.io);
        defer self.mu.unlock(self.io);
        errdefer self.poisoned = true;
        if (self.poisoned) return error.ArchivePoisoned;
        const first_seq = self.next_seq;
        for (events) |*ev| {
            _ = try self.appendStampedLocked(ev);
        }
        if (before_durable) |hook| try hook(hook_ctx.?, first_seq, self.next_seq - 1);
        if (self.writer.pending.items.len > 0 and now_us - self.last_sync_us >= self.flush_interval_us) {
            flush_trace.* = observability.Span.start(
                self.allocator,
                "ingest",
                "flushAndRotateLocked",
                trace_context,
                &.{},
            );
            try self.writer.flush(); // partial block: durability over block size
        }
        if (self.writer.bytes().len > self.file_written) {
            if (flush_trace.* == null)
                flush_trace.* = observability.Span.start(
                    self.allocator,
                    "ingest",
                    "flushAndRotateLocked",
                    trace_context,
                    &.{},
                );
            try self.syncToDisk(null);
            self.last_sync_us = now_us;
            if (self.writer.bytes().len >= self.max_segment_bytes)
                try self.rotateLockedTraced(flush_trace.*.?.context());
        }
        if (flush_trace.*) |*trace| trace.succeed();
        return first_seq;
    }

    fn appendBatchAsync(
        self: *Archive,
        events: []segment.Event,
        now_us: i64,
        hook_ctx: ?*anyopaque,
        before_durable: ?*const fn (*anyopaque, u64, u64) anyerror!void,
        trace_context: observability.Context,
    ) !u64 {
        const pipeline = &self.async_flush.?;
        pipeline.admission_mu.lockUncancelable(self.io);
        var admission_locked = true;
        defer if (admission_locked) pipeline.admission_mu.unlock(self.io);

        var prepared: std.ArrayList(writer_mod.PreparedBlock) = .empty;
        defer {
            for (prepared.items) |*block| block.deinit(self.allocator);
            prepared.deinit(self.allocator);
        }

        self.mu.lockUncancelable(self.io);
        var writer_locked = true;
        defer if (writer_locked) self.mu.unlock(self.io);
        errdefer self.poisoned = true;
        if (self.poisoned) return error.ArchivePoisoned;
        const first_seq = self.next_seq;
        for (events) |*ev| {
            if (try self.appendStampedPreparedLocked(ev)) |block|
                try prepared.append(self.allocator, block);
        }
        if (before_durable) |hook| try hook(hook_ctx.?, first_seq, self.next_seq - 1);
        if (self.writer.pending.items.len > 0 and now_us - self.last_sync_us >= self.flush_interval_us) {
            if (try self.writer.prepareFlush()) |block| try prepared.append(self.allocator, block);
        }

        const jobs = try self.allocator.alloc(AsyncFlushJob, prepared.items.len);
        defer self.allocator.free(jobs);
        for (jobs, prepared.items) |*job, *block| {
            job.* = .{ .prepared = block };
            pipeline.assign(job);
        }
        self.mu.unlock(self.io);
        writer_locked = false;

        for (jobs) |*job| pipeline.submit(job);
        pipeline.admission_mu.unlock(self.io);
        admission_locked = false;

        try self.finishAsyncJobs(jobs, trace_context);
        if (jobs.len > 0) {
            self.mu.lockUncancelable(self.io);
            self.last_sync_us = now_us;
            self.mu.unlock(self.io);
        }
        try self.rotateIfFullAsync(trace_context);
        return first_seq;
    }

    fn finishAsyncJobs(self: *Archive, jobs: []AsyncFlushJob, trace_context: observability.Context) !void {
        const pipeline = &self.async_flush.?;
        var first_error: ?anyerror = null;
        for (jobs) |*job| {
            job.compressed.waitUncancelable(self.io);
            pipeline.waitTurn(job.id);
            var trace = observability.Span.start(
                self.allocator,
                "ingest",
                "commitAsyncFlush",
                trace_context,
                &.{},
            );
            defer trace.deinit();
            self.mu.lockUncancelable(self.io);
            if (job.failure) |err| {
                trace.fail(err);
                self.poisoned = true;
                if (first_error == null) first_error = err;
            } else if (self.poisoned) {
                trace.fail(error.ArchivePoisoned);
                if (first_error == null) first_error = error.ArchivePoisoned;
            } else {
                self.writer.commitPrepared(job.frame.?) catch |err| {
                    trace.fail(err);
                    self.poisoned = true;
                    if (first_error == null) first_error = err;
                };
                if (!self.poisoned) self.syncToDisk(job.prepared.max_seq) catch |err| {
                    trace.fail(err);
                    self.poisoned = true;
                    if (first_error == null) first_error = err;
                };
            }
            self.mu.unlock(self.io);
            if (job.frame) |frame| self.allocator.free(frame);
            pipeline.finishTurn();
            trace.succeed();
        }
        if (first_error) |err| return err;
    }

    fn rotateIfFullAsync(self: *Archive, trace_context: observability.Context) !void {
        const pipeline = &self.async_flush.?;
        self.mu.lockUncancelable(self.io);
        const full = self.writer.bytes().len >= self.max_segment_bytes;
        self.mu.unlock(self.io);
        if (!full) return;

        var trace = observability.Span.start(self.allocator, "ingest", "rotateIfFull", trace_context, &.{});
        defer trace.deinit();
        defer trace.succeed();
        try trace.recordError(self.rotateDrained(pipeline, trace.context()));
    }

    fn rotateDrained(self: *Archive, pipeline: *AsyncFlush, trace_context: observability.Context) !void {
        pipeline.admission_mu.lockUncancelable(self.io);
        defer pipeline.admission_mu.unlock(self.io);
        pipeline.drain();
        self.mu.lockUncancelable(self.io);
        defer self.mu.unlock(self.io);
        errdefer self.poisoned = true;
        if (self.poisoned) return error.ArchivePoisoned;
        if (self.writer.bytes().len >= self.max_segment_bytes) try self.rotateLockedTraced(trace_context);
    }

    fn appendStampedLocked(self: *Archive, event: *segment.Event) !u64 {
        if (self.stamper) |stamp| stamp(self.stamper_ctx.?, event) catch |err| {
            if (self.stats) |stats| _ = stats.ingest_append_errors_total.fetchAdd(1, .monotonic);
            return err;
        };
        const seq = self.next_seq;
        event.seq = seq;
        self.writer.append(event.*) catch |err| {
            if (self.stats) |stats| _ = stats.ingest_append_errors_total.fetchAdd(1, .monotonic);
            return err;
        };
        self.next_seq += 1;
        if (self.stats) |stats| _ = stats.ingest_events_appended_total.fetchAdd(1, .monotonic);
        if (self.on_append) |hook| try hook(self.on_append_ctx.?, event.*);
        return seq;
    }

    fn appendStampedPreparedLocked(self: *Archive, event: *segment.Event) !?writer_mod.PreparedBlock {
        if (self.stamper) |stamp| stamp(self.stamper_ctx.?, event) catch |err| {
            if (self.stats) |stats| _ = stats.ingest_append_errors_total.fetchAdd(1, .monotonic);
            return err;
        };
        const seq = self.next_seq;
        event.seq = seq;
        const prepared = self.writer.appendPrepared(event.*) catch |err| {
            if (self.stats) |stats| _ = stats.ingest_append_errors_total.fetchAdd(1, .monotonic);
            return err;
        };
        self.next_seq += 1;
        if (self.stats) |stats| _ = stats.ingest_events_appended_total.fetchAdd(1, .monotonic);
        if (self.on_append) |hook| try hook(self.on_append_ctx.?, event.*);
        return prepared;
    }

    pub fn lastSeq(self: *Archive) u64 {
        self.mu.lockUncancelable(self.io);
        defer self.mu.unlock(self.io);
        return self.next_seq - 1;
    }

    /// Heal resident metadata after an atomic segment rewrite. Returns the
    /// number of missing or checksum-mismatched entries refreshed from disk.
    pub fn reconcileManifest(self: *Archive) !u64 {
        var reconciled: u64 = 0;
        var it = self.dir.iterate();
        while (try it.next(self.io)) |entry| {
            const idx = parseSegmentIndex(entry.name) orelse continue;
            var file = try self.dir.openFile(self.io, entry.name, .{});
            defer file.close(self.io);
            const header = segment.readHeaderFile(self.io, &file) catch |err| switch (err) {
                error.ActiveSegment => continue,
                else => return err,
            };
            if (self.manifest.segmentByIndex(idx)) |resident| {
                if (resident.header.checksum == header.checksum) continue;
            }
            try self.manifest.refresh(self.dir, idx, true);
            reconciled += 1;
        }
        return reconciled;
    }

    pub fn lockRewrites(self: *Archive) void {
        self.rewrite_mu.lockUncancelable(self.io);
    }

    pub fn unlockRewrites(self: *Archive) void {
        self.rewrite_mu.unlock(self.io);
    }

    /// compress + write any pending block, fsync, advance the watermark.
    pub fn flushBlock(self: *Archive) !void {
        return self.flushBlockTraced(observability.root_context);
    }

    pub fn flushBlockTraced(self: *Archive, trace_context: observability.Context) !void {
        if (self.async_flush != null) return self.flushBlockAsync(trace_context);
        var trace = observability.Span.start(self.allocator, "ingest", "flushAndRotateLocked", trace_context, &.{});
        defer trace.deinit();
        defer trace.succeed();
        try trace.recordError(self.flushBlockSync());
    }

    fn flushBlockSync(self: *Archive) !void {
        self.mu.lockUncancelable(self.io);
        defer self.mu.unlock(self.io);
        errdefer self.poisoned = true;
        try self.flushBlockLocked();
    }

    fn flushBlockAsync(self: *Archive, trace_context: observability.Context) !void {
        const pipeline = &self.async_flush.?;
        pipeline.admission_mu.lockUncancelable(self.io);
        var admission_locked = true;
        defer if (admission_locked) pipeline.admission_mu.unlock(self.io);

        self.mu.lockUncancelable(self.io);
        var writer_locked = true;
        defer if (writer_locked) self.mu.unlock(self.io);
        errdefer self.poisoned = true;
        if (self.poisoned) return error.ArchivePoisoned;
        var prepared = (try self.writer.prepareFlush()) orelse return;
        defer prepared.deinit(self.allocator);
        var job: AsyncFlushJob = .{ .prepared = &prepared };
        pipeline.assign(&job);
        self.mu.unlock(self.io);
        writer_locked = false;

        pipeline.submit(&job);
        pipeline.admission_mu.unlock(self.io);
        admission_locked = false;
        try self.finishAsyncJobs((&job)[0..1], trace_context);
    }

    fn flushBlockLocked(self: *Archive) !void {
        try self.writer.flush();
        if (self.writer.bytes().len > self.file_written) try self.syncToDisk(null);
    }

    fn syncToDisk(self: *Archive, async_durable_seq: ?u64) !void {
        const all = self.writer.bytes();
        try segment_io.writePositionalAll(&self.file, self.io, all[self.file_written..], self.file_written);
        try segment_io.sync(&self.file, self.io);
        if (self.stats) |stats| {
            const flushed: u64 = self.writer.block_count - self.synced_block_count;
            if (flushed > 0) _ = stats.ingest_blocks_flushed_total.fetchAdd(flushed, .monotonic);
            stats.ingest_active_segment_bytes.store(all.len - segment.header_size, .monotonic);
        }
        self.synced_block_count = self.writer.block_count;
        // everything up to (not including) pending is now durable
        const durable_seq = async_durable_seq orelse self.next_seq - 1 - self.writer.pending.items.len;
        if (self.on_durable) |hook| try hook(self.on_durable_ctx.?, durable_seq, durable_seq + 1);
        // Advance only after metadata accepts the same boundary. If the hook
        // fails, the next flush safely rewrites and re-syncs this suffix.
        self.file_written = all.len;
        self.committed_seq.store(durable_seq, .release);
    }

    /// seal the active segment (footer + finalized header) and open the
    /// next. safe from another thread (the compactor's force-rotate).
    pub fn rotate(self: *Archive) !void {
        if (self.async_flush) |*pipeline| {
            pipeline.admission_mu.lockUncancelable(self.io);
            defer pipeline.admission_mu.unlock(self.io);
            pipeline.drain();
        }
        self.mu.lockUncancelable(self.io);
        defer self.mu.unlock(self.io);
        errdefer self.poisoned = true;
        try self.rotateLocked();
    }

    pub fn rotateTraced(self: *Archive, trace_context: observability.Context) !void {
        if (self.async_flush) |*pipeline| {
            pipeline.admission_mu.lockUncancelable(self.io);
            defer pipeline.admission_mu.unlock(self.io);
            pipeline.drain();
        }
        self.mu.lockUncancelable(self.io);
        defer self.mu.unlock(self.io);
        errdefer self.poisoned = true;
        try self.rotateLockedTraced(trace_context);
    }

    fn rotateLocked(self: *Archive) !void {
        const seal_start = Io.Timestamp.now(self.io, .awake);
        try self.flushBlockLocked();
        if (self.writer.block_count == 0) return; // nothing to seal
        try self.writer.seal();
        const all = self.writer.bytes();
        try segment_io.writePositionalAll(&self.file, self.io, all[self.file_written..], self.file_written);
        try segment_io.sync(&self.file, self.io);
        try segment_io.writePositionalAll(&self.file, self.io, all[0..segment.header_size], 0);
        try segment_io.sync(&self.file, self.io);
        try self.manifest.refresh(self.dir, self.seg_index, false);
        self.file.close(self.io);
        self.writer.deinit();
        if (self.stats) |stats| {
            const elapsed_us = seal_start.durationTo(Io.Timestamp.now(self.io, .awake)).toMicroseconds();
            stats.observeSegmentSeal(@intCast(@max(0, elapsed_us)));
            _ = stats.ingest_segments_rotated_total.fetchAdd(1, .monotonic);
        }
        self.seg_index += 1;
        try self.openNextSegment();
    }

    fn rotateLockedTraced(self: *Archive, trace_context: observability.Context) !void {
        var trace = observability.Span.start(self.allocator, "ingest", "rotateLocked", trace_context, &.{});
        defer trace.deinit();
        defer trace.succeed();
        try trace.recordError(self.rotateLocked());
        trace.setAttribute(.{ .key = "active_idx", .value = .{ .int = @intCast(self.seg_index) } });
    }

    /// terminal cutover (bootstrap-live source at merge): seal the active
    /// segment so the merge drain can read it; an empty active (header
    /// only, no blocks) is deleted instead. the archive is unusable after.
    pub fn sealAndClose(self: *Archive) !void {
        self.mu.lockUncancelable(self.io);
        defer self.mu.unlock(self.io);
        errdefer self.poisoned = true;
        // Cross the ordinary fsync + metadata hook before sealing. Bypassing
        // syncToDisk here would make the final bootstrap-live block durable
        // without its relay cursor and promoted verifier state.
        const seal_start = Io.Timestamp.now(self.io, .awake);
        try self.flushBlockLocked();
        if (self.writer.block_count == 0) {
            var name_buf: [64]u8 = undefined;
            const name = formatSegmentName(&name_buf, self.seg_index);
            self.file.close(self.io);
            self.dir.deleteFile(self.io, name) catch {};
        } else {
            try self.writer.seal();
            const all = self.writer.bytes();
            try segment_io.writePositionalAll(&self.file, self.io, all[self.file_written..], self.file_written);
            try segment_io.sync(&self.file, self.io);
            try segment_io.writePositionalAll(&self.file, self.io, all[0..segment.header_size], 0);
            try segment_io.sync(&self.file, self.io);
            try self.manifest.refresh(self.dir, self.seg_index, false);
            self.file.close(self.io);
            if (self.stats) |stats| {
                const elapsed_us = seal_start.durationTo(Io.Timestamp.now(self.io, .awake)).toMicroseconds();
                stats.observeSegmentSeal(@intCast(@max(0, elapsed_us)));
            }
        }
        self.writer.deinit();
        self.manifest.deinit();
        self.dir.close(self.io);
    }

    /// graceful shutdown: flush what's pending (the segment stays active —
    /// upstream's steady-state shutdown doesn't seal either; recovery
    /// handles it next start).
    pub fn close(self: *Archive) void {
        if (self.poisoned) {
            log.err("archive is poisoned; refusing shutdown flush (recovery will truncate to the durable prefix)", .{});
            return;
        }
        self.flushBlock() catch |err| log.warn("archive close flush failed: {s}", .{@errorName(err)});
    }
};

pub const formatSegmentName = manifest_mod.formatSegmentName;
pub const parseSegmentIndex = manifest_mod.parseSegmentIndex;

// === tests ===

const testing = std.testing;

test "async flush pipeline performs four real zstd compressions concurrently" {
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();
    const worker_count = 4;
    const block_bytes = 8 * 1024 * 1024;

    var pipeline = try AsyncFlush.init(testing.allocator, io, worker_count);
    defer pipeline.deinit();
    try pipeline.start(worker_count);

    var prepared: [worker_count]writer_mod.PreparedBlock = undefined;
    defer for (&prepared) |*block| block.deinit(testing.allocator);
    var jobs: [worker_count]AsyncFlushJob = undefined;
    var prng = std.Random.DefaultPrng.init(0xfeed_face_dead_beef);
    for (&prepared, 0..) |*block, i| {
        const bytes = try testing.allocator.alloc(u8, block_bytes);
        prng.random().bytes(bytes);
        block.* = .{ .bytes = bytes, .max_seq = i + 1 };
    }
    for (&jobs, &prepared) |*job, *block| {
        job.* = .{ .prepared = block };
        pipeline.assign(job);
        pipeline.submit(job);
    }
    for (&jobs) |*job| job.compressed.waitUncancelable(io);
    try testing.expectEqual(@as(usize, worker_count), pipeline.workers_peak.load(.acquire));
    for (&jobs) |*job| {
        pipeline.waitTurn(job.id);
        testing.allocator.free(job.frame.?);
        pipeline.finishTurn();
    }
}

test "async flush commits deterministic JSS blocks and durability in preparation order" {
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();
    var sync_tmp = testing.tmpDir(.{});
    defer sync_tmp.cleanup();
    var async_tmp = testing.tmpDir(.{});
    defer async_tmp.cleanup();
    var sync_path_buf: [Io.Dir.max_path_bytes]u8 = undefined;
    var async_path_buf: [Io.Dir.max_path_bytes]u8 = undefined;
    const sync_path_len = try sync_tmp.dir.realPath(io, &sync_path_buf);
    const async_path_len = try async_tmp.dir.realPath(io, &async_path_buf);

    var synchronous = try Archive.init(testing.allocator, io, sync_path_buf[0..sync_path_len]);
    defer synchronous.deinit();
    synchronous.writer.max_events_per_block = 1;
    synchronous.max_events_per_block = 1;
    var sync_rows: [8]segment.Event = undefined;
    for (&sync_rows, 0..) |*event, i| event.* = testEvent(i + 1);
    _ = try synchronous.appendBatch(&sync_rows, 1);

    var asynchronous = try Archive.init(testing.allocator, io, async_path_buf[0..async_path_len]);
    defer asynchronous.deinit();
    const DurableReceipt = struct {
        values: [8]u64 = undefined,
        len: usize = 0,

        fn record(ctx: *anyopaque, durable_seq: u64, _: u64) anyerror!void {
            const self: *@This() = @ptrCast(@alignCast(ctx));
            if (self.len == self.values.len) return error.TooManyDurableBoundaries;
            self.values[self.len] = durable_seq;
            self.len += 1;
        }
    };
    var receipt: DurableReceipt = .{};
    asynchronous.on_durable_ctx = &receipt;
    asynchronous.on_durable = DurableReceipt.record;
    asynchronous.writer.max_events_per_block = 1;
    asynchronous.max_events_per_block = 1;
    try asynchronous.enableAsyncFlush(4);
    var async_rows: [8]segment.Event = undefined;
    for (&async_rows, 0..) |*event, i| event.* = testEvent(i + 1);
    _ = try asynchronous.appendBatch(&async_rows, 1);

    try testing.expectEqualSlices(u8, synchronous.writer.bytes(), asynchronous.writer.bytes());
    try testing.expectEqual(@as(u64, 8), asynchronous.committed_seq.load(.acquire));
    try testing.expectEqual(@as(u32, 8), asynchronous.writer.block_count);
    try testing.expectEqual(@as(usize, 8), receipt.len);
    for (receipt.values, 1..) |durable_seq, expected|
        try testing.expectEqual(@as(u64, expected), durable_seq);
    asynchronous.disableAsyncFlush();
}

fn testEvent(seq_hint: u64) segment.Event {
    return .{
        .seq = 0, // archive assigns
        .witnessed_at = @intCast(1_700_000_000_000_000 + seq_hint),
        .indexed_at = 0,
        .kind = .create,
        .did = "did:plc:archivetestaccount",
        .collection = "app.bsky.feed.post",
        .rkey = "3l3qo2vuowo2b",
        .rev = "3l3qo2vutsw2b",
        .payload = "\xa1\x64\x74\x65\x78\x74\x62\x68\x69",
    };
}

test "segment name round-trip" {
    var buf: [64]u8 = undefined;
    try testing.expectEqualStrings("seg_0000000000.jss", formatSegmentName(&buf, 0));
    try testing.expectEqualStrings("seg_0000000001.jss", formatSegmentName(&buf, 1));
    try testing.expectEqual(@as(?u64, 0), parseSegmentIndex("seg_0000000000.jss"));
    try testing.expectEqual(@as(?u64, 35), parseSegmentIndex(formatSegmentName(&buf, 35)));
    try testing.expectEqual(@as(?u64, null), parseSegmentIndex("seg_x.jss"));
}

test "active segment initialization faults remove the incomplete file" {
    segment_io.disarm();
    defer segment_io.disarm();
    const cases = [_]struct { spec: []const u8, expected: anyerror }{
        .{ .spec = "write:1:enospc", .expected = error.NoSpaceLeft },
        .{ .spec = "sync:1:eio", .expected = error.InputOutput },
        .{ .spec = "sync:2:eio", .expected = error.InputOutput },
    };
    for (cases) |case| {
        var tmp = testing.tmpDir(.{});
        defer tmp.cleanup();
        var path_buf: [Io.Dir.max_path_bytes]u8 = undefined;
        const path_len = try tmp.dir.realPath(testing.io, &path_buf);
        try testing.expect(segment_io.armSpec(case.spec));
        try testing.expectError(case.expected, Archive.init(testing.allocator, testing.io, path_buf[0..path_len]));
        segment_io.disarm();
        var segments = try tmp.dir.openDir(testing.io, "segments", .{ .iterate = true });
        defer segments.close(testing.io);
        var it = segments.iterate();
        try testing.expectEqual(@as(?Io.Dir.Entry, null), try it.next(testing.io));

        var recovered = try Archive.init(testing.allocator, testing.io, path_buf[0..path_len]);
        recovered.deinit();
    }
}

test "archive: append, rotate, recover across restart" {
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    // Archive.init uses cwd-relative paths; build one under the tmp dir
    var path_buf: [std.Io.Dir.max_path_bytes]u8 = undefined;
    const n = try tmp.dir.realPath(io, &path_buf);
    const tmp_path = path_buf[0..n];

    var data_dir_buf: [std.Io.Dir.max_path_bytes]u8 = undefined;
    const data_dir = try std.fmt.bufPrint(&data_dir_buf, "{s}/arch", .{tmp_path});
    var seal_stats: metrics.Stats = .{};

    {
        var a = try Archive.init(testing.allocator, io, data_dir);
        defer a.deinit();
        a.stats = &seal_stats;
        a.max_events_per_block = 10;
        a.writer.max_events_per_block = 10;
        a.max_segment_bytes = 2000; // force rotation

        var i: u64 = 0;
        while (i < 100) : (i += 1) {
            const seq = try a.append(testEvent(i), @intCast(i));
            try testing.expectEqual(i + 1, seq);
        }
        try testing.expect(a.seg_index > 0); // rotated at least once
        try testing.expect(a.committed_seq.load(.acquire) == 100);
        try testing.expectEqual(
            seal_stats.ingest_segments_rotated_total.load(.monotonic),
            seal_stats.segment_seal_duration_count.load(.monotonic),
        );
        a.close();
        // simulate crash: no seal of the current active segment
    }

    {
        var a = try Archive.init(testing.allocator, io, data_dir);
        defer a.deinit();
        // Recovery resumes the leftover active segment; seqs continue with no
        // reuse and the segment index does not churn merely because of restart.
        const seq = try a.append(testEvent(999), 999);
        try testing.expectEqual(@as(u64, 101), seq);
        a.close();
    }

    {
        var a = try Archive.init(testing.allocator, io, data_dir);
        defer a.deinit();
        // The next restart resumes the same segment with seq 101 durable.
        try testing.expectEqual(@as(u64, 102), a.next_seq);
        a.close();
    }

    {
        var a = try Archive.init(testing.allocator, io, data_dir);
        defer a.deinit();
        // Repeated recovery must retain the active segment's high-water mark.
        try testing.expectEqual(@as(u64, 102), a.next_seq);
        const seq = try a.append(testEvent(1000), 1000);
        try testing.expectEqual(@as(u64, 102), seq);
    }

    // Every sealed file parses, and the one active file contains the durable
    // suffix that was resumed across all three restarts.
    var dir = try Io.Dir.cwd().openDir(io, data_dir, .{});
    defer dir.close(io);
    var seg_dir = try dir.openDir(io, "segments", .{ .iterate = true });
    defer seg_dir.close(io);
    var it = seg_dir.iterate();
    var sealed_count: usize = 0;
    var active_count: usize = 0;
    var total_events: u64 = 0;
    while (try it.next(io)) |entry| {
        const bytes = try seg_dir.readFileAlloc(io, entry.name, testing.allocator, .limited(1 << 20));
        defer testing.allocator.free(bytes);
        var sealed = segment.Sealed.parse(testing.allocator, bytes) catch |err| switch (err) {
            error.ActiveSegment => {
                active_count += 1;
                var walk: writer_mod.ActiveIterator = .{ .bytes = bytes };
                while (try walk.next(testing.allocator)) |block_value| {
                    var block = block_value;
                    defer block.deinit(testing.allocator);
                    total_events += block.events.len;
                }
                continue;
            },
            else => return err,
        };
        defer sealed.deinit(testing.allocator);
        sealed_count += 1;
        total_events += sealed.header.event_count;
    }
    try testing.expect(sealed_count >= 1);
    try testing.expectEqual(@as(usize, 1), active_count);
    try testing.expectEqual(@as(u64, 101), total_events); // event 102 is unflushed-active
}

test "archive recovery fails closed on a complete corrupt active frame" {
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();
    var root_buf: [Io.Dir.max_path_bytes]u8 = undefined;
    const root_len = try tmp.dir.realPath(io, &root_buf);
    var data_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined;
    const data_dir = try std.fmt.bufPrint(&data_dir_buf, "{s}/archive", .{root_buf[0..root_len]});

    {
        var archive = try Archive.init(testing.allocator, io, data_dir);
        defer archive.deinit();
        archive.max_events_per_block = 1;
        archive.writer.max_events_per_block = 1;
        _ = try archive.append(testEvent(1), 1);
        archive.close();
    }

    var segment_path_buf: [Io.Dir.max_path_bytes]u8 = undefined;
    const segment_path = try std.fmt.bufPrint(
        &segment_path_buf,
        "{s}/segments/seg_0000000000.jss",
        .{data_dir},
    );
    var file = try Io.Dir.cwd().openFile(io, segment_path, .{ .mode = .read_write });
    const stat = try file.stat(io);
    const before = try testing.allocator.alloc(u8, @intCast(stat.size));
    defer testing.allocator.free(before);
    try testing.expectEqual(before.len, try file.readPositionalAll(io, before, 0));
    // Preserve the complete length-prefixed frame while corrupting its zstd
    // magic. This is not a torn tail and must never be truncated or sealed.
    const frame_start = segment.header_size + 8;
    before[frame_start] ^= 0xff;
    try file.writePositionalAll(io, before[frame_start..][0..1], frame_start);
    try file.sync(io);
    file.close(io);

    try testing.expectError(
        error.DecompressFailed,
        Archive.init(testing.allocator, io, data_dir),
    );

    const after = try Io.Dir.cwd().readFileAlloc(
        io,
        segment_path,
        testing.allocator,
        .limited(1 << 20),
    );
    defer testing.allocator.free(after);
    try testing.expectEqualSlices(u8, before, after);
}

test "archive recovery truncates only a torn suffix and resumes the same segment" {
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();
    var root_buf: [Io.Dir.max_path_bytes]u8 = undefined;
    const root_len = try tmp.dir.realPath(io, &root_buf);
    var data_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined;
    const data_dir = try std.fmt.bufPrint(&data_dir_buf, "{s}/archive", .{root_buf[0..root_len]});

    {
        var archive = try Archive.init(testing.allocator, io, data_dir);
        defer archive.deinit();
        archive.max_events_per_block = 1;
        archive.writer.max_events_per_block = 1;
        _ = try archive.append(testEvent(1), 1);
        archive.close();
    }

    var segment_path_buf: [Io.Dir.max_path_bytes]u8 = undefined;
    const segment_path = try std.fmt.bufPrint(
        &segment_path_buf,
        "{s}/segments/seg_0000000000.jss",
        .{data_dir},
    );
    var file = try Io.Dir.cwd().openFile(io, segment_path, .{ .mode = .read_write });
    const durable_len = (try file.stat(io)).size;
    var torn: [12]u8 = undefined;
    std.mem.writeInt(u64, torn[0..8], 1 << 20, .little);
    @memset(torn[8..], 0xff);
    try file.writePositionalAll(io, &torn, durable_len);
    try file.sync(io);
    file.close(io);

    {
        var recovered = try Archive.init(testing.allocator, io, data_dir);
        defer recovered.deinit();
        try testing.expectEqual(@as(u64, 0), recovered.seg_index);
        try testing.expectEqual(@as(usize, @intCast(durable_len)), recovered.file_written);
        try testing.expectEqual(durable_len, (try recovered.file.stat(io)).size);
        const seq = try recovered.append(testEvent(2), 2);
        try testing.expectEqual(@as(u64, 2), seq);
        recovered.close();
    }
    var reopened = try Archive.init(testing.allocator, io, data_dir);
    defer reopened.deinit();
    try testing.expectEqual(@as(u64, 3), reopened.next_seq);
}

test "archive recovery checksum-verifies the sealed high-water floor" {
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();
    var root_buf: [Io.Dir.max_path_bytes]u8 = undefined;
    const root_len = try tmp.dir.realPath(io, &root_buf);
    var data_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined;
    const data_dir = try std.fmt.bufPrint(&data_dir_buf, "{s}/archive", .{root_buf[0..root_len]});

    {
        var archive = try Archive.init(testing.allocator, io, data_dir);
        defer archive.deinit();
        archive.max_events_per_block = 1;
        archive.writer.max_events_per_block = 1;
        _ = try archive.append(testEvent(1), 1);
        try archive.rotate();
    }

    var sealed_path_buf: [Io.Dir.max_path_bytes]u8 = undefined;
    const sealed_path = try std.fmt.bufPrint(
        &sealed_path_buf,
        "{s}/segments/seg_0000000000.jss",
        .{data_dir},
    );
    var sealed_file = try Io.Dir.cwd().openFile(io, sealed_path, .{ .mode = .read_write });
    var corrupt_max_seq: [8]u8 = undefined;
    std.mem.writeInt(u64, &corrupt_max_seq, 999_999, .little);
    try sealed_file.writePositionalAll(io, &corrupt_max_seq, 34);
    try sealed_file.sync(io);
    sealed_file.close(io);

    try testing.expectError(
        error.BadChecksum,
        Archive.init(testing.allocator, io, data_dir),
    );
}

test "archive: metadata failure does not publish or strand durability" {
    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}/durable-hook", .{path_buf[0..n]});

    var archive = try Archive.init(testing.allocator, io, data_dir);
    defer archive.deinit();
    var stats: metrics.Stats = .{};
    archive.stats = &stats;
    archive.max_events_per_block = 1;
    archive.writer.max_events_per_block = 1;
    const Hook = struct {
        fail: bool = true,
        calls: usize = 0,
        fn run(ctx: *anyopaque, _: u64, _: u64) !void {
            const self: *@This() = @ptrCast(@alignCast(ctx));
            self.calls += 1;
            if (self.fail) return error.InjectedMetadataFailure;
        }
    };
    var hook: Hook = .{};
    archive.on_durable_ctx = &hook;
    archive.on_durable = Hook.run;

    try testing.expectError(error.InjectedMetadataFailure, archive.append(testEvent(1), 1));
    try testing.expectEqual(@as(u64, 0), archive.committed_seq.load(.acquire));
    try testing.expectEqual(@as(u64, 1), stats.ingest_blocks_flushed_total.load(.monotonic));
    try testing.expectEqual(@as(u64, 0), stats.ingest_append_errors_total.load(.monotonic));
    try testing.expect(stats.ingest_active_segment_bytes.load(.monotonic) > 0);
    var rejected = try archive.coldSnapshot(testing.allocator, 1);
    defer rejected.deinit();
    try testing.expectEqual(@as(usize, segment.header_size), rejected.active_start);
    try testing.expectEqual(rejected.active_start, rejected.active_length);
    hook.fail = false;
    try archive.flushBlock();
    var accepted = try archive.coldSnapshot(testing.allocator, 1);
    defer accepted.deinit();
    try testing.expectEqual(@as(usize, segment.header_size), accepted.active_start);
    try testing.expect(accepted.active_length > accepted.active_start);
    try testing.expectEqual(@as(u64, 1), archive.committed_seq.load(.acquire));
    // The retry re-establishes the metadata boundary but does not recount an
    // already-fsynced block.
    try testing.expectEqual(@as(u64, 1), stats.ingest_blocks_flushed_total.load(.monotonic));
    try testing.expectEqual(@as(usize, 2), hook.calls);
}

test "archive: timestamp lookup failure poisons transaction before sequence assignment" {
    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 archive = try Archive.init(testing.allocator, io, path_buf[0..path_len]);
    defer archive.deinit();
    var stats: metrics.Stats = .{};
    archive.stats = &stats;
    const Stamper = struct {
        fn run(_: *anyopaque, _: *segment.Event) !void {
            return error.InjectedTimestampLookupFailure;
        }
    };
    var context: u8 = 0;
    archive.stamper_ctx = &context;
    archive.stamper = Stamper.run;

    var rows = [_]segment.Event{testEvent(1)};
    try testing.expectError(error.InjectedTimestampLookupFailure, archive.appendBatch(&rows, 1));
    try testing.expectEqual(@as(u64, 1), archive.next_seq);
    try testing.expectEqual(@as(u64, 0), archive.committed_seq.load(.acquire));
    try testing.expectEqual(@as(usize, 0), archive.writer.pending.items.len);
    try testing.expectEqual(@as(u64, 1), stats.ingest_append_errors_total.load(.monotonic));
    try testing.expectError(error.ArchivePoisoned, archive.append(testEvent(2), 2));
}

test "archive: terminal seal publishes final partial block through metadata hook" {
    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 archive = try Archive.init(testing.allocator, io, path_buf[0..path_len]);
    archive.flush_interval_us = std.math.maxInt(i64);
    const Hook = struct {
        called: u64 = 0,
        durable: u64 = 0,

        fn run(ctx: *anyopaque, durable: u64, _: u64) !void {
            const self: *@This() = @ptrCast(@alignCast(ctx));
            self.called += 1;
            self.durable = durable;
        }
    };
    var hook: Hook = .{};
    archive.on_durable_ctx = &hook;
    archive.on_durable = Hook.run;
    _ = try archive.append(testEvent(1), 1);
    try testing.expectEqual(@as(u64, 0), hook.called);
    try archive.sealAndClose();
    try testing.expectEqual(@as(u64, 1), hook.called);
    try testing.expectEqual(@as(u64, 1), hook.durable);
}