znippy_plugin_git/

indexer.rs

//! The shared index path: **one indexer per account, off the ack path.**
//!
//! Every [`ArchiveWrite`] arm uses this. It is not per-impl and it is not part
//! of the durability contract — it is what happens *after* `append` has already
//! returned to the pushing client.
//!
//! ```text
//!   push ──► ArchiveWrite::append ──► ack to client
//!                    │
//!                    └─ (pack_id, offset, len) ──► account's channel
//!                                                        │
//!                                       ┌────────────────┘  worker SLEEPS on
//!                                       ▼                   recv() when empty
//!                                  drain a batch
//!                                       │
//!                                  gatling fork-join  (NEVER rayon — LAW 3)
//!                                       │           one pack row per job
//!                                       ▼
//!                                  ObjectAbsorb   ── the pack's OBJECT rows,
//!                                       │            oids and all
//!                                       ▼
//!                                  Arrow index tables + "indexed" bit
//! ```
//!
//! # One indexer per account
//!
//! Each account gets its **own** [`AccountIndexer`]: its own channel, its own
//! worker. An account pushing a monorepo cannot delay another account's small
//! push, because there is no queue and no worker between them to contend for.
//! The target box is 192 cores / 8 TB; an idle account costs a **parked thread
//! and nothing else** — the worker blocks in `recv()`, it does not poll and it
//! does not spin. `idle_indexers_sleep_they_do_not_spin` measures that as CPU
//! time rather than trusting the sentence.
//!
//! (This is deliberately *not* gunnar's `P-004` situation. There the fan-out is
//! thousands of concurrent short-lived **requests**, which is why the serve path
//! pins `thread_limit: Some(1)`. Here it is long-lived, mostly-idle **accounts**,
//! and a parked thread per account is the cheap answer.)
//!
//! # Zero-copy handoff
//!
//! [`IndexJob`] is `{ pack_id, offset, len }` — 24 bytes, `Copy`, no pointer
//! into the pack. **Extents cross the channel, never buffers.** The worker
//! `pread`s the extent back when it gets to it, so a 2 GiB push costs the
//! channel 24 bytes and the ack path zero copies.
//!
//! # Fan-out inside a drain is gatling
//!
//! A drained batch is handed to
//! [`gatling_forkjoin::gatling_for_each`](znippy_zoomies::gatling_forkjoin::gatling_for_each):
//! N workers self-dispatch off one atomic cursor, no barrier, `std::thread::scope`.
//! **rayon is banned across this constellation (LAW 3)** and `rayon_free_law.rs`
//! enforces it.
//!
//! # The object-level half
//!
//! A pack row is derivable from the extent alone — version, object count, a hash
//! of the stored bytes. **Object rows are not.** They need the pack walked, its
//! delta chains applied and every oid hashed, which is git-shaped work this file
//! must not know how to do. So the drain calls *out*, through
//! [`ObjectAbsorb`], and [`GitStore`](crate::git_ops::GitStore) implements it.
//!
//! The absorber it calls is **the same object a falling-back read calls**, not a
//! background copy of it (LAW 5): one absorb, one gate, one place the `objects`
//! table is written. The two paths cannot drift because there is only one of
//! them.
//!
//! Ordering inside one drained job is fixed and it is the whole safety argument:
//!
//! ```text
//!   pack row built  ─►  objects absorbed  ─►  indexed bit set
//! ```
//!
//! The bit goes up **last**, so it cannot be set over an index that does not yet
//! hold the pack's objects. An absorb that fails leaves the bit clear and the
//! error counted ([`AccountIndexer::absorb_failures`]) — the pack stays on the
//! fallback path, which is slow and right, rather than being declared indexed,
//! which would be fast and wrong.
//!
//! # Why an early read cannot be wrong
//!
//! The index tables are built **last**. A read arriving between the ack and the
//! index is therefore reading an archive whose index does not mention the pack.
//! Rather than let it conclude "absent", every pack carries an **indexed bit**,
//! which is membership in the published-row map — one structure, so the bit and
//! the row cannot drift apart. [`AccountIndexer::lookup`] returns
//! [`Lookup::Indexed`] only when the bit is set, and otherwise hands back the
//! journal's extent list so the caller falls back to **scanning** — slower,
//! never wrong. For
//! [`FastWriter`](crate::archive_write::FastWriter) there is no journal to fall
//! back to and the answer is [`Lookup::Unknowable`]: one more thing that arm
//! does not promise.

