plugins/native/znippy-plugin-git/src/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
}
}