znippy_plugin_git/

exploded.rs

//! **§14's exploded objects: the CONTRACT — who hands a resolved object over,
//! and what the counters mean.**
//!
//! The store itself is [`crate::exploded_arrow`]: one Arrow IPC table, one row
//! per object, payload in the row. This module is what both ends agree on —
//! [`PayloadSink`] and the sinks that are not a store ([`NoSink`],
//! [`CaptureOne`]), plus [`ExplodedStats`], so no caller can tell the medium
//! apart by its instruments.
//!
//! # What §14 decided, and what it cost
//!
//! §14 fixed the shape — *the verbatim pack bytes are the truth and the resolved
//! objects are a derived side table* — and left one sub-question open, "resolve
//! eagerly or lazily?". **Decided 2026-08-08: EAGER**, built by the background
//! indexer, with disk explicitly not a consideration ("we don't care if disk is
//! tripled — we want wall speed minimum"). So there is no size cap, no threshold
//! and no lazy fallback that skips an object to save space; what an operator can
//! choose is [`crate::exploded_arrow::ExplodePolicy`], and that is a setting
//! rather than a heuristic.
//!
//! ```text
//!   push ─► verbatim bytes fsynced ─► ack        ← the truth, §14
//!                    │
//!                    └─ channel ─► indexer ─► resolve ─► ┬─► objects table  (oid → extent)
//!                                                        └─► THIS SINK ─► the Arrow table
//! ```
//!
//! **The decision was right and the first medium was wrong.** MEASURED on
//! `linux.git`, oden 2026-08-13, 11 697 976 objects: 6.4 GB of verbatim pack
//! resolves to **16.8 GB** of content — 2.6×, inside the 3× that "tripled"
//! budgeted — and redb held those 16.8 GB in a **204 GB** file, 12× page churn,
//! while serialising the `gatling_for_each` fan-out behind one write transaction
//! at 96.8% of a single core. The eager decision stands; redb is gone from this
//! path.
//!
//! # Three readers, and the third is why it is not optional
//!
//! | reader | had to do | now does |
//! |---|---|---|
//! | a content read | resolve a delta chain, inflate its base | **one point lookup** |
//! | a thin pack's external base ([`crate::resolve::BaseSource`]) | read and re-resolve a whole pack | **one point lookup** |
//! | `Derived::graph` / `reachable()` **after a clean reopen** | nothing — the payloads were stored nowhere | **a scan of the commits** |
//!
//! The third row is a correctness hole, not a speed one. Commit and tree
//! payloads were held only in RAM, so a store that shut down **cleanly** came
//! back with every pack's `indexed` bit legitimately set, nothing re-queued, and
//! an empty commit graph — MEASURED on oden 2026-08-08, 2687 rows and **0 of 551
//! commits**, with `reachable()` quietly returning a commit alone instead of its
//! closure and `gc` seeing an empty live set.
//!
//! # Droppable, and that is not a caveat
//!
//! Every row is re-derivable from the verbatim pack bytes, so the table can be
//! deleted at any time without consulting a client:
//! [`crate::git_ops::Absorber::adopt_journal`] compares its row count against
//! the `objects` table's and, if it is short, declares **no** pack absorbed — so
//! every pack is re-queued and re-exploded. Absent means fall back and rebuild;
//! it never means wrong. That is §13.12's `indexed`-bit logic applied unchanged
//! to a derived table, and
//! [`crate::git_ops::tests::dropping_the_exploded_table_and_reopening_still_answers`]
//! asserts it by deleting the file.
//!
//! # What it costs, measured
//!
//! MEASURED on oden 2026-08-08, release, one real 2687-object / 5.4 MiB pack,
//! four runs per column, **1-minute loadavg 1.87–2.04**. The two columns are the
//! same binary with the sink swapped to [`NoSink`] and the fold stubbed out, so
//! nothing but the table differs between them.
//!
//! | | without the table | with it | |
//! |---|---:|---:|---:|
//! | **ack** (`put_pack` returns) | 31.0–34.2 ms | 31.0–31.3 ms | **unchanged** |
//! | **drain** (one pack absorbed) | 166.7–168.6 ms | 316.2–329.8 ms | 1.9x |
//! | **indexer throughput** | 15 940–16 120 rows/s | 8 146–8 499 rows/s | **0.51x** |
//!
//! **The ack path is unchanged by construction, not by measurement.** `put_pack`
//! walks, checks the closure, appends verbatim, fsyncs twice and queues 24 bytes,
//! and not one of those lines touches this. The measured ack ranges overlap and
//! the wider one is the *left* column, which is how a null result looks; no ack
//! figure here is evidence of anything.
//!
//! Those figures are redb's. They are kept because the ack column is a
//! construction argument that still holds, and the drain column is the honest
//! record of what the eager decision cost when it was taken — not a claim about
//! what the Arrow table costs, which has not been measured on that pack.