use std::collections::HashMap;
use std::fs::File;
use std::os::unix::fs::FileExt;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::mpsc::{Receiver, Sender, TryRecvError, channel};
use std::sync::{Arc, Condvar, Mutex};

use anyhow::{Result, anyhow};
use sha1::{Digest, Sha1};
use znippy_common::arrow::array::{ArrayRef, StringArray, UInt32Array, UInt64Array};
use znippy_common::arrow::datatypes::{DataType, Field, Schema, SchemaRef};
use znippy_common::arrow::record_batch::RecordBatch;
use znippy_zoomies::background::Job;
use znippy_zoomies::gatling_forkjoin::gatling_for_each;

use crate::archive_write::{ArchiveWrite, Extent, read_journal};

/// At most this many jobs are folded into one gatling fan-out. Bounds the
/// latency an early-arriving job pays behind a burst.
const MAX_DRAIN_BATCH: usize = 4096;

/// One unit of index work. **Offsets and extents only — never a buffer.**
///
/// 24 bytes, `Copy`, `'static`: nothing in it borrows the pushed pack, so the
/// ack path can drop the client's buffer the instant `append` returns.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct IndexJob {
    /// Dense id assigned at append time; the key the "indexed" bit is kept under.
    pub pack_id: u64,
    /// Where the verbatim pack bytes start in the archive.
    pub offset: u64,
    /// How many bytes.
    pub len: u64,
}

/// One row of the built index — everything derivable from the pack bytes, which
/// is exactly the work that was kept off the ack path.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct IndexRow {
    pub pack_id: u64,
    pub offset: u64,
    pub len: u64,
    /// Packfile format version from the header (`2` or `3`), `0` if unparseable.
    pub version: u32,
    /// Object count from the header, `0` if unparseable.
    pub object_count: u32,
    /// SHA-1 over the verbatim pack bytes, hex. Not the pack's own trailing
    /// checksum — a checksum *of what was stored*, so a later scrub can prove
    /// the archive still holds what was acked.
    pub sha1: String,
}

/// The index table's Arrow schema.
pub fn index_schema() -> SchemaRef {
    Arc::new(Schema::new(vec![
        Field::new("pack_id", DataType::UInt64, false),
        Field::new("blob_offset", DataType::UInt64, false),
        Field::new("blob_size", DataType::UInt64, false),
        Field::new("pack_version", DataType::UInt32, false),
        Field::new("object_count", DataType::UInt32, false),
        Field::new("pack_sha1", DataType::Utf8, false),
    ]))
}

fn rows_to_batch(rows: &[IndexRow]) -> Result<RecordBatch> {
    let ids: ArrayRef = Arc::new(UInt64Array::from_iter_values(rows.iter().map(|r| r.pack_id)));
    let offs: ArrayRef = Arc::new(UInt64Array::from_iter_values(rows.iter().map(|r| r.offset)));
    let lens: ArrayRef = Arc::new(UInt64Array::from_iter_values(rows.iter().map(|r| r.len)));
    let vers: ArrayRef = Arc::new(UInt32Array::from_iter_values(rows.iter().map(|r| r.version)));
    let cnts: ArrayRef = Arc::new(UInt32Array::from_iter_values(
        rows.iter().map(|r| r.object_count),
    ));
    let sha: ArrayRef = Arc::new(StringArray::from_iter_values(
        rows.iter().map(|r| r.sha1.as_str()),
    ));
    RecordBatch::try_new(index_schema(), vec![ids, offs, lens, vers, cnts, sha])
        .map_err(|e| anyhow!("index batch: {e}"))
}

