//! in-memory hot tail — bounded deque of encoded v1 frames
//!
//! pull-based fan-out, following upstream's readable-log shape
//! (docs/upstream-harness.md): the ingest thread appends encode-once frames;
//! each subscriber thread reads at its own absolute index and blocks on a
//! condvar at the tip. no per-subscriber queues, so a slow reader costs
//! nothing but its own lag. eviction is by byte budget; a reader that falls
//! below the floor resumes from the oldest retained frame (there is no cold
//! store yet — that comes with the archive).

const std = @import("std");
const wire = @import("wire.zig");

const Io = std.Io;
const Allocator = std.mem.Allocator;

pub const SkipV1 = enum { none, sync, resync_replacement };

pub const Entry = struct {
    kind: wire.Kind,
    did: []u8,
    collection: []u8,
    time_us: i64,
    /// local archive seq (0 when the row wasn't archived)
    seq: u64,
    /// v1 wire; empty for kinds v1 never emits (#sync)
    json: []u8,
    /// v2 superset wire
    json_v2: []u8,
    skip_v1: SkipV1,

    fn deinit(self: Entry, allocator: Allocator) void {
        allocator.free(self.did);
        allocator.free(self.collection);
        allocator.free(self.json);
        allocator.free(self.json_v2);
    }
};

/// One immutable copied entry in a subscriber pull batch. The caller supplies
/// a short-lived arena, so all slices remain valid until that batch has been
/// processed without pinning the writer's deque against eviction.
pub const BatchEntry = struct {
    kind: wire.Kind,
    did: []const u8,
    collection: []const u8,
    seq: u64,
    json: []const u8,
    skip_v1: SkipV1,
};

pub const Batch = struct {
    entries: []BatchEntry,
};

pub const ReadResult = union(enum) {
    batch: Batch,
    fell_behind,
    closed,
};

