//! Backfill engine — bounded parallel listRepos/getRepo crawl.
//!
//! listRepos remains a serial page walk. Whole pages accumulate until the
//! configured entry target, then eligible repositories are shuffled and fan
//! out across long-lived workers. Worker count is the only concurrency bound,
//! as it is upstream. Repository completion follows the
//! archive writer's durability boundary independently of the dispatch batch:
//!
//!   repo/not_started -> concurrent archive appends -> archive fsync
//!   -> every covered repo completion -> listRepos cursor after full drain
//!
//! A straggler can hold the listRepos cursor but cannot hold already durable
//! repository completions. A crash re-fetches only repositories whose final
//! rows had not crossed a writer durability boundary.

const std = @import("std");
const zat = @import("zat");
const archive_mod = @import("../../storage/archive.zig");
const crashpoint = @import("../../runtime/crashpoint.zig");
const fetched_car = @import("fetched_car.zig");
const prepared_fetch = @import("prepared_fetch.zig");
const metrics_mod = @import("../../runtime/metrics.zig");
const meta_store = @import("../../storage/meta_store.zig");
const observability = @import("../../runtime/observability.zig");
const repo_store = @import("repo_store.zig");
const repos = @import("repos.zig");
const segment = @import("../../storage/segment.zig");

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

const list_page_limit = 1000;
// Preparation is serialized after parallel disk-backed downloads, so a single
// repo may use the whole configured transient budget without deadlocking on
// other partial indexes. The lock is released once completeness is proven;
// prepared repositories then emit concurrently into the ordered archive.
const transient_retries: u8 = 3;
const rate_limit_retries: u8 = 20;
const rate_limit_ceiling_s: u64 = 330;
pub const default_workers: usize = 100;
pub const default_batch_size: usize = 100_000;

pub const Config = struct {
    relay_http: []const u8,
    /// Identity resolver used to persist upstream's account diagnostics
    /// metadata. Null is reserved for focused offline unit tests.
    plc_url: ?[]const u8 = null,
    data_dir: []const u8,
    /// Upstream's debug-only explicit-DID mode. It bypasses listRepos,
    /// resolves identity metadata, and skips merge discovery.
    selected_repos: []const []const u8 = &.{},
    /// 0 = unlimited (production); >0 = debug partial crawl
    max_repos: usize = 0,
    workers: usize = default_workers,
    /// Target listRepos entries per page-aligned dispatch batch. Zero uses
    /// Jetstream's production default, not Atmos's smaller library default.
    batch_size: usize = default_batch_size,
    metrics: ?*metrics_mod.Stats = null,
};

pub const Stats = struct {
    completed: u64 = 0,
    failed: u64 = 0,
    rows: u64 = 0,
    drops: repos.DropCounts = .{},
};

const Job = struct { did: []const u8 };

const Result = struct {
    did: []const u8,
    status: repo_store.Status,
    rev: []u8 = &.{},
    latest_rev: []u8 = &.{},
    host: []u8 = &.{},
    last_error: []u8 = &.{},
    error_class: repo_store.ErrorClass = .unknown,
    attempts: u32 = 0,
    started_us: i64 = 0,
    /// Last archive sequence emitted by this repository. Meaningful only when
    /// rows > 0.
    last_seq: u64 = 0,
    /// True when this result came through the repository handler and therefore
    /// has an archive watermark, including a successful CAR with zero rows.
    /// RepoNotFound is a complete terminal result without that watermark and
    /// follows upstream's immediate OnFail persistence path instead.
    completion_requires_durability: bool = false,
    /// Wall-clock completion time captured when the handler finished. Deferred
    /// metadata durability must not rewrite this as the later flush time.
    completion_real_us: i64 = 0,
    /// Monotonic time when a successful completion entered the writer
    /// completion queue.
    completion_queued_awake_us: i64 = 0,
    rows: u64 = 0,
    drops: repos.DropCounts = .{},
};

const IdentityMetadata = struct {
    handle: []u8 = &.{},
    pds: []u8 = &.{},

    fn deinit(self: *IdentityMetadata, allocator: Allocator) void {
        if (self.handle.len > 0) allocator.free(self.handle);
        if (self.pds.len > 0) allocator.free(self.pds);
        self.* = .{};
    }
};

fn resolveIdentityMetadata(allocator: Allocator, resolver: *zat.DidResolver, raw_did: []const u8) !?IdentityMetadata {
    const did = zat.Did.parse(raw_did) orelse return null;
    var doc = resolver.resolve(did) catch |err| {
        log.warn("backfill identity resolution {s} failed: {s}", .{ raw_did, @errorName(err) });
        return null;
    };
    defer doc.deinit();

    var out: IdentityMetadata = .{};
    errdefer out.deinit(allocator);
    if (doc.handle()) |raw_handle| if (zat.Handle.parse(raw_handle) != null) {
        out.handle = try allocator.dupe(u8, raw_handle);
        for (out.handle) |*byte| byte.* = std.ascii.toLower(byte.*);
    };
    if (doc.pdsEndpoint()) |pds| out.pds = try allocator.dupe(u8, pds);
    return out;
}

fn selectedReposPage(allocator: Allocator, dids: []const []const u8) !repos.ListReposPage {
    const refs = try allocator.alloc(repos.RepoRef, dids.len);
    for (dids, refs) |did, *ref| ref.* = .{ .did = did, .rev = "", .active = true };
    return .{ .repos = refs, .cursor = null };
}

fn stateWithStatus(state: repo_store.RepoState, status: repo_store.Status) repo_store.RepoState {
    var next = state;
    next.status = status;
    return next;
}

fn terminalState(result: Result, previous: ?repo_store.RepoState, terminal_us: i64) repo_store.RepoState {
    const downloaded = result.status == .complete and result.rev.len > 0;
    return .{
        .status = result.status,
        .active = if (previous) |state| state.active else true,
        .rev = if (result.rev.len > 0) result.rev else if (previous) |state| state.rev else "",
        .latest_rev = if (result.latest_rev.len > 0) result.latest_rev else if (previous) |state| state.latest_rev else "",
        .host = if (result.host.len > 0) result.host else if (previous) |state| state.host else "",
        .handle = if (previous) |state| state.handle else "",
        .pds = if (previous) |state| state.pds else "",
        .last_error = if (result.status == .failed) result.last_error else "",
        .error_class = if (result.status == .failed) result.error_class else .unknown,
        .attempts = if (result.status == .failed) result.attempts else 0,
        .last_attempt_us = terminal_us,
        .started_us = if (previous) |state| if (state.started_us != 0) state.started_us else result.started_us else result.started_us,
        .completed_us = if (result.status == .complete) terminal_us else if (previous) |state| state.completed_us else 0,
        .retry_count = if (result.status == .failed) if (previous) |state| state.retry_count else 0 else 0,
        .next_attempt_us = if (result.status == .failed) if (previous) |state| state.next_attempt_us else 0 else 0,
        .updated_us = if (downloaded) terminal_us else if (previous) |state| state.updated_us else 0,
        .record_count = if (previous) |state| state.record_count else 0,
        .total_bytes = if (previous) |state| state.total_bytes else 0,
    };
}

fn resolveAndRecordSelectedIdentity(allocator: Allocator, io: Io, plc_url: []const u8, store: *repo_store.Store, did: []const u8, state: repo_store.RepoState) !void {
    var resolver = zat.DidResolver.init(io, allocator);
    defer resolver.deinit();
    resolver.plc_url = plc_url;
    var identity = (try resolveIdentityMetadata(allocator, &resolver, did)) orelse return;
    defer identity.deinit(allocator);
    const host = try repo_store.hostFromPDS(allocator, identity.pds);
    defer if (host.len > 0) allocator.free(host);
    var next = state;
    next.handle = identity.handle;
    next.pds = identity.pds;
    next.host = host;
    try store.put(did, next);
}