/// Build one index row by reading the extent back out of the archive.
///
/// This is the work that is *not* on the ack path: a `pread` of the whole pack
/// plus a SHA-1 over it. Runs on a gatling worker.
fn build_row(archive: &File, job: IndexJob) -> IndexRow {
    // **Streamed, never the whole pack in one buffer.** This was
    // `vec![0u8; job.len]` + one `read_exact_at` — for torvalds/linux an
    // 8.12 GiB allocation on the 16 GB appliance, held while the SHA-1 ran,
    // on top of everything the absorb holds. One reused 1 MiB window now.
    const WINDOW: usize = 1 << 20;
    let unreadable = IndexRow {
        pack_id: job.pack_id,
        offset: job.offset,
        len: job.len,
        version: 0,
        object_count: 0,
        sha1: String::new(),
    };
    let mut buf = vec![0u8; WINDOW.min(job.len as usize).max(1)];
    let mut h = Sha1::new();
    let mut head = [0u8; 12];
    let mut done = 0u64;
    while done < job.len {
        let want = ((job.len - done) as usize).min(buf.len());
        if archive.read_exact_at(&mut buf[..want], job.offset + done).is_err() {
            return unreadable;
        }
        if done < 12 {
            let n = (12 - done as usize).min(want);
            head[done as usize..done as usize + n].copy_from_slice(&buf[..n]);
        }
        h.update(&buf[..want]);
        done += want as u64;
    }
    // P-4: a malformed or hostile entry never panics — a short or non-`PACK`
    // buffer degrades to zeroes and the row still writes.
    let (version, object_count) = if job.len >= 12 && &head[0..4] == b"PACK" {
        (
            u32::from_be_bytes([head[4], head[5], head[6], head[7]]),
            u32::from_be_bytes([head[8], head[9], head[10], head[11]]),
        )
    } else {
        (0, 0)
    };
    IndexRow {
        pack_id: job.pack_id,
        offset: job.offset,
        len: job.len,
        version,
        object_count,
        sha1: hex::encode(h.finalize()),
    }
}

/// **The object-level ingress**, called by the drain and by nothing else here.
///
/// The indexer knows extents; it does not know what a git object is, and it must
/// not learn — the split is what keeps the pack grammar out of this file and the
/// channel out of the store's. The implementor is
/// [`GitStore`](crate::git_ops::GitStore)'s absorber, which is the *same* object
/// a read falls back to, so the background path and the fallback path are one
/// path (LAW 5).
///
/// The argument is an [`IndexJob`] — 24 bytes, `Copy`, no buffer. Zero-copy
/// handoff survives the extra hop.
pub trait ObjectAbsorb: Send + Sync {
    /// Absorb this pack's objects into the object-level index.
    ///
    /// Must be **idempotent for a pack already absorbed**: the ack path records a
    /// pack as pending and hands it to the channel in that order, and a fast
    /// worker can arrive at either end first.
    fn absorb(&self, job: IndexJob) -> Result<()>;
}

/// What a read gets when it asks for a pack.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Lookup {
    /// The index has it. O(1), and the row carries everything derived.
    Indexed(Box<IndexRow>),
    /// Not indexed **yet**. Here is the journal's extent list — the durable
    /// record written on the ack path — so the caller can scan. Slower, never
    /// wrong.
    ScanJournal(Vec<Extent>),
    /// Not indexed and there is no journal to scan
    /// ([`FastWriter`](crate::archive_write::FastWriter)). Nothing can answer.
    Unknowable,
}