pub const Tail = struct {
    allocator: Allocator,
    io: Io,
    mutex: Io.Mutex = .init,
    cond: Io.Condition = .init,
    entries: std.ArrayList(Entry) = .empty,
    /// Prefix slots are retired in O(1); compactLocked amortizes their removal
    /// instead of shifting the whole retained log for every eviction.
    head: usize = 0,
    /// absolute index of entries.items[head]
    base: u64 = 0,
    cur_bytes: usize = 0,
    /// Bytes belonging to seq-bearing rows newer than published_seq. Kept as
    /// an accumulator so a Prometheus scrape stays O(1) under the tail lock.
    pinned_bytes: usize = 0,
    max_bytes: usize,
    closed: bool = false,
    /// When enabled, rows newer than the durable watermark are pinned against
    /// eviction (upstream ReadableLog: the hot log is the only copy until the
    /// archive fsyncs them). Visibility is not affected: readers always see
    /// every appended row, matching upstream ReadFrom.
    durability_gated: bool = false,
    published_seq: u64 = 0,
    /// Highest nonzero seq ever appended (monotone; survives eviction).
    /// O(1) upper bound for cursor resolution: a cursor above it cannot be
    /// resident, so indexForSeq and publishedNextSeq never scan for it. The
    /// 2026-08-09 ingest convoy came from exactly that scan: every seq-cursor
    /// subscriber walked all ~165k resident entries under this mutex on every
    /// read-loop pass once the deque reached full retention.
    max_appended_seq: u64 = 0,

    pub fn init(allocator: Allocator, io: Io, max_bytes: usize) Tail {
        return .{ .allocator = allocator, .io = io, .max_bytes = max_bytes };
    }

    pub fn enableDurabilityGate(self: *Tail, durable_seq: u64) void {
        self.mutex.lockUncancelable(self.io);
        defer self.mutex.unlock(self.io);
        self.durability_gated = true;
        self.published_seq = durable_seq;
        self.pinned_bytes = 0;
        for (self.liveItems()) |entry| {
            if (entry.seq != 0 and entry.seq > durable_seq)
                self.pinned_bytes += entryBytes(entry);
        }
        self.evictLocked();
    }

    /// Advance the durable watermark after Archive's fsync + metadata
    /// transaction. Unpins the newly durable rows so eviction may reclaim
    /// them; does not change reader visibility (upstream advanceDurable).
    pub fn publishDurable(self: *Tail, durable_seq: u64) void {
        self.mutex.lockUncancelable(self.io);
        defer self.mutex.unlock(self.io);
        const old_durable = self.published_seq;
        if (durable_seq > old_durable) {
            // Seq-bearing rows are ordered. Walk only the undurable suffix,
            // not the entire retained log, then stop at the prior watermark.
            const live = self.liveItems();
            var i = live.len;
            while (i > 0) {
                i -= 1;
                const entry = live[i];
                if (entry.seq == 0 or entry.seq > durable_seq) continue;
                if (entry.seq <= old_durable) break;
                self.pinned_bytes -= entryBytes(entry);
            }
            self.published_seq = durable_seq;
        }
        self.evictLocked();
    }

    pub fn deinit(self: *Tail) void {
        for (self.liveItems()) |e| e.deinit(self.allocator);
        self.entries.deinit(self.allocator);
    }

    /// append a frame; copies all slices. wakes blocked readers.
    pub fn append(self: *Tail, kind: wire.Kind, did: []const u8, collection: []const u8, time_us: i64, seq: u64, json: []const u8, json_v2: []const u8, skip_v1: SkipV1) !void {
        const entry: Entry = .{
            .kind = kind,
            .did = try self.allocator.dupe(u8, did),
            .collection = try self.allocator.dupe(u8, collection),
            .time_us = time_us,
            .seq = seq,
            .json = try self.allocator.dupe(u8, json),
            .json_v2 = try self.allocator.dupe(u8, json_v2),
            .skip_v1 = skip_v1,
        };
        self.mutex.lockUncancelable(self.io);
        defer self.mutex.unlock(self.io);
        try self.entries.append(self.allocator, entry);
        if (entry.seq != 0) self.max_appended_seq = @max(self.max_appended_seq, entry.seq);
        const entry_bytes = entryBytes(entry);
        self.cur_bytes += entry_bytes;
        if (self.durability_gated and entry.seq != 0 and entry.seq > self.published_seq)
            self.pinned_bytes += entry_bytes;
        self.evictLocked();
        self.cond.broadcast(self.io);
    }

    fn evictLocked(self: *Tail) void {
        while (self.cur_bytes > self.max_bytes and self.head < self.entries.items.len) {
            const oldest = self.entries.items[self.head];
            // Rows newer than the archive watermark are the only copy a hot
            // subscriber can read. Match upstream's ReadableLog: exceed the
            // retention budget rather than evicting an undurable row.
            if (self.durability_gated and oldest.seq != 0 and oldest.seq > self.published_seq) break;
            const evicted = oldest;
            self.head += 1;
            self.cur_bytes -= entryBytes(evicted);
            evicted.deinit(self.allocator);
            self.base += 1;
        }
        self.compactLocked();
    }

    fn liveItems(self: *const Tail) []Entry {
        return self.entries.items[self.head..];
    }

    fn compactLocked(self: *Tail) void {
        if (self.head == 0) return;
        const live_len = self.entries.items.len - self.head;
        if (live_len == 0) {
            self.entries.clearRetainingCapacity();
            self.head = 0;
            return;
        }
        // Retire a large prefix in one move. Requiring both an absolute and a
        // relative threshold avoids churn for tiny tails while keeping stale
        // slots bounded for long-running append/evict workloads.
        if (self.head < 4096 or self.head < live_len) return;
        std.mem.copyForwards(Entry, self.entries.items[0..live_len], self.entries.items[self.head..]);
        self.entries.shrinkRetainingCapacity(live_len);
        self.head = 0;
    }

    fn entryBytes(e: Entry) usize {
        return e.json.len + e.json_v2.len + e.did.len + e.collection.len + 64;
    }

    pub const Gauges = struct {
        entries: usize,
        bytes: usize,
        pinned_bytes: usize,
        pinned_overrun_bytes: usize,
        base: u64,
        tip: u64,
    };

    pub fn gauges(self: *Tail) Gauges {
        self.mutex.lockUncancelable(self.io);
        defer self.mutex.unlock(self.io);
        return .{
            .entries = self.liveItems().len,
            .bytes = self.cur_bytes,
            .pinned_bytes = self.pinned_bytes,
            .pinned_overrun_bytes = self.pinned_bytes -| self.max_bytes,
            .base = self.base,
            .tip = self.base + self.liveItems().len,
        };
    }

    /// index of the next frame to be appended (start here for live tail).
    pub fn tip(self: *Tail) u64 {
        self.mutex.lockUncancelable(self.io);
        defer self.mutex.unlock(self.io);
        return self.base + self.liveItems().len;
    }

    /// Next index after the retained log. A new live subscriber starts here,
    /// matching upstream ReadFrom's at-tip semantics.
    pub fn publishedTip(self: *Tail) u64 {
        self.mutex.lockUncancelable(self.io);
        defer self.mutex.unlock(self.io);
        return self.base + self.liveItems().len;
    }

    /// Next local archive seq after the resident hot rows. Zero means no
    /// seq-bearing row is resident yet, matching upstream's unwarmed writer
    /// sentinel used by cursor resolution.
    pub fn publishedNextSeq(self: *Tail) u64 {
        self.mutex.lockUncancelable(self.io);
        defer self.mutex.unlock(self.io);
        if (self.max_appended_seq == 0) return 0;
        return self.max_appended_seq +| 1;
    }

    /// First sequence still resident in the hot log. Cold readers stop
    /// immediately before this sequence, then resolve it back to a hot
    /// index. A moving floor may force another cold batch; it must never turn
    /// into a skipped range.
    pub fn publishedFloorSeq(self: *Tail) ?u64 {
        self.mutex.lockUncancelable(self.io);
        defer self.mutex.unlock(self.io);
        for (self.liveItems()) |entry| {
            if (entry.seq != 0) return entry.seq;
        }
        return null;
    }

    /// time of the oldest retained entry, or null when empty.
    pub fn oldestTime(self: *Tail) ?i64 {
        self.mutex.lockUncancelable(self.io);
        defer self.mutex.unlock(self.io);
        const live = self.liveItems();
        if (live.len == 0) return null;
        return live[0].time_us;
    }

    /// smallest index whose entry has time_us >= cursor; tip if none.
    /// v1 cursor semantics: a below-floor cursor clamps to the oldest
    /// retained frame, never an error.
    pub fn indexForCursor(self: *Tail, cursor: i64) u64 {
        self.mutex.lockUncancelable(self.io);
        defer self.mutex.unlock(self.io);
        const live = self.liveItems();
        for (live, 0..) |e, i| {
            if (e.time_us >= cursor) return self.base + i;
        }
        return self.base + live.len;
    }

    /// copy the next frame at/after absolute index `idx` that matches `pred`,
    /// allocated from `allocator`. blocks at the tip until append or close.
    /// returns null when the tail is closed.
    pub fn next(
        self: *Tail,
        allocator: Allocator,
        idx: *u64,
        stop: *const std.atomic.Value(bool),
        v2: bool,
        ctx: anytype,
        comptime pred: fn (@TypeOf(ctx), wire.Kind, []const u8, []const u8, SkipV1) bool,
    ) !?[]u8 {
        self.mutex.lockUncancelable(self.io);
        defer self.mutex.unlock(self.io);
        while (true) {
            if (self.closed or stop.load(.acquire)) return null;
            if (idx.* < self.base) idx.* = self.base; // fell below the floor
            const rel: usize = @intCast(idx.* - self.base);
            const live = self.liveItems();
            if (rel >= live.len) {
                self.cond.wait(self.io, &self.mutex) catch return null;
                continue;
            }
            const e = live[rel];
            idx.* += 1;
            if (!v2 and e.skip_v1 != .none) {
                _ = pred(ctx, e.kind, e.did, e.collection, e.skip_v1);
                continue;
            }
            const json = if (v2) e.json_v2 else e.json;
            if (json.len == 0) continue; // kind not on this wire (#sync on v1)
            if (!pred(ctx, e.kind, e.did, e.collection, .none)) continue;
            return try allocator.dupe(u8, json);
        }
    }

    /// Pull at most `max` RAW log entries. Filtering and v1 skip rules happen
    /// after this boundary in the subscriber, matching upstream ReadFrom:
    /// selective clients cannot turn a bounded scan into an unbounded one.
    pub fn readBatch(
        self: *Tail,
        allocator: Allocator,
        idx: *u64,
        stop: *const std.atomic.Value(bool),
        v2: bool,
        max: usize,
    ) !ReadResult {
        self.mutex.lockUncancelable(self.io);
        defer self.mutex.unlock(self.io);
        const limit = @max(max, 1);
        while (true) {
            if (self.closed or stop.load(.acquire)) return .closed;
            if (idx.* < self.base) return .fell_behind;
            const rel: usize = @intCast(idx.* - self.base);
            const live = self.liveItems();
            if (rel >= live.len) {
                self.cond.wait(self.io, &self.mutex) catch return .closed;
                continue;
            }

            const visible = @min(live.len - rel, limit);
            const result = try allocator.alloc(BatchEntry, visible);
            var initialized: usize = 0;
            errdefer {
                for (result[0..initialized]) |entry| {
                    allocator.free(entry.did);
                    allocator.free(entry.collection);
                    allocator.free(entry.json);
                }
                allocator.free(result);
            }
            for (result, live[rel..][0..visible]) |*out, entry| {
                const did = try allocator.dupe(u8, entry.did);
                errdefer allocator.free(did);
                const collection = try allocator.dupe(u8, entry.collection);
                errdefer allocator.free(collection);
                const json = try allocator.dupe(u8, if (v2) entry.json_v2 else entry.json);
                out.* = .{ .kind = entry.kind, .did = did, .collection = collection, .seq = entry.seq, .json = json, .skip_v1 = entry.skip_v1 };
                initialized += 1;
            }
            idx.* += visible;
            return .{ .batch = .{ .entries = result } };
        }
    }

    /// smallest index whose entry has seq >= cursor (v2 seq cursors);
    /// null when the cursor is below the retained floor.
    pub fn indexForSeq(self: *Tail, cursor: u64) ?u64 {
        self.mutex.lockUncancelable(self.io);
        defer self.mutex.unlock(self.io);
        const live = self.liveItems();
        if (live.len == 0) return null;
        // A cursor above every seq ever appended cannot be resident: answer
        // at-tip without scanning, exactly what the full walk below returns
        // for it. At-tip and future cursors are the steady state of every
        // caught-up subscriber, so this bound keeps the read loop O(1).
        if (self.max_appended_seq != 0 and cursor > self.max_appended_seq)
            return self.base + live.len;
        // Non-archived diagnostic/control rows (seq=0) do not define the
        // retained sequence floor, and a tail of only such rows has no floor
        // at all: the caller must consult the archive instead. (Upstream's
        // read log is dense — append panics off-by-one — so it resolves
        // cursors with O(1) arithmetic; carrying seq=0 repair rows is our
        // divergence, and this lookup accommodates it in O(log n) instead of
        // the O(n) walk that convoyed ingest appends under this mutex — the
        // second scan from that family, after max_appended_seq above.)
        var first_nonzero: usize = 0;
        while (first_nonzero < live.len and live[first_nonzero].seq == 0) first_nonzero += 1;
        if (first_nonzero == live.len) return null;
        if (live[first_nonzero].seq > cursor and cursor > 0) return null;
        // leftmost nonzero-seq entry with seq >= cursor; zero rows are
        // transparent (probed past within the current half)
        var lo: usize = first_nonzero;
        var hi: usize = live.len;
        while (lo < hi) {
            const mid = lo + (hi - lo) / 2;
            var probe = mid;
            while (probe < hi and live[probe].seq == 0) probe += 1;
            if (probe == hi) {
                // [mid, hi) is all control rows: no candidate there
                hi = mid;
                continue;
            }
            if (live[probe].seq >= cursor) hi = probe else lo = probe + 1;
        }
        // the search can terminate on a control row; the linear scan's answer
        // is the first nonzero entry at or after it (identical by ordering)
        while (lo < live.len and live[lo].seq == 0) lo += 1;
        return self.base + lo;
    }

    /// wake all blocked readers without closing (e.g. so a reader can
    /// re-check its own liveness flag).
    pub fn wake(self: *Tail) void {
        self.mutex.lockUncancelable(self.io);
        defer self.mutex.unlock(self.io);
        self.cond.broadcast(self.io);
    }

    /// wake all blocked readers and make them return null.
    pub fn close(self: *Tail) void {
        self.mutex.lockUncancelable(self.io);
        defer self.mutex.unlock(self.io);
        self.closed = true;
        self.cond.broadcast(self.io);
    }
};