const JobQueue = struct {
    buffer: []Job,
    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: *JobQueue, io: Io, job: Job) !void {
        self.mu.lockUncancelable(io);
        defer self.mu.unlock(io);
        while (self.len == self.buffer.len and !self.closed)
            try self.not_full.wait(io, &self.mu);
        if (self.closed) return error.Closed;
        self.buffer[(self.head + self.len) % self.buffer.len] = job;
        self.len += 1;
        self.not_empty.signal(io);
    }

    fn pop(self: *JobQueue, io: Io) !Job {
        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 count(self: *JobQueue, io: Io) usize {
        self.mu.lockUncancelable(io);
        defer self.mu.unlock(io);
        return self.len;
    }

    fn close(self: *JobQueue, 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 Work = struct {
    allocator: Allocator,
    io: Io,
    config: Config,
    archive: *archive_mod.Archive,
    queue: JobQueue,
    group: Io.Group = .init,
    worker_count: usize,
    result_mu: Io.Mutex = .init,
    result_ready: Io.Condition = .init,
    results: std.ArrayList(Result) = .empty,
    submitted: usize = 0,
    finished: usize = 0,
    fatal_error: ?anyerror = null,
    trace_context: observability.Context,

    fn init(
        allocator: Allocator,
        io: Io,
        config: Config,
        archive: *archive_mod.Archive,
        trace_context: observability.Context,
    ) !Work {
        const worker_count = effectiveWorkerCount(config.workers);
        const queue_capacity = worker_count * 2;
        const work: Work = .{
            .allocator = allocator,
            .io = io,
            .config = config,
            .archive = archive,
            .queue = .{ .buffer = try allocator.alloc(Job, queue_capacity) },
            .worker_count = worker_count,
            .trace_context = trace_context,
        };
        return work;
    }

    fn deinit(self: *Work) void {
        for (self.results.items) |result| freeResult(self.allocator, result);
        self.results.deinit(self.allocator);
        self.allocator.free(self.queue.buffer);
    }

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

    fn shutdown(self: *Work, completed: bool) void {
        self.queue.close(self.io);
        if (completed)
            self.group.await(self.io) catch {}
        else
            self.group.cancel(self.io);
        self.publishGauges();
    }

    fn submit(self: *Work, job: Job) !void {
        self.result_mu.lockUncancelable(self.io);
        self.submitted += 1;
        self.result_mu.unlock(self.io);
        self.queue.put(self.io, job) catch |err| {
            self.result_mu.lockUncancelable(self.io);
            self.submitted -= 1;
            self.result_mu.unlock(self.io);
            return err;
        };
        self.publishGauges();
    }

    fn submittedCount(self: *Work) usize {
        self.result_mu.lockUncancelable(self.io);
        defer self.result_mu.unlock(self.io);
        return self.submitted;
    }

    fn waitThrough(self: *Work, target: usize) void {
        self.result_mu.lockUncancelable(self.io);
        defer self.result_mu.unlock(self.io);
        while (self.finished < target)
            self.result_ready.waitUncancelable(self.io, &self.result_mu);
    }

    fn takeResults(self: *Work) std.ArrayList(Result) {
        self.result_mu.lockUncancelable(self.io);
        defer self.result_mu.unlock(self.io);
        const out = self.results;
        self.results = .empty;
        return out;
    }

    fn getFatal(self: *Work) ?anyerror {
        self.result_mu.lockUncancelable(self.io);
        defer self.result_mu.unlock(self.io);
        return self.fatal_error;
    }

    fn workerLoop(self: *Work) void {
        // No in-flight byte ceiling: worker count is the concurrency bound, as
        // it is upstream, which has no such budget. The version here blocked in
        // acquire() when exhausted, which is half of what deadlocked
        // experiment 4 and throttled experiment 5 at its cap.
        const allocator = self.allocator;
        while (self.queue.pop(self.io)) |job| {
            self.publishGauges();
            if (self.getFatal() != null) {
                self.finish(null);
                continue;
            }
            if (self.config.metrics) |m| _ = m.backfill_workers_active.fetchAdd(1, .monotonic);
            const result = downloadOne(
                self.allocator,
                allocator,
                self.io,
                self.config.data_dir,
                self.config.relay_http,
                job.did,
                self.archive,
                self.config.metrics,
                self.trace_context,
            ) catch |err| {
                if (self.config.metrics) |m| _ = m.backfill_workers_active.fetchSub(1, .monotonic);
                if (isFatalDownloadError(err)) {
                    self.finishFatal(err);
                } else {
                    log.warn("backfill {s} failed: {s}", .{ job.did, @errorName(err) });
                    self.finish(.{ .did = job.did, .status = .failed });
                }
                continue;
            };
            if (self.config.metrics) |m| _ = m.backfill_workers_active.fetchSub(1, .monotonic);
            self.finish(result);
        } else |_| {}
    }

    fn finishFatal(self: *Work, err: anyerror) void {
        self.result_mu.lockUncancelable(self.io);
        defer self.result_mu.unlock(self.io);
        if (self.fatal_error == null) self.fatal_error = err;
        self.finished += 1;
        self.result_ready.broadcast(self.io);
    }

    fn finish(self: *Work, maybe_result: ?Result) void {
        self.result_mu.lockUncancelable(self.io);
        defer self.result_mu.unlock(self.io);
        defer {
            self.finished += 1;
            self.result_ready.broadcast(self.io);
        }
        if (maybe_result) |raw_result| {
            var result = raw_result;
            if (result.status == .complete) {
                result.completion_real_us = Io.Timestamp.now(self.io, .real).toMicroseconds();
                result.completion_queued_awake_us = Io.Timestamp.now(self.io, .awake).toMicroseconds();
            }
            self.results.append(self.allocator, result) catch {
                freeResult(self.allocator, result);
                if (self.fatal_error == null) self.fatal_error = error.OutOfMemory;
                return;
            };
        }
    }

    /// Take whatever is ready without waiting. The submit loop uses this so it
    /// can release memory before it blocks on a full queue; waiting there would
    /// be the deadlock this exists to avoid.
    fn takeReadyNow(self: *Work) ?std.ArrayList(Result) {
        self.result_mu.lockUncancelable(self.io);
        defer self.result_mu.unlock(self.io);
        if (self.results.items.len == 0) return null;
        const out = self.results;
        self.results = .empty;
        return out;
    }

    fn takeReadyThrough(self: *Work, target: usize) ?std.ArrayList(Result) {
        self.result_mu.lockUncancelable(self.io);
        defer self.result_mu.unlock(self.io);
        while (self.results.items.len == 0 and self.finished < target and self.fatal_error == null)
            self.result_ready.waitUncancelable(self.io, &self.result_mu);
        if (self.results.items.len == 0) return null;
        const out = self.results;
        self.results = .empty;
        return out;
    }

    fn publishGauges(self: *Work) void {
        const m = self.config.metrics orelse return;
        m.backfill_queue_depth.store(self.queue.count(self.io), .monotonic);
    }
};

fn isFatalDownloadError(err: anyerror) bool {
    return err == error.ArchiveAppendFailed or err == error.OutOfMemory;
}

fn effectiveWorkerCount(requested: usize) usize {
    return if (requested == 0) default_workers else requested;
}

fn effectiveBatchSize(requested: usize) usize {
    return if (requested == 0) default_batch_size else requested;
}

fn shuffleJobs(jobs: []Job, seed: u64) void {
    var prng = std.Random.DefaultPrng.init(seed);
    prng.random().shuffle(Job, jobs);
}

/// Stream equivalent of upstream's completionBatcher. Successful repository
/// results are retained until the archive durable hook reports a sequence
/// boundary covering their final row. Empty repositories are eligible at the
/// next hook or forced drain. The listRepos dispatch batch is deliberately not
/// a completion boundary.
const CompletionBatcher = struct {
    allocator: Allocator,
    io: Io,
    config: Config,
    store: *repo_store.Store,
    queued: std.ArrayList(Result) = .empty,
    accumulated: Stats = .{},
    mu: Io.Mutex = .init,
    previous_ctx: ?*anyopaque = null,
    previous_hook: ?*const fn (*anyopaque, u64, u64) anyerror!void = null,

    fn init(allocator: Allocator, io: Io, config: Config, store: *repo_store.Store) CompletionBatcher {
        return .{ .allocator = allocator, .io = io, .config = config, .store = store };
    }

    fn deinit(self: *CompletionBatcher) void {
        for (self.queued.items) |result| freeResult(self.allocator, result);
        self.queued.deinit(self.allocator);
    }

    fn install(self: *CompletionBatcher, archive: *archive_mod.Archive) void {
        self.previous_ctx = archive.on_durable_ctx;
        self.previous_hook = archive.on_durable;
        archive.on_durable_ctx = self;
        archive.on_durable = onArchiveDurable;
    }

    fn restore(self: *CompletionBatcher, archive: *archive_mod.Archive) void {
        archive.on_durable_ctx = self.previous_ctx;
        archive.on_durable = self.previous_hook;
    }

    fn queue(self: *CompletionBatcher, result: Result) !void {
        std.debug.assert(result.status == .complete);
        std.debug.assert(result.completion_requires_durability);
        errdefer freeResult(self.allocator, result);
        self.mu.lockUncancelable(self.io);
        defer self.mu.unlock(self.io);
        for (self.queued.items, 0..) |queued, index| {
            if (!std.mem.eql(u8, queued.did, result.did)) continue;
            freeResult(self.allocator, queued);
            self.queued.items[index] = result;
            if (self.config.metrics) |m| {
                _ = m.backfill_completion_queued_total.fetchAdd(1, .monotonic);
                m.backfill_completion_queue_depth.store(self.queued.items.len, .monotonic);
            }
            return;
        }
        try self.queued.append(self.allocator, result);
        if (self.config.metrics) |m| {
            _ = m.backfill_completion_queued_total.fetchAdd(1, .monotonic);
            m.backfill_completion_queue_depth.store(self.queued.items.len, .monotonic);
        }
    }

    fn persistImmediate(self: *CompletionBatcher, result: Result) !void {
        std.debug.assert(!result.completion_requires_durability);
        defer freeResult(self.allocator, result);
        var scratch = std.heap.ArenaAllocator.init(self.allocator);
        defer scratch.deinit();
        const previous = try self.store.get(scratch.allocator(), result.did);
        const terminal_us = if (result.status == .complete and result.completion_real_us > 0)
            result.completion_real_us
        else
            Io.Timestamp.now(self.io, .real).toMicroseconds();
        self.store.put(result.did, terminalState(result, previous, terminal_us)) catch |err| {
            if (result.status == .failed) {
                if (self.config.metrics) |m|
                    _ = m.backfill_on_fail_store_errors_total.fetchAdd(1, .monotonic);
            }
            return err;
        };
        self.mu.lockUncancelable(self.io);
        defer self.mu.unlock(self.io);
        switch (result.status) {
            .complete => {
                self.accumulated.completed += 1;
                if (self.config.metrics) |m| {
                    _ = m.backfill_completed_total.fetchAdd(1, .monotonic);
                    m.backfill_progress_completed.store(@intCast(self.accumulated.completed), .monotonic);
                }
            },
            .failed => {
                self.accumulated.failed += 1;
                if (self.config.metrics) |m| _ = m.backfill_failed_total.fetchAdd(1, .monotonic);
            },
            else => {},
        }
    }

    fn onArchiveDurable(ctx: *anyopaque, durable_seq: u64, next_seq: u64) anyerror!void {
        const self: *CompletionBatcher = @ptrCast(@alignCast(ctx));
        if (self.previous_hook) |hook| try hook(self.previous_ctx.?, durable_seq, next_seq);
        try self.stageDurable(durable_seq, false);
    }

    fn drain(self: *CompletionBatcher, archive: *archive_mod.Archive) !void {
        if (self.config.metrics) |m| _ = m.backfill_forced_checkpoint_flushes_total.fetchAdd(1, .monotonic);
        try archive.flushBlock();
        try self.stageDurable(archive.committed_seq.load(.acquire), true);
    }

    fn stageDurable(self: *CompletionBatcher, durable_seq: u64, force: bool) !void {
        self.mu.lockUncancelable(self.io);
        defer self.mu.unlock(self.io);

        var eligible: usize = 0;
        for (self.queued.items) |result| {
            if (result.rows == 0 or result.last_seq <= durable_seq) {
                eligible += 1;
            } else if (force) {
                if (self.config.metrics) |m| _ = m.backfill_completion_stage_errors_total.fetchAdd(1, .monotonic);
                return error.BackfillCompletionNotDurable;
            }
        }
        if (eligible == 0) return;

        var scratch = std.heap.ArenaAllocator.init(self.allocator);
        defer scratch.deinit();
        const alloc = scratch.allocator();
        const transitions = try alloc.alloc(repo_store.Transition, eligible);
        var transition_index: usize = 0;
        for (self.queued.items) |result| {
            if (result.rows != 0 and result.last_seq > durable_seq) continue;
            const previous = try self.store.get(alloc, result.did);
            transitions[transition_index] = .{
                .did = result.did,
                .state = terminalState(result, previous, result.completion_real_us),
            };
            transition_index += 1;
        }
        // Data is durable; the acknowledgement is not yet. A cut here is the
        // one the durability ordering exists to survive.
        crashpoint.hit(.after_archive_durable_before_completion_commit);
        self.store.putMany(transitions) catch |err| {
            if (self.config.metrics) |m| _ = m.backfill_completion_stage_errors_total.fetchAdd(1, .monotonic);
            return err;
        };

        const durable_awake_us = Io.Timestamp.now(self.io, .awake).toMicroseconds();
        var retained: usize = 0;
        for (self.queued.items) |result| {
            if (result.rows != 0 and result.last_seq > durable_seq) {
                self.queued.items[retained] = result;
                retained += 1;
                continue;
            }
            crashpoint.hit(.after_repo_complete);
            self.accumulated.completed += 1;
            self.accumulated.rows += result.rows;
            self.accumulated.drops.invalid_collection += result.drops.invalid_collection;
            self.accumulated.drops.invalid_rkey += result.drops.invalid_rkey;
            self.accumulated.drops.field_too_long += result.drops.field_too_long;
            if (self.config.metrics) |m| {
                _ = m.backfill_completed_total.fetchAdd(1, .monotonic);
                _ = m.backfill_drops.getPtr(.invalid_collection).fetchAdd(result.drops.invalid_collection, .monotonic);
                _ = m.backfill_drops.getPtr(.invalid_rkey).fetchAdd(result.drops.invalid_rkey, .monotonic);
                _ = m.backfill_drops.getPtr(.field_too_long).fetchAdd(result.drops.field_too_long, .monotonic);
                if (result.completion_queued_awake_us > 0 and durable_awake_us >= result.completion_queued_awake_us)
                    m.observeBackfillCompletionWait(@intCast(durable_awake_us - result.completion_queued_awake_us));
            }
            freeResult(self.allocator, result);
        }
        self.queued.items.len = retained;
        if (self.config.metrics) |m| {
            m.backfill_progress_completed.store(@intCast(self.accumulated.completed), .monotonic);
            m.backfill_completion_queue_depth.store(retained, .monotonic);
            _ = m.backfill_completion_durable_batches_total.fetchAdd(1, .monotonic);
            _ = m.backfill_completion_durable_repos_total.fetchAdd(eligible, .monotonic);
        }
    }

    fn snapshot(self: *CompletionBatcher) Stats {
        self.mu.lockUncancelable(self.io);
        defer self.mu.unlock(self.io);
        return self.accumulated;
    }
};

fn consumeReady(
    work: *Work,
    completions: *CompletionBatcher,
    maybe_results: ?std.ArrayList(Result),
) !void {
    var results = maybe_results orelse return;
    defer results.deinit(work.allocator);
    for (results.items) |result| {
        if (result.status == .complete and result.completion_requires_durability)
            try completions.queue(result)
        else
            try completions.persistImmediate(result);
    }
}

fn finishWorkBatch(
    work: *Work,
    completions: *CompletionBatcher,
    jobs: []Job,
    shuffle: bool,
) !void {
    if (jobs.len > 0 and crashpoint.wouldHit(.mid_backfill_download))
        crashpoint.hit(.mid_backfill_download);

    if (shuffle) {
        shuffleJobs(jobs, @bitCast(Io.Timestamp.now(work.io, .awake).toMicroseconds()));
    }
    // Submit and consume together. The job queue is bounded at
    // worker_count * 2, so a whole-network batch (100,000) is far larger than
    // it: submitting the whole batch first blocks this fiber on a full queue
    // while finished repositories pile up in `results`, each still owning its
    // CAR against the in-flight budget. The budget then exhausts, workers can
    // no longer allocate, they stop draining the queue, and the submit this
    // fiber is blocked on never completes. Consuming as we go hands those
    // repositories to the completion batcher, which releases them at the
    // archive's durability boundary and keeps the budget circulating.
    //
    // Every test corpus is smaller than the queue, which is why this only
    // ever appeared at whole-network scale.
    for (jobs) |job| {
        if (work.getFatal()) |err| return err;
        try consumeReady(work, completions, work.takeReadyNow());
        try work.submit(job);
    }
    const target = work.submittedCount();
    while (true) {
        if (work.getFatal()) |err| return err;
        const results = work.takeReadyThrough(target) orelse break;
        try consumeReady(work, completions, results);
    }
    if (work.getFatal()) |err| return err;
    try completions.drain(work.archive);
}

pub fn run(
    allocator: Allocator,
    io: Io,
    config: Config,
    archive: *archive_mod.Archive,
    store: *repo_store.Store,
) !Stats {
    return runTraced(allocator, io, config, archive, store, observability.root_context);
}

pub fn runTraced(
    allocator: Allocator,
    io: Io,
    config: Config,
    archive: *archive_mod.Archive,
    store: *repo_store.Store,
    trace_context: observability.Context,
) !Stats {
    var trace = observability.Span.start(allocator, "ingest/backfill", "Run", trace_context, &.{});
    defer trace.deinit();
    defer trace.succeed();
    return trace.recordError(runBackfill(allocator, io, config, archive, store, trace.context()));
}

fn runBackfill(
    allocator: Allocator,
    io: Io,
    config: Config,
    archive: *archive_mod.Archive,
    store: *repo_store.Store,
    trace_context: observability.Context,
) !Stats {
    if (config.selected_repos.len > 0 and config.max_repos > 0) return error.ConflictingBackfillSelection;
    if (config.selected_repos.len > 0 and config.plc_url == null) return error.SelectedBackfillRequiresIdentityResolver;
    var scratch_buf: [Io.Dir.max_path_bytes]u8 = undefined;
    const scratch_dir = try std.fmt.bufPrint(&scratch_buf, "{s}/backfill/repo-scratch", .{config.data_dir});
    try fetched_car.resetScratch(io, scratch_dir);

    var list_client = zat.XrpcClient.init(io, allocator, config.relay_http);
    defer list_client.deinit();

    // Upstream's explicit-DID runner is intentionally serial and preserves
    // the caller's order. It is a diagnostic/reconciliation path, not the
    // parallel whole-network engine.
    var work_config = config;
    if (config.selected_repos.len > 0) work_config.workers = 1;
    var work = try Work.init(allocator, io, work_config, archive, trace_context);
    defer work.deinit();
    try work.start();
    var work_completed = false;
    defer work.shutdown(work_completed);

    var completions = CompletionBatcher.init(allocator, io, config, store);
    defer completions.deinit();
    completions.install(archive);
    defer completions.restore(archive);

    if (config.metrics) |m| m.backfill_progress_completed.store(0, .monotonic);
    var selected: usize = 0;
    // An explicit selection is a self-contained debug run. It must not leave
    // an old full-network discovery watermark for merge to consume later.
    if (config.selected_repos.len > 0) try store.meta.deleteDurable(bootstrap_cursor_key);
    const full_network = config.max_repos == 0 and config.selected_repos.len == 0;
    var cursor_owned: ?[]u8 = if (full_network)
        try loadRelayListReposCursor(store.meta, allocator)
    else
        null;
    defer if (cursor_owned) |owned| allocator.free(owned);
    var cursor: ?[]const u8 = if (cursor_owned) |owned| owned else null;
    var batch_arena = std.heap.ArenaAllocator.init(allocator);
    defer batch_arena.deinit();
    var jobs: std.ArrayList(Job) = .empty;
    var batch_entries: usize = 0;
    const batch_size = effectiveBatchSize(config.batch_size);

    paging: while (true) {
        var arena = std.heap.ArenaAllocator.init(allocator);
        defer arena.deinit();
        const alloc = arena.allocator();

        const selected_page = config.selected_repos.len > 0 and cursor == null;
        const page = if (selected_page)
            try selectedReposPage(alloc, config.selected_repos)
        else if (config.selected_repos.len > 0)
            break
        else
            try listReposPage(alloc, &list_client, cursor);
        var dispatched: std.ArrayList(repo_store.Transition) = .empty;
        var discovered_inactive: std.ArrayList(repo_store.Transition) = .empty;
        var discovered_active: usize = 0;
        batch_entries += page.repos.len;

        for (page.repos) |ref| {
            if (config.max_repos > 0 and selected >= config.max_repos) break;
            var existing = try store.get(alloc, ref.did);
            if (selected_page) {
                if (existing == null) {
                    try store.put(ref.did, .{ .status = .not_started, .active = true, .started_us = Io.Timestamp.now(io, .real).toMicroseconds() });
                    if (config.metrics) |m| _ = m.backfill_discovered_total.fetchAdd(1, .monotonic);
                    existing = try store.get(alloc, ref.did);
                }
                // Upstream reconciles an inactive selected row before
                // recording identity metadata, so a crash cannot briefly
                // attribute the newly resolved host as inactive.
                if (existing) |state| if (state.active != ref.active) {
                    var next = state;
                    next.active = ref.active;
                    try store.put(ref.did, next);
                    if (config.metrics) |m| _ = m.backfill_active_flips_total.fetchAdd(1, .monotonic);
                    existing = try store.get(alloc, ref.did);
                };
                try resolveAndRecordSelectedIdentity(alloc, io, config.plc_url.?, store, ref.did, existing.?);
                existing = try store.get(alloc, ref.did);
            }
            if (existing) |state| {
                if (state.active != ref.active) {
                    try store.put(ref.did, .{
                        .status = state.status,
                        .active = ref.active,
                        .rev = state.rev,
                        .latest_rev = state.latest_rev,
                        .host = state.host,
                        .handle = state.handle,
                        .pds = state.pds,
                        .last_error = state.last_error,
                        .error_class = state.error_class,
                        .attempts = state.attempts,
                        .last_attempt_us = state.last_attempt_us,
                        .started_us = state.started_us,
                        .completed_us = state.completed_us,
                        .retry_count = state.retry_count,
                        .next_attempt_us = state.next_attempt_us,
                        .updated_us = state.updated_us,
                        .record_count = state.record_count,
                        .total_bytes = state.total_bytes,
                    });
                    if (config.metrics) |m| _ = m.backfill_active_flips_total.fetchAdd(1, .monotonic);
                }
            }
            if (!ref.active) {
                if (existing == null)
                    try discovered_inactive.append(alloc, .{ .did = ref.did, .state = .{ .status = .not_started, .active = false } });
                continue;
            }
            if (existing) |state| {
                if (state.status != .failed and state.status != .not_started) continue;
            }
            try jobs.append(batch_arena.allocator(), .{ .did = try batch_arena.allocator().dupe(u8, ref.did) });
            if (existing == null) discovered_active += 1;
            try dispatched.append(alloc, .{ .did = ref.did, .state = if (existing) |state| stateWithStatus(state, .not_started) else .{ .status = .not_started } });
            selected += 1;
        }

        // Discovery is durable per page, but the relay cursor advances only
        // after every page accumulated into the dispatch batch is terminal.
        try store.putMany(discovered_inactive.items);
        if (config.metrics) |m| _ = m.backfill_discovered_total.fetchAdd(discovered_inactive.items.len, .monotonic);
        try store.putMany(dispatched.items);
        if (config.metrics) |m| _ = m.backfill_discovered_total.fetchAdd(discovered_active, .monotonic);
        const final_page = page.cursor == null;
        if (page.cursor) |next| {
            const owned = try allocator.dupe(u8, next);
            if (cursor_owned) |old| allocator.free(old);
            cursor_owned = owned;
            cursor = owned;
        }
        const max_reached = config.max_repos > 0 and selected >= config.max_repos;
        const batch_due = selected_page or max_reached or final_page or (full_network and batch_entries >= batch_size);
        if (!batch_due) continue;

        try finishWorkBatch(&work, &completions, jobs.items, !selected_page);
        if (full_network) try saveListReposCheckpoint(
            store.meta,
            if (page.cursor) |next| next else "",
            if (cursor) |last| last else "",
        );
        _ = batch_arena.reset(.retain_capacity);
        jobs = .empty;
        batch_entries = 0;

        if (max_reached or final_page or selected_page) break :paging;
    }

    {
        const close_started = Io.Timestamp.now(io, .awake);
        defer if (config.metrics) |m| m.observeOrchestratorState(.close_backfill, @intCast(@max(
            close_started.durationTo(Io.Timestamp.now(io, .awake)).toMicroseconds(),
            0,
        )));
        try archive.flushBlock();
        try store.flush();
    }
    work.publishGauges();
    work_completed = true;
    return completions.snapshot();
}

test "bootstrap identity metadata comes from a real DID resolution response" {
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();
    var listener = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable;
    defer listener.deinit(io);
    const Fixture = struct {
        fn serve(server: *Io.net.Server, task_io: Io) !void {
            var stream = try server.accept(task_io);
            defer stream.close(task_io);
            var read_buf: [2048]u8 = undefined;
            var write_buf: [4096]u8 = undefined;
            var reader = stream.reader(task_io, &read_buf);
            var writer = stream.writer(task_io, &write_buf);
            var http_server = std.http.Server.init(&reader.interface, &writer.interface);
            var request = try http_server.receiveHead();
            try request.respond(
                "{\"id\":\"did:plc:abcdefghijklmnopqrstuvwx\",\"alsoKnownAs\":[\"at://Alice.Test\"],\"verificationMethod\":[],\"service\":[{\"id\":\"#atproto_pds\",\"type\":\"AtprotoPersonalDataServer\",\"serviceEndpoint\":\"https://pds.example\"}]}",
                .{ .status = .ok, .keep_alive = false, .extra_headers = &.{.{ .name = "content-type", .value = "application/json" }} },
            );
        }
    };
    var server_future = try io.concurrent(Fixture.serve, .{ &listener, io });
    defer _ = server_future.cancel(io) catch {};
    var plc_buf: [96]u8 = undefined;
    const plc_url = try std.fmt.bufPrint(&plc_buf, "http://127.0.0.1:{d}", .{listener.socket.address.getPort()});
    var resolver = zat.DidResolver.init(io, testing.allocator);
    defer resolver.deinit();
    resolver.plc_url = plc_url;

    var identity = (try resolveIdentityMetadata(testing.allocator, &resolver, "did:plc:abcdefghijklmnopqrstuvwx")).?;
    defer identity.deinit(testing.allocator);
    try testing.expectEqualStrings("alice.test", identity.handle);
    try testing.expectEqualStrings("https://pds.example", identity.pds);
    try server_future.await(io);
}

test "explicit selected backfill resolves identity, bypasses listRepos, and persists its index" {
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();
    var listener = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable;
    defer listener.deinit(io);
    const Fixture = struct {
        fn serve(server: *Io.net.Server, task_io: Io) !void {
            for (0..2) |_| {
                var stream = try server.accept(task_io);
                defer stream.close(task_io);
                var read_buf: [4096]u8 = undefined;
                var write_buf: [4096]u8 = undefined;
                var reader = stream.reader(task_io, &read_buf);
                var writer = stream.writer(task_io, &write_buf);
                var http_server = std.http.Server.init(&reader.interface, &writer.interface);
                var request = try http_server.receiveHead();
                if (std.mem.startsWith(u8, request.head.target, "/did:plc:")) {
                    try request.respond(
                        "{\"id\":\"did:plc:abcdefghijklmnopqrstuvwx\",\"alsoKnownAs\":[\"at://selected.test\"],\"verificationMethod\":[],\"service\":[{\"id\":\"#atproto_pds\",\"type\":\"AtprotoPersonalDataServer\",\"serviceEndpoint\":\"https://PDS.EXAMPLE:443\"}]}",
                        .{ .status = .ok, .keep_alive = false },
                    );
                } else if (std.mem.startsWith(u8, request.head.target, "/xrpc/com.atproto.sync.getRepo?")) {
                    try request.respond("{\"error\":\"RepoNotFound\"}", .{
                        .status = .not_found,
                        .keep_alive = false,
                        .extra_headers = &.{.{ .name = "content-type", .value = "application/json" }},
                    });
                } else {
                    // A listRepos call would prove this is only API-shaped,
                    // rather than upstream's distinct selected-DID path.
                    return error.UnexpectedSelectedBackfillRequest;
                }
            }
        }
    };
    var server_future = try io.concurrent(Fixture.serve, .{ &listener, io });
    defer _ = server_future.cancel(io) catch {};
    var url_buf: [96]u8 = undefined;
    const url = try std.fmt.bufPrint(&url_buf, "http://127.0.0.1:{d}", .{listener.socket.address.getPort()});
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();
    var data_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined;
    const data_dir_len = try tmp.dir.realPath(io, &data_dir_buf);
    const data_dir = data_dir_buf[0..data_dir_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 store = try repo_store.Store.init(testing.allocator, io, &meta);
    defer store.deinit();
    try meta.putDurable(bootstrap_cursor_key, "stale-full-network-cursor");
    const selected = [_][]const u8{"did:plc:abcdefghijklmnopqrstuvwx"};
    var instrumentation: metrics_mod.Stats = .{};
    const stats = try run(testing.allocator, io, .{
        .relay_http = url,
        .plc_url = url,
        .data_dir = data_dir,
        .selected_repos = &selected,
        .workers = 1,
        .metrics = &instrumentation,
    }, &archive, &store);
    // A RepoNotFound response is a successful durable completion upstream:
    // there is no CAR to handle, but the DID must not be retried forever.
    try testing.expectEqual(@as(u64, 1), stats.completed);
    try testing.expectEqual(@as(u64, 1), instrumentation.backfill_discovered_total.load(.monotonic));
    try testing.expectEqual(@as(u64, 0), instrumentation.backfill_completion_queued_total.load(.monotonic));
    try testing.expectEqual(@as(usize, 0), instrumentation.backfill_completion_queue_depth.load(.monotonic));
    try testing.expectEqual(@as(u64, 0), instrumentation.backfill_completion_durable_batches_total.load(.monotonic));
    try testing.expectEqual(@as(u64, 0), instrumentation.backfill_completion_durable_repos_total.load(.monotonic));
    try testing.expectEqual(@as(u64, 0), instrumentation.backfill_completion_queue_wait_count.load(.monotonic));
    try testing.expectEqual(@as(u64, 1), instrumentation.backfill_completed_total.load(.monotonic));
    try testing.expectEqual(@as(usize, 1), instrumentation.backfill_progress_completed.load(.monotonic));
    try testing.expectEqual(@as(u64, 1), instrumentation.backfill_forced_checkpoint_flushes_total.load(.monotonic));
    try testing.expectEqual(@as(u64, 0), instrumentation.backfill_handle_repo_duration_count.load(.monotonic));
    const state = (try store.get(testing.allocator, selected[0])).?;
    defer store.freeState(testing.allocator, state);
    try testing.expectEqual(repo_store.Status.complete, state.status);
    try testing.expectEqual(@as(u32, 0), state.attempts);
    try testing.expect(state.completed_us > 0);
    try testing.expectEqualStrings("selected.test", state.handle);
    try testing.expectEqualStrings("https://PDS.EXAMPLE:443", state.pds);
    const indexed = (try store.lookupDIDByHandle(testing.allocator, "SELECTED.TEST")).?;
    defer testing.allocator.free(indexed);
    try testing.expectEqualStrings(selected[0], indexed);
    try testing.expect(try meta.getAlloc(testing.allocator, list_repos_cursor_key) == null);
    try testing.expect(try meta.getAlloc(testing.allocator, bootstrap_cursor_key) == null);
    try server_future.await(io);
}

test "explicit selected backfill repairs identity metadata without redownloading a complete repo" {
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();
    var listener = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable;
    defer listener.deinit(io);
    const Fixture = struct {
        fn serve(server: *Io.net.Server, task_io: Io) !void {
            var stream = try server.accept(task_io);
            defer stream.close(task_io);
            var read_buf: [2048]u8 = undefined;
            var write_buf: [4096]u8 = undefined;
            var reader = stream.reader(task_io, &read_buf);
            var writer = stream.writer(task_io, &write_buf);
            var http_server = std.http.Server.init(&reader.interface, &writer.interface);
            var request = try http_server.receiveHead();
            if (!std.mem.startsWith(u8, request.head.target, "/did:plc:")) return error.UnexpectedSelectedBackfillRequest;
            try request.respond(
                "{\"id\":\"did:plc:abcdefghijklmnopqrstuvwx\",\"alsoKnownAs\":[\"at://repaired.test\"],\"verificationMethod\":[],\"service\":[{\"id\":\"#atproto_pds\",\"type\":\"AtprotoPersonalDataServer\",\"serviceEndpoint\":\"https://repaired-pds.example\"}]}",
                .{ .status = .ok, .keep_alive = false },
            );
        }
    };
    var server_future = try io.concurrent(Fixture.serve, .{ &listener, io });
    defer _ = server_future.cancel(io) catch {};
    var url_buf: [96]u8 = undefined;
    const url = try std.fmt.bufPrint(&url_buf, "http://127.0.0.1:{d}", .{listener.socket.address.getPort()});
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();
    var data_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined;
    const data_dir_len = try tmp.dir.realPath(io, &data_dir_buf);
    const data_dir = data_dir_buf[0..data_dir_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 store = try repo_store.Store.init(testing.allocator, io, &meta);
    defer store.deinit();
    const did = "did:plc:abcdefghijklmnopqrstuvwx";
    try store.put(did, .{ .status = .complete, .active = false, .rev = "backfill", .latest_rev = "latest" });
    const selected = [_][]const u8{did};
    var instrumentation: metrics_mod.Stats = .{};
    const stats = try run(testing.allocator, io, .{
        .relay_http = url,
        .plc_url = url,
        .data_dir = data_dir,
        .selected_repos = &selected,
        .workers = 1,
        .metrics = &instrumentation,
    }, &archive, &store);
    try testing.expectEqual(@as(u64, 0), stats.completed);
    try testing.expectEqual(@as(u64, 1), instrumentation.backfill_active_flips_total.load(.monotonic));
    try testing.expectEqual(@as(u64, 0), instrumentation.backfill_discovered_total.load(.monotonic));
    const state = (try store.get(testing.allocator, did)).?;
    defer store.freeState(testing.allocator, state);
    try testing.expectEqual(repo_store.Status.complete, state.status);
    try testing.expect(state.active);
    try testing.expectEqualStrings("backfill", state.rev);
    try testing.expectEqualStrings("latest", state.latest_rev);
    try testing.expectEqualStrings("repaired.test", state.handle);
    try testing.expectEqualStrings("https://repaired-pds.example", state.pds);
    try server_future.await(io);
}

test "backfill worker zero resolves to Jetstream production default" {
    try std.testing.expectEqual(default_workers, effectiveWorkerCount(0));
    try std.testing.expectEqual(@as(usize, 17), effectiveWorkerCount(17));
}

test "terminal transitions preserve upstream RepoStatus diagnostics exactly" {
    const previous: repo_store.RepoState = .{
        .status = .failed,
        .active = false,
        .rev = "initial-rev",
        .latest_rev = "prior-live-rev",
        .host = "old-pds.example",
        .handle = "alice.test",
        .pds = "https://identity-pds.example",
        .last_error = "old failure",
        .error_class = .timeout,
        .attempts = 4,
        .last_attempt_us = 10,
        .started_us = 5,
        .completed_us = 7,
        .retry_count = 3,
        .next_attempt_us = 99,
        .updated_us = 8,
        .record_count = 42,
        .total_bytes = 1024,
    };

    const completed = terminalState(.{
        .did = "did:plc:alice",
        .status = .complete,
        .rev = @constCast("new-rev"),
        .latest_rev = @constCast("new-rev"),
        .host = @constCast("new-pds.example"),
        .attempts = 2,
        .started_us = 20,
    }, previous, 100);
    try testing.expect(!completed.active);
    try testing.expectEqualStrings("new-rev", completed.rev);
    try testing.expectEqualStrings("new-rev", completed.latest_rev);
    try testing.expectEqualStrings("new-pds.example", completed.host);
    try testing.expectEqualStrings("alice.test", completed.handle);
    try testing.expectEqualStrings("https://identity-pds.example", completed.pds);
    try testing.expectEqualStrings("", completed.last_error);
    try testing.expectEqual(repo_store.ErrorClass.unknown, completed.error_class);
    try testing.expectEqual(@as(u32, 0), completed.attempts);
    try testing.expectEqual(@as(u32, 0), completed.retry_count);
    try testing.expectEqual(@as(i64, 0), completed.next_attempt_us);
    try testing.expectEqual(@as(i64, 5), completed.started_us);
    try testing.expectEqual(@as(i64, 100), completed.last_attempt_us);
    try testing.expectEqual(@as(i64, 100), completed.completed_us);
    try testing.expectEqual(@as(i64, 100), completed.updated_us);
    try testing.expectEqual(@as(i64, 42), completed.record_count);
    try testing.expectEqual(@as(i64, 1024), completed.total_bytes);

    const failed = terminalState(.{
        .did = "did:plc:alice",
        .status = .failed,
        .last_error = @constCast("new failure"),
        .error_class = .http_5xx,
        .attempts = 2,
        .started_us = 20,
    }, previous, 101);
    try testing.expectEqualStrings("initial-rev", failed.rev);
    try testing.expectEqualStrings("prior-live-rev", failed.latest_rev);
    try testing.expectEqualStrings("old-pds.example", failed.host);
    try testing.expectEqualStrings("new failure", failed.last_error);
    try testing.expectEqual(@as(u32, 2), failed.attempts);
    try testing.expectEqual(@as(u32, 3), failed.retry_count);
    try testing.expectEqual(@as(i64, 99), failed.next_attempt_us);
    try testing.expectEqual(@as(i64, 7), failed.completed_us);
    try testing.expectEqual(@as(i64, 8), failed.updated_us);

    const not_found = terminalState(.{
        .did = "did:plc:alice",
        .status = .complete,
        .started_us = 20,
    }, previous, 102);
    try testing.expectEqualStrings("initial-rev", not_found.rev);
    try testing.expectEqualStrings("prior-live-rev", not_found.latest_rev);
    try testing.expectEqual(@as(i64, 8), not_found.updated_us);
    try testing.expectEqual(@as(i64, 102), not_found.completed_us);
    try testing.expectEqual(@as(u32, 0), not_found.attempts);
}

test "resource exhaustion is fatal rather than a durable large-repo skip" {
    try std.testing.expect(isFatalDownloadError(error.OutOfMemory));
    try std.testing.expect(isFatalDownloadError(error.ArchiveAppendFailed));
}

const list_repos_cursor_key = "relay/list_repos_cursor";
const bootstrap_cursor_key = "bootstrap/last_listrepos_cursor";

pub fn saveListReposCheckpoint(meta: *meta_store.Store, relay_cursor: []const u8, bootstrap_cursor: []const u8) !void {
    var writes = meta.batch();
    defer writes.deinit();
    writes.put(list_repos_cursor_key, relay_cursor);
    if (bootstrap_cursor.len > 0) writes.put(bootstrap_cursor_key, bootstrap_cursor);
    try writes.commitDurable();
}

pub fn loadRelayListReposCursor(meta: *meta_store.Store, allocator: Allocator) !?[]u8 {
    return meta.getAlloc(allocator, list_repos_cursor_key);
}

pub fn loadBootstrapLastListReposCursor(meta: *meta_store.Store, allocator: Allocator) !?[]u8 {
    return meta.getAlloc(allocator, bootstrap_cursor_key);
}

/// A batch takes ~40 minutes to drain, during which `listRepos` is never
/// called, so the client's pooled keep-alive connection goes idle and the
/// relay closes it. The next page reused that dead socket and `receiveHead`
/// read zero bytes, which unwound to `main` and ended the process at every
/// batch boundary; Docker restarted it and the crawl resumed from the durable
/// cursor, which made a fatal bug look like a scheduling quirk.
///
/// zat 0.3.23 replays that case in the transport, where it belongs, so the
/// retry that used to live here is gone. `ade3d5d` had worked around the same
/// race on the getRepo path by disabling keep-alive; that is now redundant but
/// left alone until it is worth re-measuring.
pub fn listReposPage(arena: Allocator, client: *zat.XrpcClient, cursor: ?[]const u8) !repos.ListReposPage {
    return listReposPageOnce(arena, client, cursor);
}

fn listReposPageOnce(arena: Allocator, client: *zat.XrpcClient, cursor: ?[]const u8) !repos.ListReposPage {
    var params = std.StringHashMap([]const u8).init(arena);
    try params.put("limit", std.fmt.comptimePrint("{d}", .{list_page_limit}));
    if (cursor) |c| try params.put("cursor", c);
    var resp = try client.query(zat.Nsid.parse("com.atproto.sync.listRepos").?, params);
    defer resp.deinit();
    if (resp.status != .ok) return error.ListReposFailed;
    return repos.parseListRepos(arena, resp.body);
}

fn downloadOne(
    result_allocator: Allocator,
    convert_allocator: Allocator,
    io: Io,
    data_dir: []const u8,
    relay_http: []const u8,
    did: []const u8,
    archive: *archive_mod.Archive,
    instrumentation: ?*metrics_mod.Stats,
    trace_context: observability.Context,
) !Result {
    var scratch_buf: [Io.Dir.max_path_bytes]u8 = undefined;
    const scratch_dir = try std.fmt.bufPrint(&scratch_buf, "{s}/backfill/repo-scratch", .{data_dir});
    var transient_attempt: u8 = 0;
    var rate_attempt: u8 = 0;
    var attempts: u32 = 0;
    var final_host: fetched_car.FinalHost = .{};
    const started_us = Io.Timestamp.now(io, .real).toMicroseconds();

    while (true) {
        var arena = std.heap.ArenaAllocator.init(convert_allocator);
        defer arena.deinit();
        const alloc = arena.allocator();
        attempts += 1;
        var outcome = prepared_fetch.fetch(result_allocator, alloc, io, scratch_dir, relay_http, did, &final_host) catch |err| {
            if (err == error.InvalidCommitRev) {
                if (instrumentation) |m| _ = m.backfill_drops.getPtr(.invalid_rev).fetchAdd(1, .monotonic);
            }
            // Upstream deliberately fails a five-minute attempt immediately
            // for this run; retrying it in-place can occupy a worker for
            // another 15 minutes. The durable steady pass will revisit it.
            if (err == error.AttemptTimeout) return failureResult(result_allocator, did, &final_host, attempts, started_us, .timeout, @errorName(err));
            if (transient_attempt >= transient_retries) return failureResult(result_allocator, did, &final_host, attempts, started_us, repo_store.classifyAttemptError(err), @errorName(err));
            try sleepBackoff(io, transient_attempt, 30);
            transient_attempt += 1;
            continue;
        };
        switch (outcome) {
            .not_found => return successResult(result_allocator, did, .complete, &final_host, attempts, started_us),
            .unavailable => return successResult(result_allocator, did, .unavailable, &final_host, attempts, started_us),
            .permanent => return failureResult(result_allocator, did, &final_host, attempts, started_us, .unknown, "permanent getRepo response"),
            .transient => {
                if (transient_attempt >= transient_retries) return failureResult(result_allocator, did, &final_host, attempts, started_us, .http_5xx, "transient getRepo response");
                try sleepBackoff(io, transient_attempt, 30);
                transient_attempt += 1;
            },
            .rate_limited => |retry_after| {
                if (rate_attempt >= rate_limit_retries) return failureResult(result_allocator, did, &final_host, attempts, started_us, .http_429, "rate limited");
                const delay = @min(retry_after orelse (@as(u64, 1) << @min(rate_attempt, 8)), rate_limit_ceiling_s);
                try io.sleep(Io.Duration.fromSeconds(@max(delay, 1)), .awake);
                rate_attempt += 1;
            },
            .ok => |*prepared| {
                defer prepared.deinit(io);
                const attributes = [_]observability.Attribute{
                    .{ .key = "did", .value = .{ .string = did } },
                };
                var trace = observability.Span.start(
                    result_allocator,
                    "ingest/backfill",
                    "handleRepo",
                    trace_context,
                    &attributes,
                );
                defer trace.deinit();
                defer trace.succeed();
                if (try trace.recordError(handlePrepared(
                    result_allocator,
                    alloc,
                    io,
                    did,
                    prepared,
                    archive,
                    instrumentation,
                    &final_host,
                    attempts,
                    started_us,
                    &transient_attempt,
                    trace.context(),
                ))) |result| return result;
            },
        }
    }
}

/// One downloaded repository's emit and flush. Null asks the caller to retry
/// the download.
fn handlePrepared(
    result_allocator: Allocator,
    alloc: Allocator,
    io: Io,
    did: []const u8,
    prepared: *prepared_fetch.Prepared,
    archive: *archive_mod.Archive,
    instrumentation: ?*metrics_mod.Stats,
    final_host: *const fetched_car.FinalHost,
    attempts: u32,
    started_us: i64,
    transient_attempt: *u8,
    trace_context: observability.Context,
) !?Result {
    const handle_started = Io.Timestamp.now(io, .awake);
    var local: Stats = .{};
    const now_us: i64 = Io.Timestamp.now(io, .real).toMicroseconds();
    const Sink = struct {
        archive: *archive_mod.Archive,
        now_us: i64,
        batch: [1024]segment.Event = undefined,
        batch_len: usize = 0,
        rows: u64 = 0,
        last_seq: u64 = 0,
        trace_context: observability.Context,

        fn flush(self: *@This()) anyerror!void {
            if (self.batch_len == 0) return;
            const first_seq = (try self.archive.appendBatchTraced(
                self.batch[0..self.batch_len],
                self.now_us,
                self.trace_context,
            )).?;
            self.last_seq = first_seq + self.batch_len - 1;
            self.batch_len = 0;
        }

        fn emit(self: *@This(), event: segment.Event) anyerror!void {
            self.batch[self.batch_len] = event;
            self.batch_len += 1;
            self.rows += 1;
            if (self.batch_len == self.batch.len) try self.flush();
        }
    };
    var sink: Sink = .{
        .archive = archive,
        .now_us = now_us,
        .trace_context = trace_context,
    };
    const head_rev = repos.emitPrepared(alloc, &prepared.repo, did, now_us, .create, &local.drops, &sink, Sink.emit) catch |err| switch (err) {
        error.OutOfMemory, error.ArchiveAppendFailed => return @errorCast(err),
        error.MissingBlock, error.IncompleteCar => {
            if (transient_attempt.* >= transient_retries) return try failureResult(result_allocator, did, final_host, attempts, started_us, .car, @errorName(err));
            try sleepBackoff(io, transient_attempt.*, 30);
            transient_attempt.* += 1;
            return null;
        },
        else => return try failureResult(result_allocator, did, final_host, attempts, started_us, .car, @errorName(err)),
    };
    sink.flush() catch |err| switch (err) {
        error.OutOfMemory, error.ArchiveAppendFailed => return @errorCast(err),
        else => return error.ArchiveAppendFailed,
    };
    if (instrumentation) |m| m.observeBackfillHandleRepo(@intCast(@max(
        handle_started.durationTo(Io.Timestamp.now(io, .awake)).toMicroseconds(),
        0,
    )));
    const owned_rev = try result_allocator.dupe(u8, head_rev);
    errdefer result_allocator.free(owned_rev);
    const owned_latest_rev = try result_allocator.dupe(u8, head_rev);
    errdefer result_allocator.free(owned_latest_rev);
    const owned_host = try result_allocator.dupe(u8, final_host.slice());
    errdefer if (owned_host.len > 0) result_allocator.free(owned_host);
    return .{
        .did = did,
        .status = .complete,
        .rev = owned_rev,
        .latest_rev = owned_latest_rev,
        .host = owned_host,
        .attempts = attempts,
        .started_us = started_us,
        .rows = sink.rows,
        .last_seq = sink.last_seq,
        .completion_requires_durability = true,
        .drops = local.drops,
    };
}

fn successResult(allocator: Allocator, did: []const u8, status: repo_store.Status, host: *const fetched_car.FinalHost, attempts: u32, started_us: i64) !Result {
    return .{
        .did = did,
        .status = status,
        .host = try allocator.dupe(u8, host.slice()),
        .attempts = attempts,
        .started_us = started_us,
    };
}

fn failureResult(allocator: Allocator, did: []const u8, host: *const fetched_car.FinalHost, attempts: u32, started_us: i64, class: repo_store.ErrorClass, message: []const u8) !Result {
    const owned_host = try allocator.dupe(u8, host.slice());
    errdefer if (owned_host.len > 0) allocator.free(owned_host);
    const owned_error = try allocator.dupe(u8, message);
    errdefer if (owned_error.len > 0) allocator.free(owned_error);
    return .{
        .did = did,
        .status = .failed,
        .host = owned_host,
        .last_error = owned_error,
        .error_class = class,
        .attempts = attempts,
        .started_us = started_us,
    };
}

fn freeResult(allocator: Allocator, result: Result) void {
    if (result.rev.len > 0) allocator.free(result.rev);
    if (result.latest_rev.len > 0) allocator.free(result.latest_rev);
    if (result.host.len > 0) allocator.free(result.host);
    if (result.last_error.len > 0) allocator.free(result.last_error);
}

fn sleepBackoff(io: Io, attempt: u8, ceiling_s: u64) !void {
    const seconds = @min(@as(u64, 1) << @min(attempt, 8), ceiling_s);
    try io.sleep(Io.Duration.fromSeconds(@intCast(seconds)), .awake);
}

const testing = std.testing;

test "listRepos cursor save/load round-trip" {
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    var path_buf: [std.fs.max_path_bytes]u8 = undefined;
    const path_len = try tmp.dir.realPath(testing.io, &path_buf);
    var meta = try meta_store.Store.open(testing.allocator, path_buf[0..path_len]);
    defer meta.deinit();
    try testing.expect(try loadRelayListReposCursor(&meta, testing.allocator) == null);
    try testing.expect(try loadBootstrapLastListReposCursor(&meta, testing.allocator) == null);
    try saveListReposCheckpoint(&meta, "did:plc:abc123", "did:plc:abc123");
    const first_resume = (try loadRelayListReposCursor(&meta, testing.allocator)).?;
    defer testing.allocator.free(first_resume);
    try testing.expectEqualStrings("did:plc:abc123", first_resume);
    const first_discovery = (try loadBootstrapLastListReposCursor(&meta, testing.allocator)).?;
    defer testing.allocator.free(first_discovery);
    try testing.expectEqualStrings("did:plc:abc123", first_discovery);
    try saveListReposCheckpoint(&meta, "did:plc:later", "did:plc:later");
    try saveListReposCheckpoint(&meta, "", "did:plc:later");
    const resume_cursor = (try loadRelayListReposCursor(&meta, testing.allocator)).?;
    defer testing.allocator.free(resume_cursor);
    try testing.expectEqual(@as(usize, 0), resume_cursor.len);
    const discovery = (try loadBootstrapLastListReposCursor(&meta, testing.allocator)).?;
    defer testing.allocator.free(discovery);
    try testing.expectEqualStrings("did:plc:later", discovery);
}

test "backfill batch size zero resolves to Jetstream production default" {
    try testing.expectEqual(default_batch_size, effectiveBatchSize(0));
    try testing.expectEqual(@as(usize, 17), effectiveBatchSize(17));
}

test "backfill dispatch batch is shuffled before worker admission" {
    const ordered = [_][]const u8{ "0", "1", "2", "3", "4", "5", "6", "7" };
    var jobs: [ordered.len]Job = undefined;
    for (&jobs, ordered) |*job, did| job.* = .{ .did = did };
    shuffleJobs(&jobs, 0x5eed5eed);
    var unchanged = true;
    for (jobs, ordered) |job, did| unchanged = unchanged and std.mem.eql(u8, job.did, did);
    try testing.expect(!unchanged);
}

test "writer durability commits completed repositories independently of a dispatch straggler" {
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();
    var data_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined;
    const data_dir_len = try tmp.dir.realPath(io, &data_dir_buf);
    const data_dir = data_dir_buf[0..data_dir_len];

    var archive = try archive_mod.Archive.init(testing.allocator, io, data_dir);
    defer archive.deinit();
    archive.max_events_per_block = 1;
    var meta = try meta_store.Store.open(testing.allocator, data_dir);
    defer meta.deinit();
    var store = try repo_store.Store.init(testing.allocator, io, &meta);
    defer store.deinit();
    const did_a = "did:plc:completion-a";
    const did_b = "did:plc:completion-b";
    const did_empty = "did:plc:completion-empty";
    try store.putMany(&.{
        .{ .did = did_a, .state = .{ .status = .not_started } },
        .{ .did = did_b, .state = .{ .status = .not_started } },
        .{ .did = did_empty, .state = .{ .status = .not_started } },
    });

    var instrumentation: metrics_mod.Stats = .{};
    var completions = CompletionBatcher.init(testing.allocator, io, .{
        .relay_http = "",
        .data_dir = data_dir,
        .metrics = &instrumentation,
    }, &store);
    defer completions.deinit();
    completions.install(&archive);
    defer completions.restore(&archive);

    const Event = struct {
        fn make(did: []const u8, rkey: []const u8) segment.Event {
            return .{
                .seq = 0,
                .witnessed_at = 1_700_000_000_000_000,
                .indexed_at = 0,
                .kind = .create,
                .did = did,
                .collection = "app.bsky.feed.post",
                .rkey = rkey,
                .rev = "3l3qo2vutsw2b",
                .payload = "\xa1\x64\x74\x65\x78\x74\x62\x68\x69",
            };
        }
    };
    var row_a = [_]segment.Event{Event.make(did_a, "3l3qo2vuowo2a")};
    const seq_a = (try archive.appendBatch(&row_a, 1_700_000_000_000_000)).?;
    {
        const rev_a = try testing.allocator.dupe(u8, "3l3qo2vutsw2b");
        errdefer testing.allocator.free(rev_a);
        const latest_a = try testing.allocator.dupe(u8, "3l3qo2vutsw2b");
        errdefer testing.allocator.free(latest_a);
        try completions.queue(.{
            .did = did_a,
            .status = .complete,
            .rev = rev_a,
            .latest_rev = latest_a,
            .rows = 1,
            .last_seq = seq_a,
            .completion_requires_durability = true,
            .completion_real_us = 1_700_000_000_123_456,
            .completion_queued_awake_us = Io.Timestamp.now(io, .awake).toMicroseconds(),
        });
    }

    // The second repository is the dispatch straggler at this point. Its
    // first durable block invokes the archive hook and must commit A without
    // waiting for B's worker result or the listRepos cursor checkpoint.
    var row_b = [_]segment.Event{Event.make(did_b, "3l3qo2vuowo2b")};
    const seq_b = (try archive.appendBatch(&row_b, 1_700_000_000_000_001)).?;
    try archive.flushBlock();
    const state_a = (try store.get(testing.allocator, did_a)).?;
    defer store.freeState(testing.allocator, state_a);
    try testing.expectEqual(repo_store.Status.complete, state_a.status);
    try testing.expectEqual(@as(i64, 1_700_000_000_123_456), state_a.completed_us);
    const state_b_before = (try store.get(testing.allocator, did_b)).?;
    defer store.freeState(testing.allocator, state_b_before);
    try testing.expectEqual(repo_store.Status.not_started, state_b_before.status);
    try testing.expect(try loadRelayListReposCursor(&meta, testing.allocator) == null);

    {
        const rev_b = try testing.allocator.dupe(u8, "3l3qo2vutsw2b");
        errdefer testing.allocator.free(rev_b);
        const latest_b = try testing.allocator.dupe(u8, "3l3qo2vutsw2b");
        errdefer testing.allocator.free(latest_b);
        try completions.queue(.{
            .did = did_b,
            .status = .complete,
            .rev = rev_b,
            .latest_rev = latest_b,
            .rows = 1,
            .last_seq = seq_b,
            .completion_requires_durability = true,
            .completion_real_us = Io.Timestamp.now(io, .real).toMicroseconds(),
            .completion_queued_awake_us = Io.Timestamp.now(io, .awake).toMicroseconds(),
        });
    }
    // RepoNotFound arrives through upstream's OnFail special case rather than
    // the CAR handler, so it is a durable immediate terminal transition.
    try completions.persistImmediate(.{
        .did = did_empty,
        .status = .complete,
        .completion_real_us = Io.Timestamp.now(io, .real).toMicroseconds(),
        .completion_queued_awake_us = Io.Timestamp.now(io, .awake).toMicroseconds(),
    });
    try completions.drain(&archive);

    const state_b = (try store.get(testing.allocator, did_b)).?;
    defer store.freeState(testing.allocator, state_b);
    const state_empty = (try store.get(testing.allocator, did_empty)).?;
    defer store.freeState(testing.allocator, state_empty);
    try testing.expectEqual(repo_store.Status.complete, state_b.status);
    try testing.expectEqual(repo_store.Status.complete, state_empty.status);
    try testing.expectEqual(@as(u64, 2), instrumentation.backfill_completion_queued_total.load(.monotonic));
    try testing.expectEqual(@as(u64, 2), instrumentation.backfill_completion_durable_repos_total.load(.monotonic));
    try testing.expectEqual(@as(usize, 0), instrumentation.backfill_completion_queue_depth.load(.monotonic));
    try testing.expectEqual(@as(u64, 3), completions.snapshot().completed);
}

test "completion queue replaces a duplicate DID without growing its durable set" {
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();
    var data_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined;
    const data_dir_len = try tmp.dir.realPath(io, &data_dir_buf);
    const data_dir = data_dir_buf[0..data_dir_len];
    var meta = try meta_store.Store.open(testing.allocator, data_dir);
    defer meta.deinit();
    var store = try repo_store.Store.init(testing.allocator, io, &meta);
    defer store.deinit();
    var instrumentation: metrics_mod.Stats = .{};
    var completions = CompletionBatcher.init(testing.allocator, io, .{
        .relay_http = "",
        .data_dir = data_dir,
        .metrics = &instrumentation,
    }, &store);
    defer completions.deinit();

    for ([_][]const u8{ "rev-old", "rev-new" }) |rev| {
        const owned_rev = try testing.allocator.dupe(u8, rev);
        errdefer testing.allocator.free(owned_rev);
        const owned_latest = try testing.allocator.dupe(u8, rev);
        errdefer testing.allocator.free(owned_latest);
        try completions.queue(.{
            .did = "did:plc:completion-replaced",
            .status = .complete,
            .rev = owned_rev,
            .latest_rev = owned_latest,
            .rows = 1,
            .last_seq = 42,
            .completion_requires_durability = true,
            .completion_real_us = Io.Timestamp.now(io, .real).toMicroseconds(),
            .completion_queued_awake_us = Io.Timestamp.now(io, .awake).toMicroseconds(),
        });
    }
    try testing.expectEqual(@as(usize, 1), completions.queued.items.len);
    try testing.expectEqualStrings("rev-new", completions.queued.items[0].rev);
    try testing.expectEqual(@as(u64, 2), instrumentation.backfill_completion_queued_total.load(.monotonic));
    try testing.expectEqual(@as(usize, 1), instrumentation.backfill_completion_queue_depth.load(.monotonic));
}

test "durable repository completion survives reopen before listRepos cursor advance" {
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();
    var data_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined;
    const data_dir_len = try tmp.dir.realPath(io, &data_dir_buf);
    const data_dir = data_dir_buf[0..data_dir_len];
    const ready_did = "did:plc:restart-ready";
    const held_did = "did:plc:restart-held";

    {
        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 store = try repo_store.Store.init(testing.allocator, io, &meta);
        defer store.deinit();
        try store.putMany(&.{
            .{ .did = ready_did, .state = .{ .status = .not_started } },
            .{ .did = held_did, .state = .{ .status = .not_started } },
        });
        var completions = CompletionBatcher.init(testing.allocator, io, .{
            .relay_http = "",
            .data_dir = data_dir,
        }, &store);
        defer completions.deinit();
        completions.install(&archive);
        defer completions.restore(&archive);

        var ready_event = [_]segment.Event{.{
            .seq = 0,
            .witnessed_at = 1_700_000_000_000_000,
            .indexed_at = 0,
            .kind = .create,
            .did = ready_did,
            .collection = "app.bsky.feed.post",
            .rkey = "3l3qo2vuowo2a",
            .rev = "3l3qo2vutsw2b",
            .payload = "\xa1\x64\x74\x65\x78\x74\x62\x68\x69",
        }};
        const ready_seq = (try archive.appendBatch(&ready_event, 1_700_000_000_000_000)).?;
        {
            const rev = try testing.allocator.dupe(u8, "3l3qo2vutsw2b");
            errdefer testing.allocator.free(rev);
            const latest = try testing.allocator.dupe(u8, "3l3qo2vutsw2b");
            errdefer testing.allocator.free(latest);
            try completions.queue(.{
                .did = ready_did,
                .status = .complete,
                .rev = rev,
                .latest_rev = latest,
                .rows = 1,
                .last_seq = ready_seq,
                .completion_requires_durability = true,
                .completion_real_us = Io.Timestamp.now(io, .real).toMicroseconds(),
                .completion_queued_awake_us = Io.Timestamp.now(io, .awake).toMicroseconds(),
            });
        }
        // held_did emits one row and then deliberately never returns a worker
        // result. Its writer flush must still make ready_did durable without
        // inventing a completion for held_did or advancing the page cursor.
        var held_event = [_]segment.Event{.{
            .seq = 0,
            .witnessed_at = 1_700_000_000_000_001,
            .indexed_at = 0,
            .kind = .create,
            .did = held_did,
            .collection = "app.bsky.feed.post",
            .rkey = "3l3qo2vuowo2b",
            .rev = "3l3qo2vutsw2b",
            .payload = "\xa1\x64\x74\x65\x78\x74\x62\x68\x69",
        }};
        _ = try archive.appendBatch(&held_event, 1_700_000_000_000_001);
        try archive.flushBlock();
        const before_restart = (try store.get(testing.allocator, ready_did)).?;
        defer store.freeState(testing.allocator, before_restart);
        try testing.expectEqual(repo_store.Status.complete, before_restart.status);
        try testing.expect(try loadRelayListReposCursor(&meta, testing.allocator) == null);
    }

    {
        var meta = try meta_store.Store.open(testing.allocator, data_dir);
        defer meta.deinit();
        var store = try repo_store.Store.init(testing.allocator, io, &meta);
        defer store.deinit();
        const ready = (try store.get(testing.allocator, ready_did)).?;
        defer store.freeState(testing.allocator, ready);
        const held = (try store.get(testing.allocator, held_did)).?;
        defer store.freeState(testing.allocator, held);
        try testing.expectEqual(repo_store.Status.complete, ready.status);
        // Startup recovery moves only the actually interrupted repository to
        // the merge-time pending pass; it must not replay ready_did.
        try testing.expectEqual(repo_store.Status.pending, held.status);
        try testing.expect(try loadRelayListReposCursor(&meta, testing.allocator) == null);
    }
}

test "bootstrap redirects preserve independent transient and rate-limit retry budgets" {
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var target = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable;
    defer target.deinit(io);
    var origin = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable;
    defer origin.deinit(io);

    var target_url_buf: [96]u8 = undefined;
    const target_url = try std.fmt.bufPrint(
        &target_url_buf,
        "http://127.0.0.1:{d}/xrpc/com.atproto.sync.getRepo?did=did:plc:retry-fixture",
        .{target.socket.address.getPort()},
    );

    const Fixture = struct {
        fn originServer(server: *Io.net.Server, task_io: Io, location: []const u8) !void {
            for (0..4) |_| {
                var stream = try server.accept(task_io);
                defer stream.close(task_io);
                var read_buf: [2048]u8 = undefined;
                var write_buf: [2048]u8 = undefined;
                var reader = stream.reader(task_io, &read_buf);
                var writer = stream.writer(task_io, &write_buf);
                var http_server = std.http.Server.init(&reader.interface, &writer.interface);
                var request = try http_server.receiveHead();
                try request.respond("", .{
                    .status = .temporary_redirect,
                    .keep_alive = false,
                    .extra_headers = &.{.{ .name = "location", .value = location }},
                });
            }
        }

        fn targetServer(server: *Io.net.Server, task_io: Io) !void {
            for (0..4) |attempt| {
                var stream = try server.accept(task_io);
                defer stream.close(task_io);
                var read_buf: [2048]u8 = undefined;
                var write_buf: [2048]u8 = undefined;
                var reader = stream.reader(task_io, &read_buf);
                var writer = stream.writer(task_io, &write_buf);
                var http_server = std.http.Server.init(&reader.interface, &writer.interface);
                var request = try http_server.receiveHead();
                switch (attempt) {
                    0, 2 => try request.respond("", .{
                        .status = .service_unavailable,
                        .keep_alive = false,
                    }),
                    1 => try request.respond("", .{
                        .status = .too_many_requests,
                        .keep_alive = false,
                        .extra_headers = &.{.{ .name = "retry-after", .value = "1" }},
                    }),
                    3 => try request.respond("{\"error\":\"RepoNotFound\"}", .{
                        .status = .not_found,
                        .keep_alive = false,
                        .extra_headers = &.{.{ .name = "content-type", .value = "application/json" }},
                    }),
                    else => unreachable,
                }
            }
        }
    };

    var origin_future = try io.concurrent(Fixture.originServer, .{ &origin, io, target_url });
    defer _ = origin_future.cancel(io) catch {};
    var target_future = try io.concurrent(Fixture.targetServer, .{ &target, io });
    defer _ = target_future.cancel(io) catch {};

    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();
    var data_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined;
    const data_dir_len = try tmp.dir.realPath(io, &data_dir_buf);
    const data_dir = data_dir_buf[0..data_dir_len];
    var archive = try archive_mod.Archive.init(testing.allocator, io, data_dir);
    defer archive.deinit();
    var origin_url_buf: [96]u8 = undefined;
    const origin_url = try std.fmt.bufPrint(&origin_url_buf, "http://127.0.0.1:{d}", .{origin.socket.address.getPort()});

    var instrumentation: metrics_mod.Stats = .{};
    const result = try downloadOne(
        testing.allocator,
        testing.allocator,
        io,
        data_dir,
        origin_url,
        "did:plc:retry-fixture",
        &archive,
        &instrumentation,
        observability.root_context,
    );
    defer freeResult(testing.allocator, result);

    try testing.expectEqual(repo_store.Status.complete, result.status);
    try testing.expectEqual(@as(u64, 0), instrumentation.backfill_handle_repo_duration_count.load(.monotonic));
    try testing.expectEqual(@as(u32, 4), result.attempts);
    var expected_host_buf: [64]u8 = undefined;
    const expected_host = try std.fmt.bufPrint(&expected_host_buf, "127.0.0.1:{d}", .{target.socket.address.getPort()});
    try testing.expectEqualStrings(expected_host, result.host);
    try origin_future.await(io);
    try target_future.await(io);
}

test "successful real CAR handling records only the handler boundary" {
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var fixture_arena = std.heap.ArenaAllocator.init(testing.allocator);
    defer fixture_arena.deinit();
    const fixture_alloc = fixture_arena.allocator();
    const record = "offline backfill handler timing fixture";
    const record_cid = try zat.cbor.Cid.forDagCbor(fixture_alloc, record);
    var tree = zat.mst.Mst.init(fixture_alloc);
    try tree.put("app.bsky.feed.post/3k2abcdefghij", record_cid);
    const data_cid = try tree.rootCid();
    const keypair = try zat.Keypair.fromSecretKey(.p256, @splat(17));
    const embedded_did = try keypair.did(fixture_alloc);
    const signed = try zat.signCommit(fixture_alloc, .{
        .did = embedded_did,
        .rev = "3k2abcdefghij",
        .data = data_cid,
    }, &keypair);
    var blocks: std.ArrayList(zat.car.Block) = .empty;
    try blocks.append(fixture_alloc, .{ .cid_raw = signed.cid.raw, .data = signed.bytes });
    try tree.collectBlocks(&blocks);
    try blocks.append(fixture_alloc, .{ .cid_raw = record_cid.raw, .data = record });
    const car_bytes = try zat.car.writeAlloc(fixture_alloc, .{
        .roots = &.{signed.cid},
        .blocks = blocks.items,
    });

    var listener = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable;
    defer listener.deinit(io);
    const Fixture = struct {
        fn serve(server: *Io.net.Server, task_io: Io, body: []const u8) !void {
            var stream = try server.accept(task_io);
            defer stream.close(task_io);
            var read_buf: [2048]u8 = undefined;
            var write_buf: [4096]u8 = undefined;
            var reader = stream.reader(task_io, &read_buf);
            var writer = stream.writer(task_io, &write_buf);
            var http_server = std.http.Server.init(&reader.interface, &writer.interface);
            var request = try http_server.receiveHead();
            try request.respond(body, .{
                .status = .ok,
                .keep_alive = false,
                .extra_headers = &.{.{ .name = "content-type", .value = "application/vnd.ipld.car" }},
            });
        }
    };
    var server_future = try io.concurrent(Fixture.serve, .{ &listener, io, car_bytes });
    defer _ = server_future.cancel(io) catch {};

    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();
    var data_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined;
    const data_dir_len = try tmp.dir.realPath(io, &data_dir_buf);
    const data_dir = data_dir_buf[0..data_dir_len];
    var archive = try archive_mod.Archive.init(testing.allocator, io, data_dir);
    defer archive.deinit();
    var url_buf: [96]u8 = undefined;
    const url = try std.fmt.bufPrint(&url_buf, "http://127.0.0.1:{d}", .{listener.socket.address.getPort()});
    var instrumentation: metrics_mod.Stats = .{};
    const result = try downloadOne(
        testing.allocator,
        testing.allocator,
        io,
        data_dir,
        url,
        "did:plc:authoritative-listrepos-did",
        &archive,
        &instrumentation,
        observability.root_context,
    );
    defer freeResult(testing.allocator, result);

    try testing.expectEqual(repo_store.Status.complete, result.status);
    try testing.expectEqual(@as(u64, 1), result.rows);
    try testing.expectEqual(@as(u64, 1), instrumentation.backfill_handle_repo_duration_count.load(.monotonic));
    try testing.expectEqual(@as(u64, 0), instrumentation.backfill_completion_queued_total.load(.monotonic));
    try server_future.await(io);
}