#[derive(Default)]
struct Built {
    /// The index tables, built last.
    batches: Vec<RecordBatch>,
    /// **The indexed bit, and the row it publishes — one structure.**
    ///
    /// Membership is the bit: present means "the index can answer for this
    /// pack", absent means "fall back to scanning". It is deliberately *not* a
    /// separate bitmap beside a row map. Two structures would need a guard
    /// watching them agree; one structure cannot disagree with itself (LAW 5 —
    /// fix by construction, do not add a guard over two copies). A pack is
    /// published by exactly one `insert`, under one lock, after its row is
    /// complete **and after its objects are absorbed**, so no reader can ever
    /// see a set bit with no row — or no object — behind it.
    published: HashMap<u64, IndexRow>,
    /// Packs whose [`ObjectAbsorb::absorb`] failed. Their bits stayed clear, so
    /// they are still answerable by the fallback; this is how many times that
    /// happened, which is otherwise invisible from outside the worker.
    absorb_failures: u64,
    /// The most recent absorb error, verbatim. A count with no message is a
    /// number nobody can act on.
    last_absorb_error: Option<String>,
}

struct Progress {
    /// Jobs submitted minus jobs committed. `0` means the indexer is caught up.
    outstanding: Mutex<u64>,
    caught_up: Condvar,
}

/// One account's indexer: its own channel, its own worker, its own tables.
pub struct AccountIndexer {
    account: String,
    tx: Option<Sender<IndexJob>>,
    built: Arc<Mutex<Built>>,
    progress: Arc<Progress>,
    journal: Option<PathBuf>,
    worker: Option<Job<()>>,
    /// The repository is being deleted: the worker drops every queued job
    /// unabsorbed and `Drop` joins it in milliseconds rather than hours.
    cancel: Arc<AtomicBool>,
}

impl AccountIndexer {
    /// Start an indexer for `account` reading extents back out of `archive`.
    /// `journal` is the ack-path durable record a not-yet-indexed read falls
    /// back to scanning; `None` for a writer that keeps none.
    ///
    /// **Pack rows only.** For the object rows as well, see
    /// [`start_with_absorber`](AccountIndexer::start_with_absorber) — a writer
    /// bench has no git store behind it and wants exactly this.
    pub fn start(account: &str, archive: Arc<File>, journal: Option<PathBuf>) -> Self {
        Self::start_with_absorber(account, archive, journal, None)
    }

    /// The same indexer with the **object-level half** wired in: each drained
    /// job's objects are absorbed through `absorber` before the pack's indexed
    /// bit goes up.
    pub fn start_with_absorber(
        account: &str,
        archive: Arc<File>,
        journal: Option<PathBuf>,
        absorber: Option<Arc<dyn ObjectAbsorb>>,
    ) -> Self {
        let (tx, rx): (Sender<IndexJob>, Receiver<IndexJob>) = channel();
        let built = Arc::new(Mutex::new(Built::default()));
        let progress = Arc::new(Progress {
            outstanding: Mutex::new(0),
            caught_up: Condvar::new(),
        });
        let w_built = built.clone();
        let w_progress = progress.clone();
        let cancel = Arc::new(AtomicBool::new(false));
        let w_cancel = cancel.clone();
        // `gatling::background::Job` — the constellation's ONE sanctioned home
        // for a long-lived background thread outside the engine itself. A raw
        // `thread::spawn` or `thread::Builder` here trips
        // `rayon_free_law::only_the_shared_gatling_engine_owns_a_worker_pool`,
        // and rightly: every thread in this tree originates in gatling. This is
        // gatling's depth-1 shape — one worker, joined at close — and the
        // fan-out *inside* it is `gatling_for_each`.
        let worker =
            Job::spawn(move || index_worker(archive, rx, w_built, w_progress, absorber, w_cancel));
        Self {
            account: account.to_string(),
            tx: Some(tx),
            built,
            progress,
            journal,
            worker: Some(worker),
            cancel,
        }
    }