// === tests ===

const testing = std.testing;

fn acceptAll(_: void, _: wire.Kind, _: []const u8, _: []const u8, _: SkipV1) bool {
    return true;
}

test "append, read, evict, cursor clamp" {
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    var t = Tail.init(testing.allocator, threaded.io(), 300);
    defer t.deinit();

    try t.append(.commit, "did:plc:a", "app.bsky.feed.post", 100, 1, "{\"n\":1}", "{\"v2\":1}", .none);
    try t.append(.commit, "did:plc:a", "app.bsky.feed.post", 200, 2, "{\"n\":2}", "{\"v2\":2}", .none);

    var idx: u64 = 0;
    var stop: std.atomic.Value(bool) = .init(false);
    const j1 = (try t.next(testing.allocator, &idx, &stop, false, {}, acceptAll)).?;
    defer testing.allocator.free(j1);
    try testing.expectEqualStrings("{\"n\":1}", j1);

    // small budget: appending more evicts the oldest
    try t.append(.commit, "did:plc:a", "app.bsky.feed.post", 300, 3, "{\"n\":3}", "{\"v2\":3}", .none);
    try t.append(.commit, "did:plc:a", "app.bsky.feed.post", 400, 4, "{\"n\":4}", "{\"v2\":4}", .none);
    try testing.expect(t.base > 0);

    try testing.expectEqual(t.base, t.indexForCursor(0));
    try testing.expectEqual(t.tip(), t.indexForCursor(999_999));
}