use std::sync::Mutex;

use anyhow::Result;

use crate::object::GitObjectKind;

/// Where every object the resolver produces is handed over.
///
/// A **sink** rather than a return value: [`crate::resolve::resolve_walked`]
/// already drops a blob payload the moment nothing else in the pack deltas
/// against it, and returning every payload instead would mean the whole pack
/// inflated in RAM at once. Streaming it out keeps the resolver's peak footprint
/// exactly what it was before this table existed.
pub trait PayloadSink: Sync {
    /// One resolved object: its oid, the type its chain resolves to, and the
    /// inflated payload. Called **once per pack entry**, never for a delta's
    /// intermediate state.
    fn explode(&self, oid: &[u8], kind: GitObjectKind, payload: &[u8]) -> Result<()>;

    /// **Will this sink read the payload of an object of this kind?**
    ///
    /// The resolver asks *before* it produces one, and does not call
    /// [`explode`](Self::explode) at all when the answer is `no`. That is not
    /// an optimisation the resolver invented: since lane Q the shipped policy
    /// is [`ExplodePolicy::Commits`](crate::exploded_arrow::ExplodePolicy),
    /// which drops trees and blobs — 90 % of the objects and essentially all of
    /// the bytes — on the sink's own doorstep. Asking first is what lets the
    /// resolver hand its objects over **in entry order** without buffering
    /// those bytes to do it, and the ordered hand-over is what makes the stream
    /// the same at one worker and at four.
    ///
    /// The default is `yes` for everything, which is the old behaviour exactly.
    /// A sink that answers `no` is responsible for its own counters — it will
    /// not be told about the object at all.
    fn wants(&self, _kind: GitObjectKind) -> bool {
        true
    }

    /// Write down whatever is buffered. Called by the resolver when the memory
    /// watchdog says the box is short ([`crate::memguard`]) and at the end of a
    /// pack; a sink that buffers nothing has nothing to do.
    fn flush(&self) -> Result<()> {
        Ok(())
    }
}

/// A sink that keeps nothing. What every caller that only wants oids passes, and
/// what makes "this resolve built no exploded rows" a visible choice rather than
/// an omission.
pub struct NoSink;

impl PayloadSink for NoSink {
    fn explode(&self, _oid: &[u8], _kind: GitObjectKind, _payload: &[u8]) -> Result<()> {
        Ok(())
    }
    /// It keeps nothing, so it is never handed anything — and a resolve with
    /// `NoSink` therefore copies no payload at all on its way out.
    fn wants(&self, _kind: GitObjectKind) -> bool {
        false
    }
}

/// A sink that keeps **one** object: the fallback path's, for when the table
/// cannot answer and the content has to come back out of the verbatim truth.
///
/// It exists because [`crate::resolve::Resolved::payload`] is not a complete
/// answer to "what does this object contain" and never was — the resolver drops
/// a blob payload the moment nothing else in the pack deltas against it, so
/// `payload` is `None` for most blobs in most packs. Re-deriving a blob's
/// content by reading that field therefore returned `None` for exactly the
/// commonest object in a repository. Through the sink the payload is seen before
/// it is dropped, so the fallback answers for every object the pack holds.
pub struct CaptureOne<'a> {
    want: &'a [u8],
    got: Mutex<Option<(GitObjectKind, Vec<u8>)>>,
}

impl<'a> CaptureOne<'a> {
    pub fn new(want: &'a [u8]) -> Self {
        Self {
            want,
            got: Mutex::new(None),
        }
    }

    /// What the resolve produced for that oid, if it produced it.
    pub fn take(self) -> Option<(GitObjectKind, Vec<u8>)> {
        self.got.into_inner().ok().flatten()
    }

    /// Has the wanted oid gone past? Lets a scan stop early.
    pub fn seen(&self) -> bool {
        self.got.lock().map(|g| g.is_some()).unwrap_or(false)
    }
}

impl PayloadSink for CaptureOne<'_> {
    fn explode(&self, oid: &[u8], kind: GitObjectKind, payload: &[u8]) -> Result<()> {
        if oid == self.want
            && let Ok(mut g) = self.got.lock()
        {
            *g = Some((kind, payload.to_vec()));
        }
        Ok(())
    }
}

/// Counters, all of them **applied output**: rows that exist, lookups that
/// happened. Nothing here is configuration echoed back.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct ExplodedStats {
    /// Rows on disk plus rows buffered for the next flush.
    pub rows: u64,
    /// Rows this process has committed.
    pub written: u64,
    /// Content reads answered **from this table**.
    pub served: u64,
    /// Content reads that fell through and re-resolved a whole pack. The number
    /// that distinguishes "the table answered" from "the answer happened to be
    /// right", which identical bytes cannot.
    pub rederived: u64,
    /// Content reads nothing could answer.
    pub absent: u64,
}