    /// The account this indexer belongs to.
    pub fn account(&self) -> &str {
        &self.account
    }

    /// Hand an extent over. Returns immediately: 24 bytes onto a channel.
    pub fn submit(&self, job: IndexJob) -> Result<()> {
        {
            let mut o = self
                .progress
                .outstanding
                .lock()
                .map_err(|_| anyhow!("indexer progress poisoned"))?;
            *o += 1;
        }
        self.tx
            .as_ref()
            .ok_or_else(|| anyhow!("indexer already closed"))?
            .send(job)
            .map_err(|_| anyhow!("indexer worker is gone"))
    }

    /// Tell the worker to stop: every job still queued is dropped unabsorbed
    /// (its pack keeps a clear bit — nothing is declared indexed), and the
    /// absorber's own cancel flag is what aborts the pack in flight.
    pub fn cancel(&self) {
        self.cancel.store(true, Ordering::SeqCst);
    }

    pub fn is_cancelled(&self) -> bool {
        self.cancel.load(Ordering::SeqCst)
    }

    /// Block until this account's index has caught up with everything submitted
    /// so far. Used by tests and by the bench's "with index" column.
    pub fn wait_caught_up(&self) {
        let mut o = self.progress.outstanding.lock().unwrap();
        while *o > 0 {
            o = self.progress.caught_up.wait(o).unwrap();
        }
    }

    /// Is this pack's **indexed bit** set?
    pub fn is_indexed(&self, pack_id: u64) -> bool {
        self.built.lock().unwrap().published.contains_key(&pack_id)
    }

    /// Number of index rows built so far.
    pub fn rows(&self) -> usize {
        self.built.lock().unwrap().published.len()
    }

    /// How many packs failed [`ObjectAbsorb::absorb`] in this indexer. Their
    /// indexed bits are clear and their reads still fall back.
    pub fn absorb_failures(&self) -> u64 {
        self.built.lock().unwrap().absorb_failures
    }

    /// The most recent absorb error, if any.
    pub fn last_absorb_error(&self) -> Option<String> {
        self.built.lock().unwrap().last_absorb_error.clone()
    }

    /// The built index tables.
    pub fn tables(&self) -> Vec<RecordBatch> {
        self.built.lock().unwrap().batches.clone()
    }

    /// Answer a read, honestly, whether or not the index is ready.
    pub fn lookup(&self, pack_id: u64) -> Lookup {
        {
            // The bit IS the row's presence. One lookup, one truth.
            let b = self.built.lock().unwrap();
            if let Some(r) = b.published.get(&pack_id) {
                return Lookup::Indexed(Box::new(r.clone()));
            }
        }
        match self.journal.as_deref() {
            Some(p) => match read_journal(p) {
                Ok(extents) => Lookup::ScanJournal(extents),
                Err(_) => Lookup::Unknowable,
            },
            None => Lookup::Unknowable,
        }
    }

    /// Close the channel and join the worker, returning the final tables.
    pub fn finish(mut self) -> Vec<RecordBatch> {
        self.tx = None;
        if let Some(w) = self.worker.take() {
            let _ = w.join();
        }
        self.built.lock().unwrap().batches.clone()
    }
}

impl Drop for AccountIndexer {
    /// **Cancel, then join — never join a worker that may be hours into a
    /// pack or parked by the memory watchdog.** The queued jobs are dropped
    /// unabsorbed and the pack in flight is aborted at its next checkpoint;
    /// their bits stay clear, so the next open re-queues them from the
    /// journal (`resumed 0%`). Nobody waits on the indexer, and that includes
    /// the process shutting down and a `repo delete`.
    fn drop(&mut self) {
        self.cancel.store(true, Ordering::SeqCst);
        self.tx = None;
        if let Some(w) = self.worker.take() {
            let _ = w.join();
        }
    }
}