test "future and at-tip cursors resolve O(1): the appended-seq bound answers without a scan" {
    // 2026-08-09 ingest convoy regression: every caught-up seq-cursor
    // subscriber walked the whole resident deque (and publishedNextSeq walked
    // it again via tooSlow) under the tail mutex on every read-loop pass.
    // The monotone max-appended bound must answer both queries without a
    // scan, and must survive eviction so the answers stay stable.
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    var t = Tail.init(testing.allocator, threaded.io(), 300);
    defer t.deinit();

    try t.append(.commit, "did:plc:a", "app.bsky.feed.post", 100, 10, "{\"n\":1}", "{\"v2\":1}", .none);
    try t.append(.commit, "did:plc:a", "app.bsky.feed.post", 200, 11, "{\"n\":2}", "{\"v2\":2}", .none);

    // at-tip (next expected) and far-future (e.g. a relay-space seq handed to
    // a local-seq cursor) both resolve to the tip, matching the full walk
    try testing.expectEqual(@as(?u64, t.tip()), t.indexForSeq(12));
    try testing.expectEqual(@as(?u64, t.tip()), t.indexForSeq(32_581_690_274));
    try testing.expectEqual(@as(u64, 12), t.publishedNextSeq());

    // evicting the seq-bearing rows must not reset the bound: the next-seq
    // anchor is "highest ever appended", not "highest still resident"
    try t.append(.commit, "did:plc:a", "app.bsky.feed.post", 300, 12, "{\"n\":3}", "{\"v2\":3}", .none);
    try t.append(.commit, "did:plc:a", "app.bsky.feed.post", 400, 13, "{\"n\":4}", "{\"v2\":4}", .none);
    try testing.expect(t.base > 0); // small budget forced eviction
    try testing.expectEqual(@as(u64, 14), t.publishedNextSeq());
    try testing.expectEqual(@as(?u64, t.tip()), t.indexForSeq(14));

    // a resident cursor still resolves exactly (no early-out short circuit)
    try testing.expectEqual(@as(?u64, t.tip() - 1), t.indexForSeq(13));
}