/// The worker. **Sleeps when its channel is empty.**
///
/// `recv()` blocks — after a bounded spin `std`'s channel parks the thread on a
/// futex, so an idle account burns no CPU at all. It wakes on the first job,
/// then drains everything already queued behind it (`try_recv` until empty, up
/// to [`MAX_DRAIN_BATCH`]) so a burst costs one fan-out rather than one per job.
fn index_worker(
    archive: Arc<File>,
    rx: Receiver<IndexJob>,
    built: Arc<Mutex<Built>>,
    progress: Arc<Progress>,
    absorber: Option<Arc<dyn ObjectAbsorb>>,
    cancel: Arc<AtomicBool>,
) {
    loop {
        // ── SLEEP HERE ──────────────────────────────────────────────────────
        let Ok(first) = rx.recv() else { return };
        let mut batch = Vec::with_capacity(64);
        batch.push(first);
        loop {
            if batch.len() >= MAX_DRAIN_BATCH {
                break;
            }
            match rx.try_recv() {
                Ok(j) => batch.push(j),
                Err(TryRecvError::Empty) | Err(TryRecvError::Disconnected) => break,
            }
        }

        // ── cancelled: the repository is going. Count the batch off and
        //    absorb nothing — a clear bit is the honest state of a pack whose
        //    directory is about to be removed. ──────────────────────────────
        if cancel.load(Ordering::SeqCst) {
            let mut o = progress.outstanding.lock().unwrap();
            *o = o.saturating_sub(batch.len() as u64);
            if *o == 0 {
                progress.caught_up.notify_all();
            }
            continue;
        }

        // The memory watchdog (crate::memguard) is the absorber's: it checks
        // before every pack it absorbs and every few thousand objects inside
        // one. This worker holds nothing between packs, so there is nothing
        // for it to flush and nothing for it to check.

        // ── gatling fork-join. NEVER rayon (LAW 3). ─────────────────────────
        let n = batch.len();
        let arch = archive.as_ref();
        let jobs = &batch;
        let rows = gatling_for_each(n, 0, |i| build_row(arch, jobs[i]));

        // ── the OBJECT-level half, out through `ObjectAbsorb` ───────────────
        //
        // Serial across packs on purpose: absorbing appends to one `objects`
        // table and re-folds one commit graph, so the parallelism that pays
        // here is the gatling fan-out above (one pread + one SHA-1 per pack),
        // not N threads contending for the same two writers. A pack whose
        // absorb fails is dropped from `ok` and keeps its bit clear.
        let mut ok: Vec<IndexRow> = Vec::with_capacity(rows.len());
        let mut failed: Vec<String> = Vec::new();
        for (job, row) in batch.iter().zip(rows) {
            match absorber.as_deref() {
                Some(a) => match a.absorb(*job) {
                    Ok(()) => ok.push(row),
                    Err(e) => failed.push(format!("pack {}: {e:#}", job.pack_id)),
                },
                None => ok.push(row),
            }
        }

        // ── index tables built LAST, after the bytes are down AND the
        //    objects are in ───────────────────────────────────────────────────
        let encoded = if ok.is_empty() {
            None
        } else {
            // A batch that cannot be encoded leaves the bits unset: every one of
            // its packs stays on the scan path.
            rows_to_batch(&ok).ok()
        };
        {
            let mut b = built.lock().unwrap();
            if let Some(rb) = encoded {
                b.batches.push(rb);
                for r in ok {
                    // Publishing the row IS setting the bit — one insert, one
                    // lock, after the row is complete and its objects are
                    // absorbed.
                    b.published.insert(r.pack_id, r);
                }
            }
            b.absorb_failures += failed.len() as u64;
            if let Some(last) = failed.pop() {
                b.last_absorb_error = Some(last);
            }
        }

        let mut o = progress.outstanding.lock().unwrap();
        *o = o.saturating_sub(n as u64);
        if *o == 0 {
            progress.caught_up.notify_all();
        }
    }
}

/// One indexer per account, created on first push from that account.
pub struct IndexerPool {
    archive: Arc<File>,
    journal: Option<PathBuf>,
    /// Shared by every account's worker. One store, one object table — the
    /// per-account split is about *queueing*, not about where the rows land.
    absorber: Option<Arc<dyn ObjectAbsorb>>,
    accounts: Mutex<HashMap<String, Arc<AccountIndexer>>>,
}

impl IndexerPool {
    /// `archive` is opened read-only by the workers to `pread` extents back.
    /// Pack rows only; see [`with_absorber`](IndexerPool::with_absorber).
    pub fn new(archive: &Path, journal: Option<PathBuf>) -> Result<Self> {
        Self::build(archive, None, journal, None)
    }

    /// The same pool with the object-level half wired in.
    ///
    /// `shared` is the writer's own open file description on the same blob file
    /// ([`ArchiveWrite::blobs_shared`]). Passing it is what keeps an open store
    /// at **one** descriptor on `objects.pack` instead of three: the workers
    /// only ever `read_exact_at`, so a description they share with the writer
    /// and with the absorber behaves identically to one of their own. `None`
    /// falls back to opening it, which is what a bench with no store does.
    pub fn with_absorber(
        archive: &Path,
        shared: Option<Arc<File>>,
        journal: Option<PathBuf>,
        absorber: Arc<dyn ObjectAbsorb>,
    ) -> Result<Self> {
        Self::build(archive, shared, journal, Some(absorber))
    }

    fn build(
        archive: &Path,
        shared: Option<Arc<File>>,
        journal: Option<PathBuf>,
        absorber: Option<Arc<dyn ObjectAbsorb>>,
    ) -> Result<Self> {
        let f = match shared {
            Some(f) => f,
            None => Arc::new(
                File::open(archive)
                    .map_err(|e| anyhow!("indexer: open {}: {e}", archive.display()))?,
            ),
        };
        Ok(Self {
            archive: f,
            journal,
            absorber,
            accounts: Mutex::new(HashMap::new()),
        })
    }

    /// This account's indexer, started on first use.
    pub fn indexer(&self, account: &str) -> Arc<AccountIndexer> {
        let mut m = self.accounts.lock().unwrap();
        m.entry(account.to_string())
            .or_insert_with(|| {
                Arc::new(AccountIndexer::start_with_absorber(
                    account,
                    self.archive.clone(),
                    self.journal.clone(),
                    self.absorber.clone(),
                ))
            })
            .clone()
    }

    /// How many accounts have an indexer.
    pub fn accounts(&self) -> usize {
        self.accounts.lock().unwrap().len()
    }

    /// Block until every account's index has caught up.
    pub fn wait_caught_up(&self) {
        let all: Vec<Arc<AccountIndexer>> =
            self.accounts.lock().unwrap().values().cloned().collect();
        for a in all {
            a.wait_caught_up();
        }
    }

    /// Cancel every account's worker; see [`AccountIndexer::cancel`].
    pub fn cancel_all(&self) {
        let all: Vec<Arc<AccountIndexer>> =
            self.accounts.lock().unwrap().values().cloned().collect();
        for a in all {
            a.cancel();
        }
    }
}