test "read batch bounds raw scanned entries before subscriber filtering" {
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    var t = Tail.init(testing.allocator, threaded.io(), 1 << 20);
    defer t.deinit();
    try t.append(.sync, "did:plc:a", "", 100, 1, "", "{\"seq\":1}", .sync);
    try t.append(.commit, "did:plc:a", "app.bsky.feed.post", 200, 2, "{\"n\":2}", "{\"seq\":2}", .none);
    try t.append(.commit, "did:plc:b", "app.bsky.feed.like", 300, 3, "{\"n\":3}", "{\"seq\":3}", .none);

    var idx: u64 = 0;
    var stop: std.atomic.Value(bool) = .init(false);
    var first_arena = std.heap.ArenaAllocator.init(testing.allocator);
    defer first_arena.deinit();
    const first = try t.readBatch(first_arena.allocator(), &idx, &stop, false, 2);
    try testing.expectEqual(@as(usize, 2), first.batch.entries.len);
    try testing.expectEqual(@as(u64, 2), idx);
    try testing.expectEqual(SkipV1.sync, first.batch.entries[0].skip_v1);
    try testing.expectEqual(@as(usize, 0), first.batch.entries[0].json.len);

    var second_arena = std.heap.ArenaAllocator.init(testing.allocator);
    defer second_arena.deinit();
    const second = try t.readBatch(second_arena.allocator(), &idx, &stop, false, 2);
    try testing.expectEqual(@as(usize, 1), second.batch.entries.len);
    try testing.expectEqual(@as(u64, 3), idx);
    try testing.expectEqualStrings("{\"n\":3}", second.batch.entries[0].json);
}