/// **Where a reopened push path resumes assigning pack ids**: after the packs
/// this archive has already acked.
///
/// A pack id is an **ordinal**, and the journal is what makes it dense: one row
/// per acked pack, in append order, so row `i` is ordinal `i` and a new push
/// takes the next one. Starting over at `0` after a restart would hand a fresh
/// pack the ordinal of one that is already absorbed, and the absorber's whole job
/// is to skip a pack whose bit is set — so the new pack's objects would be
/// silently dropped on the floor while its bytes sat durable in the archive. That
/// is a wrong `absent`, which is the one failure mode this crate does not accept.
///
/// Derived rather than passed in, so no caller can forget it: a writer with no
/// journal ([`FastWriter`](crate::archive_write::FastWriter)) keeps no durable
/// record of its acks and so starts at `0`, which is the same thing that arm says
/// about everything else it does not promise.
///
/// **Pack rows, not raw rows.** A `gc` appends a tombstone row for a pack whose
/// every object it found dead
/// ([`retire_packs`](crate::archive_write::retire_packs)), and that row is not a
/// pack and holds no ordinal. Counting it would make the next push's id skip one
/// and disagree with the ordinal the next open derives for the same pack.
fn packs_already_acked(journal: Option<&Path>) -> Result<u64> {
    match journal {
        Some(p) if p.exists() => {
            Ok(crate::archive_write::acked_packs(&read_journal(p)?).len() as u64)
        }
        _ => Ok(0),
    }
}

/// A git server's push path: one [`ArchiveWrite`] arm plus the shared indexer.
///
/// This is what the bench drives, and the reason the indexer is *shared* rather
/// than per-impl: swapping the writer changes the durability contract and
/// nothing else.
pub struct PushPath {
    writer: Box<dyn ArchiveWrite>,
    pool: IndexerPool,
    next_pack_id: AtomicU64,
}

impl PushPath {
    /// Wrap a writer. `archive` must be the file the writer appends to;
    /// `journal` the ack-path record a pre-index read falls back to, if the
    /// writer keeps one.
    pub fn new(writer: Box<dyn ArchiveWrite>, archive: &Path, journal: Option<PathBuf>) -> Result<Self> {
        Ok(Self {
            next_pack_id: AtomicU64::new(packs_already_acked(journal.as_deref())?),
            writer,
            pool: IndexerPool::new(archive, journal)?,
        })
    }

    /// The same push path whose drain also absorbs **object** rows.
    ///
    /// This is the one a git store builds: [`push_pack`](PushPath::push_pack)
    /// still returns after the two fsyncs and nothing about the ack changes, but
    /// the job that leaves on the channel now ends in oids rather than in one
    /// pack row.
    /// **The pool reads through the writer's own descriptor.** Asked here
    /// rather than passed in by the caller, so no caller can forget it and
    /// silently put a third handle on `objects.pack`
    /// (`tests/fd_per_store.rs` counts them).
    pub fn with_absorber(
        writer: Box<dyn ArchiveWrite>,
        archive: &Path,
        journal: Option<PathBuf>,
        absorber: Arc<dyn ObjectAbsorb>,
    ) -> Result<Self> {
        let shared = writer.blobs_shared();
        Ok(Self {
            next_pack_id: AtomicU64::new(packs_already_acked(journal.as_deref())?),
            writer,
            pool: IndexerPool::with_absorber(archive, shared, journal, absorber)?,
        })
    }

    /// The arm's name, for a bench row.
    pub fn name(&self) -> &'static str {
        self.writer.name()
    }

    /// What this arm's `append` actually promises.
    pub fn durability(&self) -> &'static str {
        self.writer.durability()
    }

    /// Store a pushed pack verbatim and queue its index job.
    ///
    /// Returns `(pack_id, extent)`. Everything expensive about the pack —
    /// parsing it, hashing it, building the tables — happens after this returns.
    pub fn push_pack(&self, account: &str, bytes: &[u8]) -> Result<(u64, Extent)> {
        let (offset, len) = self.writer.append(bytes)?;
        let pack_id = self.next_pack_id.fetch_add(1, Ordering::SeqCst);
        self.pool.indexer(account).submit(IndexJob {
            pack_id,
            offset,
            len,
        })?;
        Ok((pack_id, (offset, len)))
    }

    /// This account's indexer.
    pub fn indexer(&self, account: &str) -> Arc<AccountIndexer> {
        self.pool.indexer(account)
    }

    /// The pool, for a caller that wants to drain every account.
    pub fn pool(&self) -> &IndexerPool {
        &self.pool
    }
}