test "large eviction burst retires a prefix without per-entry shifts" {
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    var t = Tail.init(testing.allocator, threaded.io(), 64 * 1024 * 1024);
    defer t.deinit();

    for (0..20_000) |i| {
        try t.append(.commit, "did:plc:burst", "app.bsky.feed.post", @intCast(i), i + 1, "{}", "{}", .none);
    }
    try testing.expectEqual(@as(usize, 20_000), t.gauges().entries);

    // A large repair can push a previously full tail over a newly relevant
    // byte floor in one writer turn. This used to orderedRemove(0) twenty
    // thousand times, shifting the retained array on every row.
    t.max_bytes = 1024;
    try t.append(.commit, "did:plc:burst", "app.bsky.feed.post", 20_001, 20_001, "{}", "{}", .none);
    const gauges = t.gauges();
    try testing.expect(gauges.bytes <= t.max_bytes);
    try testing.expect(gauges.base > 19_000);
    try testing.expect(gauges.entries < 20);
    try testing.expect(t.entries.items.len < 20);
}

test "appended frames are readable before durability publication" {
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    var t = Tail.init(testing.allocator, threaded.io(), 1 << 20);
    defer t.deinit();
    t.enableDurabilityGate(0);

    // Upstream ReadFrom serves to the appended tip; durability only pins
    // rows against eviction. A live reader must see the row immediately.
    try t.append(.commit, "did:plc:a", "app.bsky.feed.post", 100, 1, "{\"n\":1}", "{\"v2\":1}", .none);
    try testing.expectEqual(@as(u64, 1), t.tip());
    try testing.expectEqual(@as(u64, 1), t.publishedTip());
    try testing.expectEqual(@as(?u64, 0), t.indexForSeq(1));

    var idx: u64 = 0;
    var stop: std.atomic.Value(bool) = .init(false);
    const json = (try t.next(testing.allocator, &idx, &stop, false, {}, acceptAll)).?;
    defer testing.allocator.free(json);
    try testing.expectEqualStrings("{\"n\":1}", json);
    try testing.expectEqual(t.gauges().bytes, t.gauges().pinned_bytes);

    t.publishDurable(1);
    try testing.expectEqual(@as(usize, 0), t.gauges().pinned_bytes);
}

test "seq cursor below the hot floor ignores leading unarchived rows" {
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    var t = Tail.init(testing.allocator, threaded.io(), 1 << 20);
    defer t.deinit();

    try t.append(.identity, "did:plc:a", "", 100, 0, "{}", "{}", .none);
    try t.append(.commit, "did:plc:a", "app.bsky.feed.post", 200, 42, "{}", "{}", .none);

    try testing.expectEqual(@as(?u64, null), t.indexForSeq(1));
    try testing.expectEqual(@as(?u64, 1), t.indexForSeq(42));
}

test "indexForSeq binary search matches the linear scan across control-row layouts" {
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    var t = Tail.init(testing.allocator, threaded.io(), 1 << 24);
    defer t.deinit();

    // interleave archived rows (seqs advancing with gaps) and seq=0 control
    // rows in runs: leading, interior singles, an interior run, and trailing
    var seq: u64 = 10;
    var prng = std.Random.DefaultPrng.init(7);
    const random = prng.random();
    try t.append(.identity, "did:plc:z", "", 1, 0, "{}", "{}", .none);
    try t.append(.identity, "did:plc:z", "", 1, 0, "{}", "{}", .none);
    for (0..500) |_| {
        seq += 1 + random.uintLessThan(u64, 5);
        try t.append(.commit, "did:plc:a", "app.bsky.feed.post", 1, seq, "{}", "{}", .none);
        if (random.uintLessThan(u8, 4) == 0) {
            for (0..1 + random.uintLessThan(usize, 3)) |_|
                try t.append(.identity, "did:plc:z", "", 1, 0, "{}", "{}", .none);
        }
    }
    try t.append(.identity, "did:plc:z", "", 1, 0, "{}", "{}", .none);

    // linear reference: the pre-binary-search behavior, verbatim
    const Ref = struct {
        fn indexForSeq(tl: *Tail, cursor: u64) ?u64 {
            const live = tl.entries.items[tl.head..];
            if (live.len == 0) return null;
            if (tl.max_appended_seq != 0 and cursor > tl.max_appended_seq)
                return tl.base + live.len;
            var saw_seq = false;
            for (live, 0..) |e, i| {
                if (e.seq == 0) continue;
                if (!saw_seq) {
                    saw_seq = true;
                    if (e.seq > cursor and cursor > 0) return null;
                }
                if (e.seq >= cursor) return tl.base + i;
            }
            if (!saw_seq) return null;
            return tl.base + live.len;
        }
    };

    // sweep: below floor, every present seq, every gap neighbor, at/past tip
    var cursor: u64 = 0;
    while (cursor <= seq + 3) : (cursor += 1) {
        try testing.expectEqual(Ref.indexForSeq(&t, cursor), t.indexForSeq(cursor));
    }
}

test "undurable rows pin readable log beyond budget until publication" {
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    var t = Tail.init(testing.allocator, threaded.io(), 1);
    defer t.deinit();
    t.enableDurabilityGate(0);

    try t.append(.commit, "did:plc:a", "app.bsky.feed.post", 100, 1, "{\"n\":1}", "{\"v2\":1}", .none);
    try t.append(.commit, "did:plc:a", "app.bsky.feed.post", 200, 2, "{\"n\":2}", "{\"v2\":2}", .none);
    const pinned = t.gauges();
    try testing.expectEqual(@as(usize, 2), pinned.entries);
    try testing.expectEqual(pinned.bytes, pinned.pinned_bytes);
    try testing.expectEqual(pinned.pinned_bytes - 1, pinned.pinned_overrun_bytes);
    try testing.expectEqual(@as(u64, 0), pinned.base);

    t.publishDurable(1);
    const one_pinned = t.gauges();
    try testing.expectEqual(@as(usize, 1), one_pinned.entries);
    try testing.expectEqual(@as(u64, 1), one_pinned.base);
    try testing.expectEqual(one_pinned.bytes, one_pinned.pinned_bytes);

    t.publishDurable(2);
    const drained = t.gauges();
    try testing.expectEqual(@as(usize, 0), drained.entries);
    try testing.expectEqual(@as(usize, 0), drained.bytes);
    try testing.expectEqual(@as(usize, 0), drained.pinned_bytes);
    try testing.expectEqual(@as(usize, 0), drained.pinned_overrun_bytes);
    try testing.expectEqual(@as(u64, 2), drained.base);
    try testing.expectEqual(@as(u64, 2), drained.tip);
}

test "v1 observes skipped rows while v2 receives them" {
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    var t = Tail.init(testing.allocator, threaded.io(), 1 << 20);
    defer t.deinit();
    try t.append(.sync, "did:plc:a", "", 100, 1, "", "{\"kind\":\"sync\"}", .sync);
    try t.append(.commit, "did:plc:a", "app.bsky.feed.post", 200, 2, "{\"kind\":\"commit\"}", "{\"kind\":\"commit\",\"seq\":2}", .none);

    const Observer = struct {
        skipped: usize = 0,
        fn wants(self: *@This(), _: wire.Kind, _: []const u8, _: []const u8, skip: SkipV1) bool {
            if (skip != .none) self.skipped += 1;
            return true;
        }
    };
    var stop: std.atomic.Value(bool) = .init(false);
    var v1_observer: Observer = .{};
    var v1_idx: u64 = 0;
    const v1 = (try t.next(testing.allocator, &v1_idx, &stop, false, &v1_observer, Observer.wants)).?;
    defer testing.allocator.free(v1);
    try testing.expectEqual(@as(usize, 1), v1_observer.skipped);
    try testing.expectEqualStrings("{\"kind\":\"commit\"}", v1);

    var v2_observer: Observer = .{};
    var v2_idx: u64 = 0;
    const v2 = (try t.next(testing.allocator, &v2_idx, &stop, true, &v2_observer, Observer.wants)).?;
    defer testing.allocator.free(v2);
    try testing.expectEqual(@as(usize, 0), v2_observer.skipped);
    try testing.expectEqualStrings("{\"kind\":\"sync\"}", v2);
}

test "close unblocks readers" {
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    var t = Tail.init(testing.allocator, threaded.io(), 1 << 20);
    defer t.deinit();
    const Reader = struct {
        fn run(tl: *Tail) void {
            var idx: u64 = 0;
            var stop: std.atomic.Value(bool) = .init(false);
            const got = tl.next(testing.allocator, &idx, &stop, false, {}, acceptAll) catch unreachable;
            std.debug.assert(got == null);
        }
    };
    const th = try std.Thread.spawn(.{}, Reader.run, .{&t});
    t.close();
    th.join();
}