use std::collections::{BTreeMap, HashMap};
use std::fs::File;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, RwLock};
use anyhow::{anyhow, bail, Context, Result};
use znippy_common::ReservedSection;
use crate::arms::StoreConfig;
use crate::exploded::ExplodedStats;
use crate::exploded_cache::ExplodedCache;
use crate::gc::{Gc, GcReport};
use crate::graph::{assign_generations, CommitNode};
use crate::index_layout::{ObjectIndex, OneTableFourColumns};
use crate::indexer::{IndexJob, ObjectAbsorb, PushPath};
use crate::object::{GitHashKind, GitObjectKind};
use crate::pack_walk::PackWalk;
use crate::read_stack::{ObjectReadStack, RebuildTriggers};
use crate::refs::{RefLog, RefUpdate};
use crate::resolve::BaseSource;
pub use git_storage_trait::{Extent, GitOps, Oid, RefRow, Stored, TxId};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LookupPath {
Serial,
Batch,
}
pub const BATCH_SATURATES_AT: usize = 100;
pub fn lookup_path(n: usize) -> LookupPath {
if n == 1 {
LookupPath::Serial
} else {
LookupPath::Batch
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct PendingPack {
pack_id: u64,
offset: u64,
len: u64,
resumed: bool,
}
impl PendingPack {
fn from_job(j: IndexJob) -> Self {
Self {
pack_id: j.pack_id,
offset: j.offset,
len: j.len,
resumed: false,
}
}
}
#[derive(Default)]
struct PackState {
words: Vec<u64>,
extent: Vec<Option<Extent>>,
unabsorbed: usize,
}
impl PackState {
fn is_absorbed(&self, pack_id: u64) -> bool {
let i = pack_id as usize;
self.words
.get(i / 64)
.is_some_and(|w| w >> (i % 64) & 1 == 1)
}
fn note(&mut self, pack_id: u64, extent: Extent) {
let i = pack_id as usize;
self.grow_to(i);
if self.extent[i].is_none() {
self.extent[i] = Some(extent);
self.unabsorbed += 1;
}
}
fn grow_to(&mut self, i: usize) {
if self.words.len() <= i / 64 {
self.words.resize(i / 64 + 1, 0);
}
if self.extent.len() <= i {
self.extent.resize(i + 1, None);
}
}
fn mark_absorbed(&mut self, pack_id: u64, extent: Extent) {
let i = pack_id as usize;
self.grow_to(i);
if self.words[i / 64] >> (i % 64) & 1 == 0 {
self.words[i / 64] |= 1 << (i % 64);
if self.extent[i].is_some() {
self.unabsorbed -= 1;
}
}
self.extent[i] = Some(extent);
}
fn pending(&self) -> Vec<PendingPack> {
self.extent
.iter()
.enumerate()
.filter(|(i, e)| e.is_some() && !self.is_absorbed(*i as u64))
.map(|(i, e)| {
let (offset, len) = e.expect("filtered on Some");
PendingPack {
pack_id: i as u64,
offset,
len,
resumed: true,
}
})
.collect()
}
}
#[derive(Default)]
struct Derived {
graph: Vec<CommitNode>,
ord_graph: Arc<crate::reach::OrdGraph>,
ordinals: crate::reach::OrdinalTable,
reach: Arc<crate::reach::ReachTable>,
reach_by_row: Arc<Vec<u32>>,
token: [u8; 32],
packs: Vec<(u64, u64)>,
}
pub(crate) const ORDINAL_COMPACT_FRACTION: usize = 4;
pub(crate) const ORDINAL_COMPACT_MIN: usize = 4096;
pub(crate) const REACH_PERSIST_EVERY: u64 = 16;
pub(crate) const REACH_BUILD_CHAIN_CACHE_BYTES: usize = 128 << 20;
pub(crate) const REACH_BUILD_CHAIN_CACHE_PER_OBJECT: usize = 512;
const FAN_OUT_AT: usize = 4096;
const GRAPH_CACHE_FILE: &str = "objects.graph";
const GRAPH_CACHE_MAGIC: &[u8; 4] = b"GRPH";
const GRAPH_CACHE_HEADER: usize = 4 + 1 + 32;
const REACH_CACHE_FILE: &str = "objects.reach";
const REACH_CACHE_MAGIC: &[u8; 4] = b"RECH";
const REACH_CACHE_HEADER: usize = 4 + 1 + 4 + 8 + 8;
const ORDINAL_CACHE_FILE: &str = "objects.ordinal";
pub(crate) fn dead_oids(
seen: Vec<Vec<u8>>,
live: &std::collections::HashSet<String>,
) -> std::collections::HashSet<Vec<u8>> {
let live_raw: std::collections::HashSet<Vec<u8>> =
live.iter().filter_map(|h| hex::decode(h).ok()).collect();
seen.into_iter().filter(|o| !live_raw.contains(o)).collect()
}
fn graph_token(oids_raw: &[u8]) -> [u8; 32] {
let mut h = znippy_common::blake3::Hasher::new();
h.update(oids_raw);
*h.finalize().as_bytes()
}
const LIVE_REACH_COMMITS: usize = 512;
pub(crate) fn live_reach_policy() -> crate::reach::ReachPolicy {
static MAX: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
let max_commits = *MAX.get_or_init(|| {
crate::arms::read_env(crate::arms::ENV_REACH_COMMITS)
.and_then(|v| v.parse::<usize>().ok())
.unwrap_or(LIVE_REACH_COMMITS)
});
crate::reach::ReachPolicy { max_commits }
}
const MAX_RESOLVE_WORKERS: usize = 4;
fn emit_workers() -> usize {
static WORKERS: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
*WORKERS.get_or_init(|| {
crate::arms::read_env(crate::arms::ENV_EMIT_WORKERS)
.and_then(|v| v.parse::<usize>().ok())
.unwrap_or(MAX_RESOLVE_WORKERS)
})
}
pub(crate) struct Absorber<S: ObjectIndex = OneTableFourColumns> {
blobs: PathBuf,
reader: Arc<File>,
map: crate::archive_map::ArchiveMap,
hash: GitHashKind,
objects: ObjectReadStack<S>,
exploded: Arc<ExplodedCache>,
derived: RwLock<Derived>,
graph_read: std::sync::atomic::AtomicU64,
graph_recomputed: std::sync::atomic::AtomicU64,
graph_cache: std::sync::atomic::AtomicBool,
reach_carried: std::sync::atomic::AtomicU64,
reach_rebuilt: std::sync::atomic::AtomicU64,
reach_loaded: std::sync::atomic::AtomicU64,
reach_extended: std::sync::atomic::AtomicU64,
reach_unpersisted: std::sync::atomic::AtomicU64,
ordinal_folded: std::sync::atomic::AtomicU64,
ordinal_tail: std::sync::atomic::AtomicU64,
packs: Mutex<PackState>,
gate: Mutex<()>,
progress: Arc<crate::progress::IndexProgress>,
cancel: std::sync::atomic::AtomicBool,
label: String,
watchdog: RwLock<Arc<crate::memguard::MemWatchdog>>,
scanned: Mutex<HashMap<u64, Arc<ScannedPack>>>,
scan_build: Mutex<()>,
unindexable: Mutex<HashMap<u64, Unindexable>>,
}
#[derive(Debug, Clone)]
pub struct Unindexable {
pub pack_id: u64,
pub extent: Extent,
pub attempts: u32,
pub reason: String,
}
pub const MAX_ABSORB_ATTEMPTS: u32 = 3;
thread_local! {
static IN_ABSORB: std::cell::Cell<Option<u64>> = const { std::cell::Cell::new(None) };
}
struct InAbsorb(Option<u64>);
impl InAbsorb {
fn enter(pack_id: u64) -> Self {
InAbsorb(IN_ABSORB.with(|c| c.replace(Some(pack_id))))
}
}
impl Drop for InAbsorb {
fn drop(&mut self) {
IN_ABSORB.with(|c| c.set(self.0));
}
}
struct ScannedPack {
walk: crate::pack_walk::PackWalk,
rows: Vec<(Vec<u8>, u32, u64)>,
archive_offset: u64,
}
impl ScannedPack {
fn find(&self, oid: &[u8]) -> Option<(usize, u64)> {
self.rows
.binary_search_by(|(o, _, _)| o.as_slice().cmp(oid))
.ok()
.map(|i| (self.rows[i].1 as usize, self.rows[i].2))
}
fn row(&self, oid: &[u8]) -> Option<crate::index_layout::IndexRow> {
let (entry, size) = self.find(oid)?;
let e = &self.walk.entries[entry];
Some(crate::index_layout::IndexRow {
ordinal: u32::MAX,
offset: e.offset + self.archive_offset,
len: e.len,
obj_type: e.obj_type,
uncompressed_size: size,
delta_base: match &e.delta_base {
crate::pack_walk::DeltaBase::Offset(o) => o + self.archive_offset,
_ => 0,
},
})
}
}
pub struct GitStore<S: ObjectIndex = OneTableFourColumns> {
root: PathBuf,
blobs: PathBuf,
archive: PathBuf,
push: PushPath,
account: String,
hash: GitHashKind,
absorber: Arc<Absorber<S>>,
refs: RefLog,
ref_gate: Mutex<()>,
gc: Box<dyn Gc + Send + Sync>,
arms: StoreConfig,
}
impl GitStore<OneTableFourColumns> {
pub fn open(root: &Path, account: &str) -> Result<Self> {
Self::open_with(root, account, GitHashKind::Sha1)
}
pub fn open_with(root: &Path, account: &str, hash: GitHashKind) -> Result<Self> {
Self::open_with_arms(
root,
account,
hash,
StoreConfig::DEFAULT.with_redb_cache_bytes(crate::arms::redb_cache_bytes()?),
)
}
}
enum BaseOf {
Whole,
Inside { at: u64 },
Outside {
offset: u64,
oid: Option<Vec<u8>>,
},
}
struct ExternalBases {
by_offset: HashMap<u64, OidKey>,
held: std::collections::HashSet<OidKey>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
struct OidKey {
bytes: [u8; 32],
len: u8,
}
impl OidKey {
fn new(oid: &[u8]) -> Result<Self> {
let len = oid.len();
if len == 0 || len > 32 {
bail!("{len} is not the width of any git object id (1..=32 bytes)");
}
let mut bytes = [0u8; 32];
bytes[..len].copy_from_slice(oid);
Ok(OidKey {
bytes,
len: len as u8,
})
}
fn as_slice(&self) -> &[u8] {
&self.bytes[..self.len as usize]
}
}
impl<S: ObjectIndex + 'static> GitStore<S> {
pub fn open_with_arms(
root: &Path,
account: &str,
hash: GitHashKind,
arms: StoreConfig,
) -> Result<Self> {
std::fs::create_dir_all(root)
.with_context(|| format!("creating the store root {}", root.display()))?;
let blobs = root.join("objects.pack");
let writer = arms.writer.create(&blobs)?;
let journal = arms.writer.journal(&blobs);
let cache_bytes = arms.redb_cache_bytes;
let objects = ObjectReadStack::<S>::open(
&root.join("objects.tail"),
RebuildTriggers::default(),
cache_bytes,
)?;
let _ = cache_bytes; match std::fs::remove_file(root.join(ORDINAL_CACHE_FILE)) {
Ok(()) => tracing::info!(
target: "znippy.git",
path = %root.join(ORDINAL_CACHE_FILE).display(),
"removed a stale objects.ordinal — the persisted index's oid column is the \
ordinal space now"
),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
Err(e) => tracing::warn!(
target: "znippy.git",
path = %root.join(ORDINAL_CACHE_FILE).display(),
error = %e,
"could not remove a stale objects.ordinal; it is dead weight, not a fault"
),
}
let exploded = Arc::new(ExplodedCache::open(
&root.join("objects.exploded"),
crate::exploded_arrow::ExplodePolicy::from_env()?,
crate::exploded_cache::budget_from_env(root)?,
)?)
.registered();
let absorber = Arc::new(Absorber::<S>::open(
&blobs,
writer.blobs_shared(),
hash,
objects,
exploded,
)?);
let acked = match journal.as_deref() {
Some(p) if p.exists() => crate::archive_write::read_journal(p).with_context(|| {
format!(
"reading {} — it is the durable half of the indexed bit and a store \
cannot be opened without knowing which packs it owes index work for",
p.display()
)
})?,
_ => Vec::new(),
};
let requeue = absorber.adopt_journal(&acked)?;
let push = PushPath::with_absorber(
writer,
&blobs,
journal,
absorber.clone() as Arc<dyn ObjectAbsorb>,
)?;
let refs = RefLog::new(root.join("refs.log"));
let archive = root.join("repository.znippy");
let store = Self {
root: root.to_path_buf(),
blobs,
archive,
push,
account: account.to_string(),
hash,
absorber,
refs,
ref_gate: Mutex::new(()),
gc: arms.gc.create(),
arms,
};
let indexer = store.push.indexer(&store.account);
for p in &requeue {
indexer.submit(IndexJob {
pack_id: p.pack_id,
offset: p.offset,
len: p.len,
})?;
}
store.refold()?;
Ok(store)
}
pub fn arms(&self) -> StoreConfig {
self.arms
}
pub fn writer_name(&self) -> &'static str {
self.push.name()
}
pub fn writer_durability(&self) -> &'static str {
self.push.durability()
}
pub(crate) fn gc_arm(&self) -> &dyn Gc {
self.gc.as_ref()
}
pub fn blobs_path(&self) -> &Path {
&self.blobs
}
pub fn archive_path(&self) -> &Path {
&self.archive
}
pub fn root(&self) -> &Path {
&self.root
}
pub fn hash_kind(&self) -> GitHashKind {
self.hash
}
pub fn index(&self) -> &ObjectReadStack<S> {
&self.absorber.objects
}
pub fn object_count(&self) -> usize {
self.absorber.objects.len()
}
pub fn commit_count(&self) -> usize {
self.absorber
.derived
.read()
.expect("derived lock")
.graph
.len()
}
pub fn graph_source_stats(&self) -> (u64, u64) {
self.absorber.graph_source_stats()
}
pub fn reach_source_stats(&self) -> (u64, u64) {
self.absorber.reach_source_stats()
}
pub fn ordinal_source_stats(&self) -> (u64, u64) {
self.absorber.ordinal_source_stats()
}
#[cfg(test)]
pub(crate) fn ordinal_oids(&self) -> Vec<String> {
let oid_len = self.hash.oid_len();
self.absorber
.derived
.read()
.expect("derived lock")
.ordinals
.view()
.concat()
.chunks_exact(oid_len)
.map(hex::encode)
.collect()
}
pub fn content(&self, oid: Oid<'_>) -> Result<Option<(GitObjectKind, Vec<u8>)>> {
self.absorber.resolved(oid)
}
pub fn content_costed(
&self,
oid: Oid<'_>,
) -> Result<Option<(GitObjectKind, Vec<u8>, Option<crate::resolve::ReadCost>)>> {
self.absorber.resolved_costed(oid)
}
pub fn exploded_stats(&self) -> ExplodedStats {
self.absorber.exploded.engine_stats()
}
pub fn explode_cache_use(&self) -> crate::exploded_cache::CacheUse {
self.absorber.exploded.use_now()
}
pub fn exploded_eager_rows(&self) -> Result<u64> {
self.absorber.exploded.eager_rows()
}
pub fn set_explode_budget(&self, bytes: u64) {
self.absorber.exploded.set_budget(bytes);
}
pub fn explode_cache_maintain(&self) -> Result<u64> {
self.absorber.exploded.maintain()
}
pub fn drop_explode_cache(&self) -> Result<u64> {
self.absorber.exploded.drop_cache()
}
pub fn exploded_path(&self) -> &Path {
self.absorber.exploded.path()
}
pub fn exploded_of_kind(&self, kind: GitObjectKind) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
self.absorber.exploded.of_kind(kind)
}
pub fn graph_snapshot(&self) -> Vec<CommitNode> {
self.absorber
.derived
.read()
.expect("derived lock")
.graph
.clone()
}
pub fn unindexed_packs(&self) -> usize {
self.absorber.unindexed_packs()
}
pub fn indexer(&self) -> std::sync::Arc<crate::indexer::AccountIndexer> {
self.push.indexer(&self.account)
}
pub fn wait_indexed(&self) {
self.push.pool().wait_caught_up();
}
pub fn cancel_index(&self) {
self.absorber.cancel.store(true, std::sync::atomic::Ordering::SeqCst);
self.push.pool().cancel_all();
}
pub fn set_mem_watchdog(&self, dog: Arc<crate::memguard::MemWatchdog>) {
*self.absorber.watchdog.write().unwrap_or_else(|e| e.into_inner()) = dog;
}
pub fn index_cancelled(&self) -> bool {
self.absorber.cancel.load(std::sync::atomic::Ordering::SeqCst)
}
pub fn index_progress(&self) -> crate::progress::Snapshot {
self.absorber.progress.snapshot()
}
pub fn absorb_pending(&self) -> Result<usize> {
self.absorber.absorb_pending()
}
pub fn index_now(&self) -> Result<usize> {
self.absorber.index_pending_now()
}
pub fn unindexable_packs(&self) -> Vec<Unindexable> {
self.absorber.unindexable_packs()
}
pub fn oids_in_extent(&self, extent: Extent) -> Result<Vec<Vec<u8>>> {
let (start, len) = extent;
if self.try_absorb_pending()?.is_none() {
let snap = self.archive_snapshot()?;
let bytes = self.extent(&snap, start, len)?;
let walked = crate::pack_walk::walk(&bytes, self.hash.oid_len())?;
let mut oids: Vec<Vec<u8>> = Vec::with_capacity(walked.entries.len());
let limits = crate::resolve::ResolveLimits {
base_budget: (crate::resolve::base_budget_from_env() / 4).max(1),
..crate::resolve::ResolveLimits::none()
};
crate::resolve::resolve_streaming(
&bytes,
&walked,
self.hash,
start,
&*self.absorber,
&crate::exploded::NoSink,
limits,
&mut |rows| {
oids.extend(rows.drain(..).map(|r| r.oid));
Ok(())
},
)
.context("resolving the pushed pack read-only to enumerate its objects")?;
return Ok(oids);
}
let end = start.saturating_add(len);
let all = self.index().oids_in_order()?;
let refs: Vec<&[u8]> = all.iter().map(Vec::as_slice).collect();
let extents = self.index().extents_batch(&refs);
Ok(all
.into_iter()
.zip(extents)
.filter_map(|(oid, at)| match at {
Some((offset, _)) if offset >= start && offset < end => Some(oid),
_ => None,
})
.collect())
}
pub fn emit_set(
&self,
oids: &[Oid<'_>],
ofs_delta_ok: bool,
thin_haves: Option<&[Oid<'_>]>,
) -> Result<Vec<crate::pack_walk::EmitEntry>> {
use crate::index_layout::ObjType;
use std::collections::hash_map::Entry;
let _ = self
.try_absorb_pending()
.context("absorbing pending index work before emitting a pack")?;
let mut asked: Vec<&[u8]> = Vec::with_capacity(oids.len());
let mut seen: HashMap<&[u8], usize> = HashMap::with_capacity(oids.len());
for oid in oids {
if let Entry::Vacant(slot) = seen.entry(*oid) {
slot.insert(asked.len());
asked.push(*oid);
}
}
let rows = self.lookup_batch_never_wait(&asked)?;
let snap = self.archive_snapshot()?;
let in_set_offsets: HashMap<u64, usize> = rows
.iter()
.enumerate()
.filter_map(|(i, r)| r.as_ref().map(|r| (r.offset, i)))
.collect();
let oid_len = self.hash_kind().oid_len();
let mut base_content: Option<(u64, Vec<u8>)> = None;
let mut client_bases: Option<ExternalBases> = None;
let mut out = Vec::with_capacity(asked.len());
for (oid, row) in asked.iter().zip(rows.iter()) {
let Some(row) = row.as_ref() else {
bail!(
"this pack was asked for {}, which this repository does not hold; emitting \
the rest would be a short pack reported as a success, so this refuses \
instead",
hex::encode(oid)
);
};
let base = match row.obj_type {
ObjType::OfsDelta => match in_set_offsets.get(&row.delta_base) {
Some(_) => BaseOf::Inside {
at: row.delta_base,
},
None => BaseOf::Outside {
offset: row.delta_base,
oid: None,
},
},
ObjType::RefDelta => {
let head = self.extent(&snap, row.offset, row.len.min(64))?;
let (_, _, n) = crate::pack_walk::type_and_size_of(&head)?;
match head.get(n..n + oid_len) {
Some(b) => match seen
.get(b)
.and_then(|i| rows[*i].as_ref())
.map(|r| r.offset)
{
Some(at) => BaseOf::Inside { at },
None => BaseOf::Outside {
offset: 0,
oid: Some(b.to_vec()),
},
},
None => BaseOf::Outside {
offset: 0,
oid: None,
},
}
}
_ => BaseOf::Whole,
};
if let BaseOf::Whole | BaseOf::Inside { .. } = base {
let inside_at = match base {
BaseOf::Inside { at } => Some(at),
_ => None,
};
let extent = crate::pack_walk::EntryBytes::Extent {
offset: row.offset,
len: row.len,
};
let (stored, obj_type, delta_base) = match (row.obj_type, inside_at) {
(ObjType::OfsDelta, Some(_)) if !ofs_delta_ok => {
let base_oid = self.oid_at_offset(row.delta_base, &asked, &rows)?;
let bytes = self.extent(&snap, row.offset, row.len)?;
(
crate::pack_walk::EntryBytes::Owned(Self::as_ref_delta(
&bytes, &base_oid,
)?),
ObjType::RefDelta,
0,
)
}
(ObjType::RefDelta, Some(at)) if ofs_delta_ok => {
let bytes = self.extent(&snap, row.offset, row.len)?;
(
crate::pack_walk::EntryBytes::Owned(Self::as_ofs_delta(
&bytes, oid_len, row.offset, at,
)?),
ObjType::OfsDelta,
at,
)
}
_ => (extent, row.obj_type, row.delta_base),
};
out.push(crate::pack_walk::EmitEntry {
oid: oid.to_vec(),
stored,
obj_type,
uncompressed_size: row.uncompressed_size,
delta_base,
offset: row.offset,
recompressed: false,
deltified: false,
});
continue;
}
if let Some(haves) = thin_haves {
if client_bases.is_none() {
client_bases = Some(self.client_bases(haves)?);
}
let ext = client_bases
.as_ref()
.expect("the client's bases were just built");
let base_oid: Option<Vec<u8>> = match &base {
BaseOf::Whole | BaseOf::Inside { .. } => None,
BaseOf::Outside {
oid: Some(named), ..
} => ext
.held
.contains(&OidKey::new(named)?)
.then(|| named.clone()),
BaseOf::Outside {
offset,
oid: None,
} => ext.by_offset.get(offset).map(|k| k.as_slice().to_vec()),
};
if let Some(base_oid) = base_oid {
let stored = if row.obj_type == ObjType::OfsDelta {
let bytes = self.extent(&snap, row.offset, row.len)?;
crate::pack_walk::EntryBytes::Owned(Self::as_ref_delta(&bytes, &base_oid)?)
} else {
crate::pack_walk::EntryBytes::Extent {
offset: row.offset,
len: row.len,
}
};
out.push(crate::pack_walk::EmitEntry {
oid: oid.to_vec(),
stored,
obj_type: ObjType::RefDelta,
uncompressed_size: row.uncompressed_size,
delta_base: 0,
offset: row.offset,
recompressed: false,
deltified: false,
});
continue;
}
}
let (kind, body) = self.content(oid)?.ok_or_else(|| {
anyhow!(
"{} is a delta whose base is outside this request, so it must be sent whole, \
and this repository cannot produce its content — neither §14's exploded \
table nor a re-derivation from the verbatim packs answered",
hex::encode(oid)
)
})?;
let candidate = if crate::delta::enabled() {
self.in_request_chain_ancestor(row.delta_base, row.offset, &in_set_offsets)?
} else {
None
};
if let Some(base_at) = candidate {
let at = *in_set_offsets
.get(&base_at)
.expect("the ancestor was found by lookup in this very map");
if base_content.as_ref().is_none_or(|(o, _)| *o != base_at) {
let base_oid = asked[at];
let (_, base_body) = self.content(base_oid)?.ok_or_else(|| {
anyhow!(
"{} is in this request and is the delta base chosen for {}, and this \
repository cannot produce its content",
hex::encode(base_oid),
hex::encode(oid)
)
})?;
base_content = Some((base_at, base_body));
}
let (_, base_body) = base_content.as_ref().expect("just filled");
if let Some(d) = crate::delta::delta(base_body, &body)? {
let stored = Self::ofs_delta_entry(&d, row.offset, base_at)?;
let (stored, obj_type, delta_base) = if ofs_delta_ok {
(
crate::pack_walk::EntryBytes::Owned(stored),
ObjType::OfsDelta,
base_at,
)
} else {
(
crate::pack_walk::EntryBytes::Owned(Self::as_ref_delta(
&stored, asked[at],
)?),
ObjType::RefDelta,
0,
)
};
out.push(crate::pack_walk::EmitEntry {
oid: oid.to_vec(),
stored,
obj_type,
uncompressed_size: row.uncompressed_size,
delta_base,
offset: row.offset,
recompressed: true,
deltified: true,
});
continue;
}
}
out.push(crate::pack_walk::EmitEntry {
oid: oid.to_vec(),
stored: crate::pack_walk::EntryBytes::Owned(Self::whole_entry(kind, &body)?),
obj_type: crate::serve::resolved_type(kind),
uncompressed_size: body.len() as u64,
delta_base: 0,
offset: row.offset,
recompressed: true,
deltified: false,
});
}
Ok(out)
}
pub(crate) fn resolve_emit_payloads<'a>(
&self,
entries: &'a [crate::pack_walk::EmitEntry],
snap: &'a crate::archive_map::Mapped,
) -> Result<Vec<std::borrow::Cow<'a, [u8]>>>
where
Self: Sync,
{
let workers = if entries.len() < FAN_OUT_AT {
1
} else {
emit_workers()
};
self.resolve_with(entries, snap, workers)
}
pub(crate) fn resolve_with<'a>(
&self,
entries: &'a [crate::pack_walk::EmitEntry],
snap: &'a crate::archive_map::Mapped,
workers: usize,
) -> Result<Vec<std::borrow::Cow<'a, [u8]>>>
where
Self: Sync,
{
use znippy_zoomies::gatling_forkjoin::gatling_for_each;
let one = |i: usize| -> Result<std::borrow::Cow<'a, [u8]>> {
let e = &entries[i];
let bytes = match &e.stored {
crate::pack_walk::EntryBytes::Owned(v) => std::borrow::Cow::Borrowed(v.as_slice()),
crate::pack_walk::EntryBytes::Extent { offset, len } => {
self.extent(snap, *offset, *len)?
}
};
crate::pack_walk::type_and_size_of(&bytes).with_context(|| {
format!(
"resolving the stored bytes of {} for emission",
hex::encode(&e.oid)
)
})?;
Ok(bytes)
};
if workers <= 1 {
return (0..entries.len()).map(one).collect();
}
gatling_for_each(entries.len(), workers, one)
.into_iter()
.collect()
}
pub(crate) fn emit_ordered(
&self,
ordered: &[crate::pack_walk::EmitEntry],
out: &mut dyn std::io::Write,
) -> Result<crate::pack_walk::EmitReport>
where
Self: Sync,
{
let snap = self
.archive_snapshot()
.context("mapping the archive to emit")?;
let payloads = self
.resolve_emit_payloads(ordered, &snap)
.context("resolving the emit set's stored bytes")?;
crate::pack_walk::emit_pack(
ordered,
self.hash_kind(),
out,
&|i| Ok(payloads[i].as_ref()),
)
}
fn in_request_chain_ancestor(
&self,
from: u64,
entry_at: u64,
in_request: &HashMap<u64, usize>,
) -> Result<Option<u64>> {
use crate::index_layout::{ObjType, ObjectIndex as _};
const MAX_LINKS: usize = 100;
const HEADER_PROBE: u64 = 64;
let oid_len = self.hash_kind().oid_len();
let snap = self.archive_snapshot()?;
let mut at = from;
for _ in 0..MAX_LINKS {
if at != 0 && at < entry_at && in_request.contains_key(&at) {
return Ok(Some(at));
}
if at == 0 {
return Ok(None);
}
let head = self.header_probe(&snap, at, HEADER_PROBE)?;
let (t, _, n) = crate::pack_walk::type_and_size_of(&head)?;
at = match t {
ObjType::Commit | ObjType::Tree | ObjType::Blob | ObjType::Tag => {
return Ok(None)
}
ObjType::OfsDelta => {
let (distance, _) = crate::pack_walk::ofs_distance_of(&head[n..])?;
match at.checked_sub(distance) {
Some(next) if next != 0 => next,
_ => return Ok(None),
}
}
ObjType::RefDelta => {
let Some(base_oid) = head.get(n..n + oid_len) else {
return Ok(None);
};
match self.index().lookup(base_oid) {
Some(row) => row.offset,
None => return Ok(None),
}
}
};
}
Ok(None)
}
fn ofs_delta_entry(d: &[u8], entry_at: u64, base_at: u64) -> Result<Vec<u8>> {
use crate::index_layout::ObjType;
use std::io::Write as _;
let distance = entry_at.checked_sub(base_at).ok_or_else(|| {
anyhow!(
"a computed delta at archive offset {entry_at} names a base at {base_at}, which \
is after it — an ofs-delta distance is backwards and this would not encode"
)
})?;
let mut out = Vec::with_capacity(d.len() / 2 + 32);
crate::pack_walk::encode_type_and_size(&mut out, ObjType::OfsDelta, d.len() as u64);
crate::pack_walk::encode_ofs_distance(&mut out, distance);
let mut z = flate2::write::ZlibEncoder::new(out, flate2::Compression::fast());
z.write_all(d)?;
Ok(z.finish()?)
}
fn client_bases(&self, haves: &[Oid<'_>]) -> Result<ExternalBases> {
use crate::index_layout::ObjectIndex as _;
let held = self.reachable_raw(haves, &[]).context(
"closing over the negotiated common tips, to learn which bases the client can supply \
for itself",
)?;
let refs: Vec<&[u8]> = held.iter().collect();
let rows = self.index().lookup_batch(&refs);
let mut by_offset = HashMap::with_capacity(refs.len());
let mut keys = std::collections::HashSet::with_capacity(refs.len());
for (oid, row) in refs.iter().zip(rows.iter()) {
let key = OidKey::new(oid)?;
if let Some(row) = row {
by_offset.insert(row.offset, key);
}
keys.insert(key);
}
Ok(ExternalBases {
by_offset,
held: keys,
})
}
fn oid_at_offset(
&self,
offset: u64,
asked: &[&[u8]],
rows: &[Option<crate::index_layout::IndexRow>],
) -> Result<Vec<u8>> {
asked
.iter()
.zip(rows)
.find(|(_, r)| r.as_ref().is_some_and(|r| r.offset == offset))
.map(|(oid, _)| oid.to_vec())
.ok_or_else(|| {
anyhow!(
"the entry at archive offset {offset} was checked to be in this request and \
then could not be found in it"
)
})
}
fn as_ref_delta(stored: &[u8], base_oid: &[u8]) -> Result<Vec<u8>> {
use crate::index_layout::ObjType;
let (t, stated_size, n) = crate::pack_walk::type_and_size_of(stored)?;
if t != ObjType::OfsDelta {
bail!("only an ofs-delta can be re-headed as a ref-delta, this entry is {t:?}");
}
let (_, d) = crate::pack_walk::ofs_distance_of(stored.get(n..).unwrap_or(&[]))?;
let mut out = Vec::with_capacity(stored.len() + base_oid.len());
crate::pack_walk::encode_type_and_size(&mut out, ObjType::RefDelta, stated_size);
out.extend_from_slice(base_oid);
out.extend_from_slice(stored.get(n + d..).unwrap_or(&[]));
Ok(out)
}
fn as_ofs_delta(
stored: &[u8],
oid_len: usize,
entry_at: u64,
base_at: u64,
) -> Result<Vec<u8>> {
use crate::index_layout::ObjType;
let (t, stated_size, n) = crate::pack_walk::type_and_size_of(stored)?;
if t != ObjType::RefDelta {
bail!("only a ref-delta can be re-headed as an ofs-delta, this entry is {t:?}");
}
let payload = stored.get(n + oid_len..).ok_or_else(|| {
anyhow!(
"a ref-delta entry at archive offset {entry_at} is {} bytes, which is not even its \
header plus a {oid_len}-byte base oid",
stored.len()
)
})?;
let mut out = Vec::with_capacity(payload.len() + 32);
crate::pack_walk::encode_type_and_size(&mut out, ObjType::OfsDelta, stated_size);
crate::pack_walk::encode_ofs_distance(
&mut out,
entry_at.checked_sub(base_at).filter(|d| *d > 0).unwrap_or(1),
);
out.extend_from_slice(payload);
Ok(out)
}
fn whole_entry(kind: crate::object::GitObjectKind, body: &[u8]) -> Result<Vec<u8>> {
use std::io::Write as _;
let mut out = Vec::with_capacity(body.len() / 2 + 32);
crate::pack_walk::encode_type_and_size(
&mut out,
crate::serve::resolved_type(kind),
body.len() as u64,
);
let mut z = flate2::write::ZlibEncoder::new(out, flate2::Compression::fast());
z.write_all(body)?;
Ok(z.finish()?)
}
#[cfg(test)]
pub(crate) fn hold_absorb_gate(&self) -> std::sync::MutexGuard<'_, ()> {
self.absorber.gate.lock().expect("absorb gate poisoned")
}
pub(crate) fn read_extent(&self, offset: u64, len: u64) -> Result<Vec<u8>> {
self.absorber.read_extent(offset, len)
}
pub(crate) fn archive_snapshot(&self) -> Result<Arc<crate::archive_map::Mapped>> {
self.absorber.map.snapshot()
}
pub(crate) fn extent<'m>(
&self,
snap: &'m crate::archive_map::Mapped,
offset: u64,
len: u64,
) -> Result<std::borrow::Cow<'m, [u8]>> {
match snap.get(offset, len) {
Some(b) => Ok(std::borrow::Cow::Borrowed(b)),
None => Ok(std::borrow::Cow::Owned(self.read_extent(offset, len)?)),
}
}
pub(crate) fn header_probe<'m>(
&self,
snap: &'m crate::archive_map::Mapped,
offset: u64,
len: u64,
) -> Result<std::borrow::Cow<'m, [u8]>> {
clamped_window(&self.absorber, snap, offset, len)
}
pub(crate) fn queue(&self, pack_id: u64, extent: Extent) -> Result<()> {
self.absorber.queue(pack_id, extent)
}
fn reach_bitmaps(&self) -> Result<Arc<crate::reach::ReachTable>> {
self.absorber.reach_bitmaps()
}
#[cfg(test)]
pub(crate) fn reach_bitmaps_with(
&self,
policy: crate::reach::ReachPolicy,
cache: bool,
) -> Result<Arc<crate::reach::ReachTable>> {
self.absorber.reach_bitmaps_with(policy, cache)
}
pub fn reach_stats(&self) -> (u64, u64, u64) {
self.absorber.reach_stats()
}
pub(crate) fn refold(&self) -> Result<()> {
self.absorber.refold()
}
}
impl<S: ObjectIndex> Absorber<S> {
fn open(
blobs: &Path,
shared: Option<Arc<File>>,
hash: GitHashKind,
objects: ObjectReadStack<S>,
exploded: Arc<ExplodedCache>,
) -> Result<Self> {
let reader = match shared {
Some(f) => f,
None => Arc::new(
File::open(blobs)
.with_context(|| format!("opening {} for reads", blobs.display()))?,
),
};
let label = blobs
.parent()
.map(|p| p.display().to_string())
.unwrap_or_else(|| blobs.display().to_string());
let graph_cache = crate::arms::graph_cache_from_env()?;
if !graph_cache {
let stale = blobs.with_file_name(GRAPH_CACHE_FILE);
match std::fs::remove_file(&stale) {
Ok(()) => {}
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
Err(e) => tracing::warn!(
target: "znippy.git",
path = %stale.display(),
error = %e,
"the graph cache is off but a stale objects.graph could not be removed"
),
}
}
Ok(Self {
blobs: blobs.to_path_buf(),
reader,
map: crate::archive_map::ArchiveMap::new(blobs),
hash,
objects,
exploded,
derived: RwLock::new(Derived::default()),
graph_read: std::sync::atomic::AtomicU64::new(0),
graph_recomputed: std::sync::atomic::AtomicU64::new(0),
graph_cache: std::sync::atomic::AtomicBool::new(graph_cache),
reach_carried: std::sync::atomic::AtomicU64::new(0),
reach_rebuilt: std::sync::atomic::AtomicU64::new(0),
reach_loaded: std::sync::atomic::AtomicU64::new(0),
reach_extended: std::sync::atomic::AtomicU64::new(0),
reach_unpersisted: std::sync::atomic::AtomicU64::new(0),
ordinal_folded: std::sync::atomic::AtomicU64::new(0),
ordinal_tail: std::sync::atomic::AtomicU64::new(0),
packs: Mutex::new(PackState::default()),
gate: Mutex::new(()),
progress: crate::progress::IndexProgress::new(&label),
cancel: std::sync::atomic::AtomicBool::new(false),
label,
watchdog: RwLock::new(crate::memguard::global()),
scanned: Mutex::new(HashMap::new()),
scan_build: Mutex::new(()),
unindexable: Mutex::new(HashMap::new()),
})
}
fn tree_source(&self) -> TreesFromStore<'_, S> {
TreesFromStore { abs: self, chains: None }
}
fn tree_source_for_build(&self, objects: usize) -> TreesFromStore<'_, S> {
let cap = objects
.saturating_mul(REACH_BUILD_CHAIN_CACHE_PER_OBJECT)
.clamp(8 << 20, REACH_BUILD_CHAIN_CACHE_BYTES);
TreesFromStore {
abs: self,
chains: Some(Mutex::new(crate::resolve::ChainCache::new(cap))),
}
}
fn unindexed_packs(&self) -> usize {
self.packs.lock().expect("pack state lock").unabsorbed
}
fn index_pending_now(&self) -> Result<usize> {
if self.unindexed_packs() == 0 {
return Ok(0);
}
if let Some(current) = IN_ABSORB.with(|c| c.get()) {
let jobs: Vec<PendingPack> = {
let s = self.packs.lock().map_err(|_| anyhow!("pack state poisoned"))?;
s.pending()
};
return self.absorb_each(&jobs, Some(current));
}
let _gate = self.gate.lock().map_err(|_| anyhow!("absorb gate poisoned"))?;
let jobs: Vec<PendingPack> = {
let s = self.packs.lock().map_err(|_| anyhow!("pack state poisoned"))?;
s.pending()
};
self.absorb_each(&jobs, None)
}
fn try_absorb_pending(&self) -> Result<Option<usize>> {
if self.unindexed_packs() == 0 {
return Ok(Some(0));
}
let _gate = match self.gate.try_lock() {
Ok(g) => g,
Err(std::sync::TryLockError::WouldBlock) => return Ok(None),
Err(std::sync::TryLockError::Poisoned(_)) => bail!("absorb gate poisoned"),
};
let jobs: Vec<PendingPack> = {
let s = self
.packs
.lock()
.map_err(|_| anyhow!("pack state poisoned"))?;
s.pending()
};
self.absorb_each(&jobs, None).map(Some)
}
fn scan_pending(
&self,
wants: &[&[u8]],
capture: Option<&[u8]>,
) -> Result<(Vec<Option<crate::index_layout::IndexRow>>, Option<(GitObjectKind, Vec<u8>)>)> {
use crate::index_layout::IndexRow;
let mut out: Vec<Option<IndexRow>> = vec![None; wants.len()];
let mut missing: HashMap<&[u8], Vec<usize>> = HashMap::with_capacity(wants.len());
for (i, w) in wants.iter().enumerate() {
missing.entry(*w).or_default().push(i);
}
if missing.is_empty() && capture.is_none() {
return Ok((out, None));
}
thread_local! { static SCANNING: std::cell::Cell<bool> = const { std::cell::Cell::new(false) }; }
let nested = SCANNING.with(|s| s.get());
struct Unset(bool);
impl Drop for Unset {
fn drop(&mut self) {
if !self.0 {
SCANNING.with(|s| s.set(false));
}
}
}
SCANNING.with(|s| s.set(true));
let _unset = Unset(nested);
let pending: Vec<PendingPack> = {
let s = self
.packs
.lock()
.map_err(|_| anyhow!("pack state poisoned"))?;
s.pending()
};
if pending.is_empty() {
return Ok((out, None));
}
let snap = self.map.snapshot()?;
let mut content: Option<(GitObjectKind, Vec<u8>)> = None;
for job in &pending {
if missing.is_empty() && (capture.is_none() || content.is_some()) {
break;
}
let scanned = match self.cached_scan(job.offset) {
Some(s) => s,
None if nested => continue,
None => self.scanned_pack(&snap, job)?,
};
missing.retain(|oid, slots| match scanned.row(oid) {
Some(r) => {
for &i in slots.iter() {
out[i] = Some(r.clone());
}
false
}
None => true,
});
if let Some(want) = capture
&& content.is_none()
&& let Some((entry, _)) = scanned.find(want)
{
let bytes: std::borrow::Cow<'_, [u8]> = match snap.get(job.offset, job.len) {
Some(b) => std::borrow::Cow::Borrowed(b),
None => std::borrow::Cow::Owned(self.read_extent(job.offset, job.len)?),
};
let sp = scanned.clone();
let offset_of = move |oid: &[u8]| sp.find(oid).map(|(e, _)| sp.walk.entries[e].offset);
content = Some(crate::resolve::materialize_entry(
&bytes,
&scanned.walk,
entry,
self.hash.oid_len(),
&offset_of,
self,
)?);
}
}
Ok((out, content))
}
fn cached_scan(&self, offset: u64) -> Option<Arc<ScannedPack>> {
self.scanned
.lock()
.unwrap_or_else(|e| e.into_inner())
.get(&offset)
.cloned()
}
fn scanned_pack(&self, snap: &crate::archive_map::Mapped, job: &PendingPack) -> Result<Arc<ScannedPack>> {
if let Some(s) = self.scanned.lock().unwrap_or_else(|e| e.into_inner()).get(&job.offset) {
return Ok(s.clone());
}
let _one_at_a_time = self.scan_build.lock().unwrap_or_else(|e| e.into_inner());
if let Some(s) = self.scanned.lock().unwrap_or_else(|e| e.into_inner()).get(&job.offset) {
return Ok(s.clone());
}
let t0 = std::time::Instant::now();
let bytes: std::borrow::Cow<'_, [u8]> = match snap.get(job.offset, job.len) {
Some(b) => std::borrow::Cow::Borrowed(b),
None => std::borrow::Cow::Owned(self.read_extent(job.offset, job.len)?),
};
let walk = crate::pack_walk::walk(&bytes, self.hash.oid_len())?;
let mut rows: Vec<(Vec<u8>, u32, u64)> = Vec::with_capacity(walk.entries.len());
let limits = crate::resolve::ResolveLimits {
base_budget: (crate::resolve::base_budget_from_env() / 4).max(1),
cancel: Some(&self.cancel),
label: &self.label,
..crate::resolve::ResolveLimits::none()
};
let report = crate::resolve::resolve_streaming(
&bytes,
&walk,
self.hash,
job.offset,
self,
&crate::exploded::NoSink,
limits,
&mut |batch| {
for r in batch.drain(..) {
let pack_off = r.offset - job.offset;
let idx = walk
.entries
.binary_search_by_key(&pack_off, |e| e.offset)
.map_err(|_| anyhow!("resolved row at {pack_off} has no walk entry"))?;
rows.push((r.oid, idx as u32, r.uncompressed_size));
}
Ok(())
},
)
.context("resolving a pending pack read-only for lookups")?;
rows.sort_unstable_by(|a, b| a.0.cmp(&b.0));
let scanned = Arc::new(ScannedPack {
walk,
rows,
archive_offset: job.offset,
});
tracing::info!(
target: "gunnar::index",
event = "index.scanned",
repo = %self.label,
pack_id = job.pack_id,
objects = scanned.rows.len(),
reinflations = report.reinflations,
elapsed_ms = t0.elapsed().as_millis() as u64,
"a reader scanned a pending pack for itself: {} objects in {} ms",
scanned.rows.len(),
t0.elapsed().as_millis()
);
self.scanned
.lock()
.unwrap_or_else(|e| e.into_inner())
.insert(job.offset, scanned.clone());
Ok(scanned)
}
fn spill(&self) -> u64 {
let dropped = {
let mut s = self.scanned.lock().unwrap_or_else(|e| e.into_inner());
let n: usize = s.values().map(|p| p.rows.len()).sum();
s.clear();
n
};
self.exploded.drop_caches();
tracing::info!(
target: "gunnar::index",
event = "index.spilled",
repo = %self.label,
scanned_rows_dropped = dropped,
"the store dropped its rebuildable caches for the memory watchdog"
);
dropped as u64
}
fn watchdog(&self) -> Arc<crate::memguard::MemWatchdog> {
self.watchdog.read().unwrap_or_else(|e| e.into_inner()).clone()
}
fn lookup_never_wait(&self, oid: &[u8]) -> Result<Option<crate::index_layout::IndexRow>> {
if let Some(r) = self.objects.lookup(oid) {
return Ok(Some(r));
}
if self.unindexed_packs() == 0 {
return Ok(self.objects.lookup(oid));
}
let (mut rows, _) = self.scan_pending(&[oid], None)?;
if let Some(r) = rows.pop().flatten() {
return Ok(Some(r));
}
Ok(self.objects.lookup(oid))
}
fn lookup_batch_never_wait(&self, oids: &[&[u8]]) -> Result<Vec<Option<crate::index_layout::IndexRow>>> {
let mut rows = self.objects.lookup_batch(oids);
if rows.iter().all(Option::is_some) {
return Ok(rows);
}
if self.unindexed_packs() == 0 {
return Ok(self.objects.lookup_batch(oids));
}
let missing: Vec<&[u8]> = oids
.iter()
.zip(&rows)
.filter(|(_, r)| r.is_none())
.map(|(o, _)| *o)
.collect();
let (found, _) = self.scan_pending(&missing, None)?;
let mut by_oid: HashMap<&[u8], crate::index_layout::IndexRow> = HashMap::with_capacity(found.len());
for (o, r) in missing.iter().zip(found) {
if let Some(r) = r {
by_oid.insert(*o, r);
}
}
for (o, slot) in oids.iter().zip(rows.iter_mut()) {
if slot.is_none()
&& let Some(r) = by_oid.get(*o)
{
*slot = Some(r.clone());
}
}
if rows.iter().any(Option::is_none) {
for (o, slot) in oids.iter().zip(rows.iter_mut()) {
if slot.is_none() {
*slot = self.objects.lookup(o);
}
}
}
Ok(rows)
}
fn queue(&self, pack_id: u64, extent: Extent) -> Result<()> {
let mut s = self
.packs
.lock()
.map_err(|_| anyhow!("pack state poisoned"))?;
s.note(pack_id, extent);
Ok(())
}
fn adopt_journal(&self, journal: &[Extent]) -> Result<Vec<PendingPack>> {
if journal.is_empty() {
return Ok(Vec::new());
}
let packs = crate::archive_write::acked_packs(journal);
let retired = crate::archive_write::retired_offsets(journal);
if packs.is_empty() {
return Ok(Vec::new());
}
let exploded_is_whole = match self.exploded.policy() {
crate::exploded_arrow::ExplodePolicy::Full => {
self.exploded.eager_rows()? >= self.objects.len() as u64
}
crate::exploded_arrow::ExplodePolicy::Off => true,
_ => self.objects.len() == 0 || self.exploded.eager_rows()? > 0,
};
let has_rows = if exploded_is_whole {
self.objects.extents_with_rows(&packs)?
} else {
vec![false; packs.len()]
};
{
let mut d = self
.derived
.write()
.map_err(|_| anyhow!("derived poisoned"))?;
d.packs = packs.clone();
}
let mut s = self
.packs
.lock()
.map_err(|_| anyhow!("pack state poisoned"))?;
let mut requeue = Vec::new();
for (i, (&extent, &absorbed)) in packs.iter().zip(&has_rows).enumerate() {
let pack_id = i as u64;
if absorbed {
s.mark_absorbed(pack_id, extent);
} else if retired.contains(&extent.0) {
s.mark_absorbed(pack_id, extent);
} else {
s.note(pack_id, extent);
requeue.push(PendingPack {
pack_id,
offset: extent.0,
len: extent.1,
resumed: true,
});
}
}
Ok(requeue)
}
fn absorb_pending(&self) -> Result<usize> {
if self.unindexed_packs() == 0 {
return Ok(0);
}
let _gate = self
.gate
.lock()
.map_err(|_| anyhow!("absorb gate poisoned"))?;
let jobs: Vec<PendingPack> = {
let s = self
.packs
.lock()
.map_err(|_| anyhow!("pack state poisoned"))?;
s.pending()
};
self.absorb_each(&jobs, None)
}
fn absorb_each(&self, jobs: &[PendingPack], skip: Option<u64>) -> Result<usize> {
let mut absorbed = 0usize;
let mut first: Option<anyhow::Error> = None;
for job in jobs {
if skip == Some(job.pack_id) || self.given_up_on(job.pack_id) {
continue;
}
match self.absorb_gated(job) {
Ok(()) => absorbed += 1,
Err(e) => {
if self.cancel.load(std::sync::atomic::Ordering::Relaxed) {
return Err(e);
}
self.note_absorb_failure(job, &e);
if first.is_none() {
first = Some(e);
}
}
}
}
match first {
None => Ok(absorbed),
Some(e) => Err(e),
}
}
fn unindexable_packs(&self) -> Vec<Unindexable> {
let mut v: Vec<Unindexable> = self
.unindexable
.lock()
.unwrap_or_else(|e| e.into_inner())
.values()
.cloned()
.collect();
v.sort_by_key(|u| u.pack_id);
v
}
fn given_up_on(&self, pack_id: u64) -> bool {
self.unindexable
.lock()
.unwrap_or_else(|e| e.into_inner())
.contains_key(&pack_id)
}
fn note_absorb_failure(&self, job: &PendingPack, err: &anyhow::Error) {
let giving_up;
{
let mut map = self.unindexable.lock().unwrap_or_else(|e| e.into_inner());
let e = map.entry(job.pack_id).or_insert_with(|| Unindexable {
pack_id: job.pack_id,
extent: (job.offset, job.len),
attempts: 0,
reason: String::new(),
});
e.attempts += 1;
e.reason = format!("{}", err.root_cause());
if e.attempts < MAX_ABSORB_ATTEMPTS {
let attempts = e.attempts;
map.remove(&job.pack_id);
tracing::warn!(
target: "znippy.git",
store = %self.label,
pack = job.pack_id,
attempts,
max = MAX_ABSORB_ATTEMPTS,
error = %err.root_cause(),
"a pack could not be indexed; it stays pending and will be tried again"
);
return;
}
giving_up = Some(e.clone());
}
if let Some(u) = giving_up {
tracing::error!(
target: "znippy.git",
store = %self.label,
event = "index.unindexable",
pack = u.pack_id,
offset = u.extent.0,
len = u.extent.1,
attempts = u.attempts,
error = %u.reason,
"this store has STOPPED trying to index this pack after {} attempts. Its bytes \
are durable and untouched, and every object it introduced is absent from the \
index — a push that deltas against one is refused by name and a clone that \
needs one fails its connectivity check, both immediately. Nothing will retry \
it: re-push the pack, or delete and re-create the repository.",
u.attempts
);
}
}
fn absorb_gated(&self, job: &PendingPack) -> Result<()> {
if self.cancel.load(std::sync::atomic::Ordering::Relaxed) {
return Err(crate::resolve::ResolveError::cancelled());
}
{
let s = self
.packs
.lock()
.map_err(|_| anyhow!("pack state poisoned"))?;
if s.is_absorbed(job.pack_id) {
return Ok(());
}
}
let _in = InAbsorb::enter(job.pack_id);
self.absorb_one(job).map_err(|e| {
e.context(format!(
"absorbing pack {} at ({}, {}) — its bytes are durable and it stays \
un-indexed; reads will keep falling back rather than answer absent",
job.pack_id, job.offset, job.len
))
})?;
let mut s = self
.packs
.lock()
.map_err(|_| anyhow!("pack state poisoned"))?;
s.mark_absorbed(job.pack_id, (job.offset, job.len));
drop(s);
self.scanned.lock().unwrap_or_else(|e| e.into_inner()).remove(&job.offset);
Ok(())
}
fn absorb_one(&self, job: &PendingPack) -> Result<()> {
let t0 = std::time::Instant::now();
self.progress.begin(job.pack_id, job.len, job.resumed);
let _tick = crate::progress::ProgressTicker::start(self.progress.clone());
let result = self.absorb_one_inner(job, t0);
self.progress.sample_memory();
self.progress.finish();
result
}
fn absorb_one_inner(&self, job: &PendingPack, t0: std::time::Instant) -> Result<()> {
use crate::progress::Phase;
let progress = &self.progress;
let snap = self.map.snapshot()?;
let bytes: std::borrow::Cow<'_, [u8]> = match snap.get(job.offset, job.len) {
Some(b) => std::borrow::Cow::Borrowed(b),
None => std::borrow::Cow::Owned(self.read_extent(job.offset, job.len)?),
};
progress.set_phase(Phase::Walk);
let walked = crate::pack_walk::walk(&bytes, self.hash.oid_len())?;
let t_walk = t0.elapsed();
let dog = self.watchdog();
let spill = || self.spill();
let limits = crate::resolve::ResolveLimits::for_indexer(
&self.label,
&self.cancel,
progress,
dog,
&spill,
);
let mut batches = 0u64;
let mut appended = 0u64;
let mut entries: Vec<crate::index_layout::IndexEntry> = Vec::with_capacity(limits.row_batch);
let report = crate::resolve::resolve_streaming(
&bytes,
&walked,
self.hash,
job.offset,
self,
&*self.exploded,
limits,
&mut |rows| {
entries.clear();
entries.extend(rows.iter().map(|r| r.index_entry()));
rows.clear();
self.objects.append(&entries)?;
appended += entries.len() as u64;
batches += 1;
Ok(())
},
)?;
drop(entries);
let t_resolve = t0.elapsed() - t_walk;
progress.set_phase(Phase::Flush);
self.exploded.flush()?;
self.exploded.kick();
{
let mut d = self
.derived
.write()
.map_err(|_| anyhow!("derived poisoned"))?;
if !d.packs.contains(&(job.offset, job.len)) {
d.packs.push((job.offset, job.len));
}
}
progress.set_phase(Phase::Fold);
let t_fold0 = std::time::Instant::now();
self.refold()?;
let _ = self.objects.maybe_rebuild()?;
let t_fold = t_fold0.elapsed();
tracing::info!(
target: "gunnar::index",
event = "index.finished",
repo = %self.label,
pack_id = job.pack_id,
pack_bytes = job.len,
peak_rss = crate::memguard::peak_rss_bytes(),
mem_available = self.watchdog().available().unwrap_or(0),
objects = report.rows,
batches = batches,
rows_appended = appended,
reinflations = report.reinflations,
evictions = report.evictions,
watchdog_drops = report.watchdog_drops,
peak_base_bytes = report.peak_base_bytes,
pauses = report.pauses,
paused_ms = report.paused_ms,
walk_ms = t_walk.as_millis() as u64,
resolve_ms = t_resolve.as_millis() as u64,
fold_ms = t_fold.as_millis() as u64,
total_ms = t0.elapsed().as_millis() as u64,
resumed = job.resumed,
"indexed {} objects in {} batches: walk {} ms, resolve {} ms, fold {} ms; peak bases {}, {} evictions, {} re-inflations, {} pauses ({} ms); peak rss {}",
report.rows,
batches,
t_walk.as_millis(),
t_resolve.as_millis(),
t_fold.as_millis(),
crate::memguard::fmt_bytes(report.peak_base_bytes),
report.evictions,
report.reinflations,
report.pauses,
report.paused_ms,
crate::memguard::fmt_bytes(crate::memguard::peak_rss_bytes())
);
Ok(())
}
fn refold(&self) -> Result<()> {
let space = self.objects.oids_raw()?;
self.ordinal_folded
.store(space.folded_rows, std::sync::atomic::Ordering::Relaxed);
self.ordinal_tail
.store(space.tail_rows, std::sync::atomic::Ordering::Relaxed);
let oids_raw = space.raw;
let token = graph_token(&oids_raw);
let oid_len = self.hash.oid_len();
let cached = if self.graph_cache.load(std::sync::atomic::Ordering::Relaxed) {
self.load_persisted_graph(&token)?
} else {
None
};
let mut d = self
.derived
.write()
.map_err(|_| anyhow!("derived poisoned"))?;
let mut space_loaded = false;
if d.ordinals.is_empty() && d.reach.is_empty() {
if let Some((ordinals, table)) = self.load_persisted_reach(oid_len)? {
d.ordinals = ordinals;
d.reach = Arc::new(table);
space_loaded = true;
}
}
let appended: Option<std::ops::Range<u32>> = if d.ordinals.is_empty() {
None
} else {
d.ordinals.follow(&oids_raw)
};
if appended.is_none() {
d.ordinals = crate::reach::OrdinalTable::from_sorted(oids_raw.clone(), oid_len);
d.reach = Arc::default();
space_loaded = false;
}
if space_loaded {
self.reach_loaded
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
}
let graph: Vec<CommitNode> = match cached {
Some(g) => {
self.graph_read.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
g
}
None => {
self.graph_recomputed
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let g = match &appended {
Some(r) if !d.graph.is_empty() => {
let mut nodes = std::mem::take(&mut d.graph);
let space = d.ordinals.view();
for o in r.clone() {
let Some(oid) = space.oid(o) else { continue };
if let Some((GitObjectKind::Commit, payload)) = self.exploded.content(oid)? {
let h = crate::object::parse_commit(&payload);
nodes.push(CommitNode {
oid: hex::encode(oid),
parents: h.parents,
tree: h.tree,
committer_time: h.committer_time,
generation: 0,
});
}
}
assign_generations(nodes)
}
_ => {
let commits = self.exploded.of_kind(GitObjectKind::Commit)?;
let g: Vec<CommitNode> = commits
.iter()
.map(|(oid, payload)| {
let h = crate::object::parse_commit(payload);
CommitNode {
oid: hex::encode(oid),
parents: h.parents,
tree: h.tree,
committer_time: h.committer_time,
generation: 0,
}
})
.collect();
assign_generations(g)
}
};
if !self.graph_cache.load(std::sync::atomic::Ordering::Relaxed) {
} else if let Err(e) = self.persist_graph(&token, &g) {
tracing::warn!(store = %self.label, "could not persist objects.graph: {e:#}");
}
g
}
};
let mut ord_graph = crate::reach::OrdGraph::from_nodes(&graph, &d.ordinals.view());
let mut reach_changed = false;
if !d.reach.is_empty() {
if let Some(r) = &appended {
let new_rows: Vec<u32> = (0..ord_graph.len() as u32)
.filter(|&row| {
let o = ord_graph.ords[row as usize];
o != crate::reach::NONE && o >= r.start
})
.collect();
if !new_rows.is_empty() {
let trees = self.tree_source();
let facts = crate::reach::ObjectFacts {
ordinal: d.ordinals.view(),
trees: &trees,
oid_len,
};
let started = std::time::Instant::now();
let extended = crate::reach::extend_reach_table(
&d.reach,
&ord_graph,
&facts,
&new_rows,
live_reach_policy(),
);
tracing::info!(
store = %self.label,
new_commits = new_rows.len(),
entries = extended.len(),
elapsed_ms = started.elapsed().as_millis() as u64,
"fold: extended the reachability table for the pushed commits"
);
d.reach = Arc::new(extended);
self.reach_extended
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
reach_changed = true;
}
}
self.reach_carried
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
}
let mut compacted = false;
if d.ordinals.extra_len() > d.ordinals.base_len() / ORDINAL_COMPACT_FRACTION
&& d.ordinals.extra_len() > ORDINAL_COMPACT_MIN
{
let fresh = crate::reach::OrdinalTable::from_sorted(oids_raw, oid_len);
let remap = d.ordinals.remap_to(&fresh.view());
let kept = d.reach.remapped(&remap).unwrap_or_default();
tracing::info!(
store = %self.label,
appended = d.ordinals.extra_len(),
base = d.ordinals.base_len(),
entries = kept.len(),
"fold: compacted the ordinal space (re-sorted, table renumbered once)"
);
d.ordinals = fresh;
d.reach = Arc::new(kept);
ord_graph = crate::reach::OrdGraph::from_nodes(&graph, &d.ordinals.view());
compacted = true;
reach_changed = true;
}
d.graph = graph;
d.reach_by_row = Arc::new(d.reach.by_row(&ord_graph));
d.ord_graph = Arc::new(ord_graph);
d.token = token;
tracing::info!(
store = %self.label,
graph_len = d.graph.len(),
space_len = d.ordinals.len(),
appended = appended.as_ref().map(|r| r.len()).unwrap_or(0),
space_rebuilt = appended.is_none(),
reach_entries = d.reach.len(),
reach_loaded = space_loaded,
"fold: derived tables"
);
if reach_changed && !d.reach.is_empty() {
let due = self
.reach_unpersisted
.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
+ 1;
if compacted || due >= REACH_PERSIST_EVERY {
self.persist_reach_locked(&d);
}
}
Ok(())
}
fn persist_reach_locked(&self, d: &Derived) {
if !self.graph_cache.load(std::sync::atomic::Ordering::Relaxed) || d.reach.is_empty() {
return;
}
match self.persist_reach(&d.ordinals, &d.reach) {
Ok(()) => {
self.reach_unpersisted
.store(0, std::sync::atomic::Ordering::Relaxed);
}
Err(e) => tracing::warn!(store = %self.label, "could not persist objects.reach: {e:#}"),
}
}
fn reach_source_stats(&self) -> (u64, u64) {
(
self.reach_carried.load(std::sync::atomic::Ordering::Relaxed),
self.reach_rebuilt.load(std::sync::atomic::Ordering::Relaxed),
)
}
fn reach_stats(&self) -> (u64, u64, u64) {
(
self.reach_loaded.load(std::sync::atomic::Ordering::Relaxed),
self.reach_extended.load(std::sync::atomic::Ordering::Relaxed),
self.reach_unpersisted.load(std::sync::atomic::Ordering::Relaxed),
)
}
fn graph_cache_path(&self) -> PathBuf {
self.blobs.with_file_name(GRAPH_CACHE_FILE)
}
fn load_persisted_graph(&self, token: &[u8; 32]) -> Result<Option<Vec<CommitNode>>> {
let bytes = match std::fs::read(self.graph_cache_path()) {
Ok(b) => b,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(e) => {
return Err(e).with_context(|| {
format!("reading {}", self.graph_cache_path().display())
});
}
};
if bytes.len() < GRAPH_CACHE_HEADER || &bytes[..4] != GRAPH_CACHE_MAGIC || bytes[4] != 1 {
return Ok(None);
}
if &bytes[5..GRAPH_CACHE_HEADER] != token {
return Ok(None);
}
match crate::graph::decode_graph(&bytes[GRAPH_CACHE_HEADER..]) {
Ok(g) => Ok(Some(g)),
Err(_) => Ok(None),
}
}
fn reach_cache_path(&self) -> PathBuf {
self.blobs.with_file_name(REACH_CACHE_FILE)
}
fn load_persisted_reach(&self, oid_len: usize) -> Result<Option<(crate::reach::OrdinalTable, crate::reach::ReachTable)>> {
if !self.graph_cache.load(std::sync::atomic::Ordering::Relaxed) {
return Ok(None);
}
let bytes = match std::fs::read(self.reach_cache_path()) {
Ok(b) => b,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(e) => {
return Err(e).with_context(|| format!("reading {}", self.reach_cache_path().display()));
}
};
if bytes.len() < REACH_CACHE_HEADER || &bytes[..4] != REACH_CACHE_MAGIC || bytes[4] != 1 {
return Ok(None);
}
let w = u32::from_le_bytes(bytes[5..9].try_into().expect("4 bytes")) as usize;
let base_len = u64::from_le_bytes(bytes[9..17].try_into().expect("8 bytes")) as usize;
let extra_len = u64::from_le_bytes(bytes[17..25].try_into().expect("8 bytes")) as usize;
if w != oid_len {
return Ok(None);
}
let Some(ipc_at) = REACH_CACHE_HEADER.checked_add(base_len).and_then(|a| a.checked_add(extra_len)) else {
return Ok(None);
};
if ipc_at > bytes.len() {
return Ok(None);
}
let base = &bytes[REACH_CACHE_HEADER..REACH_CACHE_HEADER + base_len];
let extra = &bytes[REACH_CACHE_HEADER + base_len..ipc_at];
let Some(ordinals) = crate::reach::OrdinalTable::from_base_and_extra(base.to_vec(), extra.to_vec(), oid_len) else {
return Ok(None);
};
match crate::reach::decode_live_table(&bytes[ipc_at..], ordinals.len() as u32) {
Ok(table) if !table.is_empty() => Ok(Some((ordinals, table))),
Ok(_) => Ok(None),
Err(e) => {
tracing::warn!(store = %self.label, "objects.reach is unreadable, rebuilding: {e:#}");
Ok(None)
}
}
}
fn persist_reach(&self, ordinals: &crate::reach::OrdinalTable, table: &crate::reach::ReachTable) -> Result<()> {
use std::io::Write as _;
let ipc = crate::reach::encode_live_table(table)?;
let (base, extra) = (ordinals.base_raw(), ordinals.extra_raw());
let mut out = Vec::with_capacity(REACH_CACHE_HEADER + base.len() + extra.len() + ipc.len());
out.extend_from_slice(REACH_CACHE_MAGIC);
out.push(1);
out.extend_from_slice(&(ordinals.oid_len() as u32).to_le_bytes());
out.extend_from_slice(&(base.len() as u64).to_le_bytes());
out.extend_from_slice(&(extra.len() as u64).to_le_bytes());
out.extend_from_slice(base);
out.extend_from_slice(extra);
out.extend_from_slice(&ipc);
let path = self.reach_cache_path();
let tmp = path.with_extension("reach.tmp");
{
let mut f = File::create(&tmp).with_context(|| format!("creating {}", tmp.display()))?;
f.write_all(&out).with_context(|| format!("writing {}", tmp.display()))?;
f.sync_all().with_context(|| format!("fsync {}", tmp.display()))?;
}
std::fs::rename(&tmp, &path).with_context(|| format!("publishing {}", path.display()))?;
Ok(())
}
fn persist_graph(&self, token: &[u8; 32], nodes: &[CommitNode]) -> Result<()> {
use std::io::Write as _;
let ipc = crate::graph::encode_graph(nodes)?;
let mut out = Vec::with_capacity(GRAPH_CACHE_HEADER + ipc.len());
out.extend_from_slice(GRAPH_CACHE_MAGIC);
out.push(1);
out.extend_from_slice(token);
out.extend_from_slice(&ipc);
let path = self.graph_cache_path();
let tmp = path.with_extension("graph.tmp");
{
let mut f = File::create(&tmp)
.with_context(|| format!("creating {}", tmp.display()))?;
f.write_all(&out)
.with_context(|| format!("writing {}", tmp.display()))?;
f.sync_all()
.with_context(|| format!("fsync {}", tmp.display()))?;
}
std::fs::rename(&tmp, &path)
.with_context(|| format!("publishing {}", path.display()))?;
Ok(())
}
fn graph_source_stats(&self) -> (u64, u64) {
(
self.graph_read.load(std::sync::atomic::Ordering::Relaxed),
self.graph_recomputed
.load(std::sync::atomic::Ordering::Relaxed),
)
}
fn ordinal_source_stats(&self) -> (u64, u64) {
(
self.ordinal_folded.load(std::sync::atomic::Ordering::Relaxed),
self.ordinal_tail.load(std::sync::atomic::Ordering::Relaxed),
)
}
fn reach_bitmaps(&self) -> Result<Arc<crate::reach::ReachTable>> {
self.reach_bitmaps_with(live_reach_policy(), true)
}
fn reach_bitmaps_with(
&self,
policy: crate::reach::ReachPolicy,
cache: bool,
) -> Result<Arc<crate::reach::ReachTable>> {
Ok(self.reach_table_with(policy, cache)?.0)
}
fn reach_table_with(
&self,
policy: crate::reach::ReachPolicy,
cache: bool,
) -> Result<(Arc<crate::reach::ReachTable>, Arc<Vec<u32>>)> {
{
let d = self
.derived
.read()
.map_err(|_| anyhow!("derived poisoned"))?;
if cache && !d.reach.is_empty() {
return Ok((Arc::clone(&d.reach), Arc::clone(&d.reach_by_row)));
}
if d.graph.is_empty() {
return Ok((Arc::default(), Arc::default()));
}
}
let mut d = self
.derived
.write()
.map_err(|_| anyhow!("derived poisoned"))?;
let trees = self.tree_source_for_build(d.ordinals.len());
let oid_len = self.hash.oid_len();
let facts = crate::reach::ObjectFacts {
ordinal: d.ordinals.view(),
trees: &trees,
oid_len,
};
let started = std::time::Instant::now();
let built = crate::reach::build_reach_table(&d.ord_graph, &facts, policy);
let by_row = built.by_row(&d.ord_graph);
self.reach_rebuilt
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let chains = trees.chain_stats();
tracing::info!(
store = %self.label,
entries = built.len(),
commits = d.graph.len(),
objects = d.ordinals.len(),
elapsed_ms = started.elapsed().as_millis() as u64,
chain_hits = chains.map(|c| c.0).unwrap_or(0),
chain_misses = chains.map(|c| c.1).unwrap_or(0),
chain_bytes = chains.map(|c| c.2).unwrap_or(0),
cache,
"built the reachability table from scratch"
);
if std::env::var_os(crate::arms::ENV_REACH_TIMING).is_some() {
eprintln!("reach build chain memo: {chains:?} (hits, misses, bytes)");
}
let built = Arc::new(built);
let by_row = Arc::new(by_row);
if cache {
d.reach = Arc::clone(&built);
d.reach_by_row = Arc::clone(&by_row);
self.persist_reach_locked(&d);
}
Ok((built, by_row))
}
fn read_extent(&self, offset: u64, len: u64) -> Result<Vec<u8>> {
use std::os::fd::AsRawFd;
let want = usize::try_from(len).context("extent length overflows usize")?;
let mut buf: Vec<u8> = Vec::with_capacity(want);
let fd = self.reader.as_raw_fd();
let mut filled = 0usize;
while filled < want {
let n = unsafe {
libc::pread(
fd,
buf.as_mut_ptr().add(filled).cast(),
want - filled,
i64::try_from(offset + filled as u64).context("extent offset overflows off_t")?,
)
};
if n < 0 {
let err = std::io::Error::last_os_error();
if err.kind() == std::io::ErrorKind::Interrupted {
continue;
}
return Err(err).with_context(|| {
format!("reading ({offset}, {len}) out of {}", self.blobs.display())
});
}
if n == 0 {
bail!(
"short read at ({offset}, {len}) out of {}: {filled} of {want} bytes before \
EOF — the archive is truncated relative to its index",
self.blobs.display()
);
}
filled += n as usize;
}
unsafe { buf.set_len(want) };
Ok(buf)
}
}
impl<S: ObjectIndex> ObjectAbsorb for Absorber<S> {
fn absorb(&self, job: IndexJob) -> Result<()> {
if self.cancel.load(std::sync::atomic::Ordering::Relaxed) {
return Err(crate::resolve::ResolveError::cancelled());
}
let dog = self.watchdog();
if dog.breached().is_some() {
self.progress.begin(job.pack_id, job.len, false);
self.progress.set_phase(crate::progress::Phase::Paused);
self.progress.add_pause();
let r = dog.pause_until_ok(&self.label, Some(&self.cancel), &mut || Ok(self.spill()));
self.progress.finish();
let paused = r?;
if paused.end.gave_up() {
return Err(crate::resolve::ResolveError::out_of_memory(
paused.end.as_str(),
paused.available,
paused.budget,
0,
0,
));
}
}
let _gate = self
.gate
.lock()
.map_err(|_| anyhow!("absorb gate poisoned"))?;
self.absorb_gated(&PendingPack::from_job(job))
}
}
impl<S: ObjectIndex + 'static> GitStore<S> {
pub(crate) fn push_path(&self) -> &PushPath {
&self.push
}
pub(crate) fn account(&self) -> &str {
&self.account
}
pub(crate) fn ref_log(&self) -> &RefLog {
&self.refs
}
pub(crate) fn ref_gate(&self) -> &Mutex<()> {
&self.ref_gate
}
pub(crate) fn external_bases_exist(&self, bytes: &[u8], w: &PackWalk) -> Result<()> {
let c = w.closure();
if !c.broken_offsets.is_empty() {
bail!(
"this pack is corrupt: {} delta base offset(s) do not land on an entry — first {}",
c.broken_offsets.len(),
c.broken_offsets[0]
);
}
let mut unheld: Vec<&[u8]> = Vec::new();
for oid in &c.external_refs {
if self.lookup_one(oid)?.is_none() {
unheld.push(oid.as_slice());
}
}
if unheld.is_empty() {
return Ok(());
}
let resolved = crate::resolve::resolve_walked(
bytes,
w,
self.hash_kind(),
0,
&*self.absorber,
&crate::exploded::NoSink,
)
.with_context(|| {
format!(
"this pack deltas against {} object(s) this repository does not have, so it was \
resolved to find out whether the pack carries them itself — first {}",
unheld.len(),
hex::encode(unheld[0])
)
})?;
let in_pack: std::collections::HashSet<&[u8]> =
resolved.iter().map(|r| r.oid.as_slice()).collect();
for oid in unheld {
if !in_pack.contains(oid) {
bail!(
"this pack deltas against {}, which this repository does not have and which \
the pack does not carry either — the push is refused rather than stored with \
a dangling base",
hex::encode(oid)
);
}
}
Ok(())
}
pub(crate) fn lookup_one(&self, oid: Oid<'_>) -> Result<Option<crate::index_layout::IndexRow>> {
self.absorber.lookup_never_wait(oid)
}
pub(crate) fn lookup_batch_never_wait(
&self,
oids: &[Oid<'_>],
) -> Result<Vec<Option<crate::index_layout::IndexRow>>> {
self.absorber.lookup_batch_never_wait(oids)
}
pub fn try_absorb_pending(&self) -> Result<Option<usize>> {
self.absorber.try_absorb_pending()
}
fn index_in_progress(&self) -> anyhow::Error {
let p = self.index_progress();
anyhow!(
"index in progress for this repository: pack {} {}% ({}/{} objects, {}) — reachability \
needs the whole graph; retry when the index row reaches 100 %",
p.pack_id,
p.percent(),
p.objects_done,
p.objects_total,
p.phase.as_str()
)
}
pub(crate) fn ref_state(&self) -> Result<BTreeMap<String, crate::refs::RefState>> {
self.refs.current()
}
pub(crate) fn ref_get(&self, name: &str) -> Result<Option<crate::refs::RefState>> {
self.refs.get(name)
}
pub(crate) fn reachable_oids(&self, want: &[Oid<'_>], have: &[Oid<'_>]) -> Result<Vec<String>> {
Ok(self
.reachable_raw_with(want, have, live_reach_policy(), true)?
.iter()
.map(hex::encode)
.collect())
}
#[cfg(test)]
pub(crate) fn reachable_oids_with(
&self,
want: &[Oid<'_>],
have: &[Oid<'_>],
policy: crate::reach::ReachPolicy,
cache: bool,
) -> Result<Vec<String>> {
Ok(self
.reachable_raw_with(want, have, policy, cache)?
.iter()
.map(hex::encode)
.collect())
}
pub(crate) fn reachable_raw(
&self,
want: &[Oid<'_>],
have: &[Oid<'_>],
) -> Result<git_storage_trait::OidList> {
self.reachable_raw_with(want, have, live_reach_policy(), true)
}
pub(crate) fn reachable_raw_with(
&self,
want: &[Oid<'_>],
have: &[Oid<'_>],
policy: crate::reach::ReachPolicy,
cache: bool,
) -> Result<git_storage_trait::OidList> {
Ok(self
.select_with(want, have, policy, cache, false)?
.map(|s| s.objects)
.unwrap_or_default())
}
pub(crate) fn select_with(
&self,
want: &[Oid<'_>],
have: &[Oid<'_>],
policy: crate::reach::ReachPolicy,
cache: bool,
require_commit_tips: bool,
) -> Result<Option<Selected>> {
if self.try_absorb_pending()?.is_none() {
return Err(self.index_in_progress());
}
let (table, by_row) = self.absorber.reach_table_with(policy, cache)?;
let d = self
.absorber
.derived
.read()
.map_err(|_| anyhow!("derived poisoned"))?;
let space = d.ordinals.view();
let oid_len = self.hash.oid_len();
let graph: &crate::reach::OrdGraph = &d.ord_graph;
let view = crate::reach::ReachView {
table: &table,
by_row: &by_row,
graph,
};
let trees = self.absorber.tree_source();
let facts = crate::reach::ObjectFacts {
ordinal: space,
trees: &trees,
oid_len,
};
if require_commit_tips {
if graph.is_empty() {
return Ok(None);
}
for oid in want {
let Some(o) = space.ordinal_of_raw(oid) else { return Ok(None) };
if !graph.is_commit(o) {
return Ok(None);
}
}
}
let mut union = roaring::RoaringBitmap::new();
let mut later: Vec<&[u8]> = Vec::new();
for oid in want {
if let Some(o) = space.ordinal_of_raw(oid) {
if let Some(bm) = graph.row_of(o).and_then(|row| view.bitmap_of_row(row)) {
union |= bm;
continue;
}
}
later.push(oid);
}
for oid in later {
let Some(o) = space.ordinal_of_raw(oid) else {
bail!(
"want {} is not in this repository — a negotiation must not be answered from a \
partial set",
hex::encode(oid)
);
};
match graph.row_of(o) {
Some(row) => match view.bitmap_of_row(row) {
Some(bm) => union |= bm,
None => crate::reach::accumulate(row, &view, &facts, &mut union),
},
None => {
union.insert(o);
peel_tag_into(self.absorber.as_ref(), o, &view, &facts, &mut union);
}
}
}
let mut had = roaring::RoaringBitmap::new();
let mut later: Vec<&[u8]> = Vec::new();
for oid in have {
if let Some(o) = space.ordinal_of_raw(oid) {
if let Some(bm) = graph.row_of(o).and_then(|row| view.bitmap_of_row(row)) {
had |= bm;
continue;
}
}
later.push(oid);
}
for oid in later {
let Some(o) = space.ordinal_of_raw(oid) else { continue };
match graph.row_of(o) {
Some(row) => match view.bitmap_of_row(row) {
Some(bm) => had |= bm,
None => crate::reach::accumulate(row, &view, &facts, &mut had),
},
None => {
had.insert(o);
}
}
}
let delta = &union - &had;
let commits = &delta & graph.commits();
let to_list = |bm: &roaring::RoaringBitmap| -> Result<git_storage_trait::OidList> {
let mut out = git_storage_trait::OidList::with_capacity(bm.len() as usize, oid_len);
for o in bm {
let raw = space
.oid(o)
.ok_or_else(|| anyhow!("ordinal {o} is outside this store's ordinal space"))?;
out.push(raw)?;
}
Ok(out)
};
let objects = to_list(&delta)?;
let commits: Vec<Vec<u8>> = commits
.iter()
.map(|o| space.oid(o).map(<[u8]>::to_vec).ok_or_else(|| anyhow!("ordinal {o} is outside this store's ordinal space")))
.collect::<Result<_>>()?;
let client_has = if have.is_empty() { git_storage_trait::OidList::new() } else { to_list(&had)? };
Ok(Some(Selected {
objects,
commits,
client_has,
}))
}
}
pub(crate) struct Selected {
pub objects: git_storage_trait::OidList,
pub commits: Vec<Vec<u8>>,
pub client_has: git_storage_trait::OidList,
}
impl<S: ObjectIndex + 'static> GitStore<S> {
pub(crate) fn live_set(&self) -> Result<std::collections::HashSet<String>> {
let refs = self.ref_state()?;
let tips: Vec<Vec<u8>> = refs
.values()
.filter_map(|s| s.target.as_deref())
.filter_map(|t| hex::decode(t).ok())
.collect();
if tips.is_empty() {
bail!(
"this repository has no ref pointing at an object, so every object in it would be \
dead. A GC that would delete everything is refused: name a ref first"
);
}
let borrowed: Vec<Oid<'_>> = tips.iter().map(|t| t.as_slice()).collect();
let mut live: std::collections::HashSet<String> =
self.reachable_oids(&borrowed, &[])?.into_iter().collect();
for s in refs.values() {
if let Some(p) = &s.peeled {
live.insert(p.clone());
}
if let Some(t) = &s.target {
live.insert(t.clone());
}
}
Ok(live)
}
pub(crate) fn drop_dead_rows(&self, dead: &std::collections::HashSet<Vec<u8>>) -> Result<u64> {
let dropped = self
.absorber
.objects
.retain(&|oid: &[u8]| !dead.contains(oid))?;
self.absorber
.exploded
.retain(&|oid: &[u8]| !dead.contains(oid))?;
self.absorber.refold()?;
Ok(dropped)
}
pub(crate) fn acked_end(&self) -> Result<u64> {
let Some(journal) = self.arms.writer.journal(&self.blobs) else {
return Ok(u64::MAX);
};
if !journal.exists() {
return Ok(0);
}
let rows = crate::archive_write::read_journal(&journal)?;
Ok(crate::archive_write::acked_packs(&rows)
.iter()
.map(|(o, l)| o + l)
.max()
.unwrap_or(0))
}
pub(crate) fn retire_dead_packs(
&self,
live: &std::collections::HashSet<String>,
acked_end: u64,
) -> Result<Vec<u64>> {
let Some(journal) = self.arms.writer.journal(&self.blobs) else {
return Ok(Vec::new());
};
if !journal.exists() {
return Ok(Vec::new());
}
let rows = crate::archive_write::read_journal(&journal)?;
let packs: Vec<(u64, u64)> = crate::archive_write::acked_packs(&rows)
.into_iter()
.filter(|(o, l)| o + l <= acked_end)
.collect();
let already = crate::archive_write::retired_offsets(&rows);
if packs.is_empty() {
return Ok(Vec::new());
}
let occupied = self.absorber.objects.extents_with_rows(&packs)?;
let mut order: Vec<usize> = (0..packs.len()).filter(|&i| packs[i].1 > 0).collect();
order.sort_unstable_by_key(|&i| packs[i].0);
let starts: Vec<u64> = order.iter().map(|&i| packs[i].0).collect();
let live_oids: Vec<Vec<u8>> = live.iter().filter_map(|h| hex::decode(h).ok()).collect();
let borrowed: Vec<&[u8]> = live_oids.iter().map(|o| o.as_slice()).collect();
let mut has_live = vec![false; packs.len()];
for (offset, _) in self
.absorber
.objects
.extents_batch(&borrowed)
.into_iter()
.flatten()
{
let p = starts.partition_point(|&s| s <= offset);
if p == 0 {
continue;
}
let i = order[p - 1];
if offset < packs[i].0 + packs[i].1 {
has_live[i] = true;
}
}
let dead: Vec<u64> = (0..packs.len())
.filter(|&i| occupied[i] && !has_live[i] && !already.contains(&packs[i].0))
.map(|i| packs[i].0)
.collect();
crate::archive_write::retire_packs(&journal, &dead)?;
Ok(dead)
}
pub(crate) fn seal_archive(
&self,
reserved: Vec<ReservedSection>,
) -> Result<crate::archive_write::SealReport> {
crate::archive_write::seal_generation_zero(
&self.blobs,
self.arms.writer.journal(&self.blobs).as_deref(),
&self.archive,
reserved,
)
}
pub(crate) fn reserved_sections(&self) -> Result<Vec<ReservedSection>> {
use znippy_common::{GUNNAR_GRAPH_MODULE, GUNNAR_REACH_MODULE};
let mut out = vec![self.refs.seal_section()?];
let _ = self.reach_bitmaps()?;
let d = self
.absorber
.derived
.read()
.map_err(|_| anyhow!("derived poisoned"))?;
if !d.graph.is_empty() {
out.push(ReservedSection::arrow(
GUNNAR_GRAPH_MODULE,
crate::graph::graph_schema(),
vec![crate::graph::build_graph_batch(&d.graph)?],
));
}
if !d.reach.is_empty() {
let sorted = self.absorber.objects.oids_raw()?.raw;
let oid_len = self.hash.oid_len();
let target = crate::reach::OrdinalSpace::new(&sorted, oid_len);
let remap = d.ordinals.remap_to(&target);
if let Some(entries) = d.reach.to_entries(&target, Some(&remap)) {
out.push(ReservedSection::arrow(
GUNNAR_REACH_MODULE,
crate::reach::reach_schema(),
vec![crate::reach::build_reach_batch(&entries)?],
));
} else {
tracing::warn!(
store = %self.absorber.label,
"seal: the live reachability table names an object the sorted space lacks; \
no __gunnar_reach__ section is written"
);
}
}
Ok(out)
}
}
pub enum SelectedStore {
OneTableFourColumns(GitStore<OneTableFourColumns>),
FourTables(GitStore<crate::index_layout::FourTables>),
PackedPayload(GitStore<crate::index_layout::PackedPayload>),
}
macro_rules! on_arm {
($self:ident, $s:ident => $body:expr) => {
match $self {
SelectedStore::OneTableFourColumns($s) => $body,
SelectedStore::FourTables($s) => $body,
SelectedStore::PackedPayload($s) => $body,
}
};
}
impl SelectedStore {
pub fn arms(&self) -> StoreConfig {
on_arm!(self, s => s.arms())
}
pub fn explode_cache_use(&self) -> crate::exploded_cache::CacheUse {
on_arm!(self, s => s.explode_cache_use())
}
pub fn set_explode_budget(&self, bytes: u64) {
on_arm!(self, s => s.set_explode_budget(bytes))
}
pub fn drop_explode_cache(&self) -> anyhow::Result<u64> {
on_arm!(self, s => s.drop_explode_cache())
}
pub fn writer_name(&self) -> &'static str {
on_arm!(self, s => s.writer_name())
}
pub fn writer_durability(&self) -> &'static str {
on_arm!(self, s => s.writer_durability())
}
pub fn index_ipc_bytes(&self) -> usize {
on_arm!(self, s => s.index().ipc_bytes())
}
pub fn index_name(&self) -> &'static str {
on_arm!(self, s => s.index().projection_name())
}
pub fn oids_in_extent(&self, extent: Extent) -> Result<Vec<Vec<u8>>> {
on_arm!(self, s => s.oids_in_extent(extent))
}
pub fn object_count(&self) -> usize {
on_arm!(self, s => s.object_count())
}
pub fn wait_indexed(&self) {
on_arm!(self, s => s.wait_indexed())
}
pub fn cancel_index(&self) {
on_arm!(self, s => s.cancel_index())
}
pub fn index_progress(&self) -> crate::progress::Snapshot {
on_arm!(self, s => s.index_progress())
}
pub fn rebuild_projection(&self) -> Result<()> {
on_arm!(self, s => s.index().rebuild())
}
pub fn seal(&self) -> Result<Vec<ReservedSection>> {
on_arm!(self, s => s.seal())
}
}
impl GitOps for SelectedStore {
fn put(&self, pack: &[u8], refs: &[RefUpdate]) -> Result<TxId> {
on_arm!(self, s => s.put(pack, refs))
}
fn put_pack(&self, bytes: &[u8]) -> Result<TxId> {
on_arm!(self, s => s.put_pack(bytes))
}
fn put_pack_observed(
&self,
bytes: &[u8],
progress: &dyn git_storage_trait::PackProgress,
) -> Result<TxId> {
on_arm!(self, s => s.put_pack_observed(bytes, progress))
}
fn publishes_pack_progress(&self) -> bool {
on_arm!(self, s => s.publishes_pack_progress())
}
fn put_refs(&self, updates: &[RefUpdate]) -> Result<TxId> {
on_arm!(self, s => s.put_refs(updates))
}
fn get(&self, oid: Oid<'_>) -> Result<Option<Stored>> {
on_arm!(self, s => s.get(oid))
}
fn has(&self, oid: Oid<'_>) -> Result<bool> {
on_arm!(self, s => s.has(oid))
}
fn size(&self, oid: Oid<'_>) -> Result<Option<u64>> {
on_arm!(self, s => s.size(oid))
}
fn extents(&self, oids: &[Oid<'_>]) -> Result<Vec<Option<Extent>>> {
on_arm!(self, s => s.extents(oids))
}
fn refs(&self) -> Result<Vec<RefRow>> {
on_arm!(self, s => s.refs())
}
fn update_ref(&self, name: &str, old: Option<Oid<'_>>, new: Option<Oid<'_>>) -> Result<TxId> {
on_arm!(self, s => s.update_ref(name, old, new))
}
fn put_refs_cas(&self, edits: &[git_storage_trait::RefCas<'_>]) -> Result<TxId> {
on_arm!(self, s => s.put_refs_cas(edits))
}
fn reachable(&self, want: &[Oid<'_>], have: &[Oid<'_>]) -> Result<Vec<Vec<u8>>> {
on_arm!(self, s => s.reachable(want, have))
}
fn gc(&self) -> Result<GcReport> {
on_arm!(self, s => s.gc())
}
}
impl crate::serve::GitServe for SelectedStore {
fn read(&self, oid: Oid<'_>) -> Result<Option<(crate::index_layout::ObjType, Vec<u8>)>> {
on_arm!(self, s => crate::serve::GitServe::read(s, oid))
}
fn header(&self, oid: Oid<'_>) -> Result<Option<(crate::index_layout::ObjType, u64)>> {
on_arm!(self, s => crate::serve::GitServe::header(s, oid))
}
fn sizes(&self, oids: &[Oid<'_>]) -> Result<Vec<Option<u64>>> {
on_arm!(self, s => crate::serve::GitServe::sizes(s, oids))
}
fn index_now(&self) -> Result<usize> {
on_arm!(self, s => crate::serve::GitServe::index_now(s))
}
fn head(&self) -> Result<Option<RefRow>> {
on_arm!(self, s => crate::serve::GitServe::head(s))
}
fn set_head(&self, target: &str) -> Result<TxId> {
on_arm!(self, s => crate::serve::GitServe::set_head(s, target))
}
fn emit_pack(
&self,
objects: &[Oid<'_>],
have: &[Oid<'_>],
caps: &git_storage_trait::Caps,
out: &mut dyn std::io::Write,
) -> Result<git_storage_trait::PackStats> {
on_arm!(self, s => crate::serve::GitServe::emit_pack(s, objects, have, caps, out))
}
fn select(
&self,
want: &[Oid<'_>],
have: &[Oid<'_>],
) -> Result<Option<git_storage_trait::ReachSet>> {
on_arm!(self, s => crate::serve::GitServe::select(s, want, have))
}
}
pub fn open_selected(
root: &Path,
account: &str,
hash: GitHashKind,
arms: StoreConfig,
) -> Result<SelectedStore> {
use crate::arms::IndexArm;
use crate::index_layout::{FourTables, PackedPayload};
Ok(match arms.index {
IndexArm::OneTableFourColumns => {
SelectedStore::OneTableFourColumns(GitStore::<OneTableFourColumns>::open_with_arms(
root, account, hash, arms,
)?)
}
IndexArm::FourTables => SelectedStore::FourTables(GitStore::<FourTables>::open_with_arms(
root, account, hash, arms,
)?),
IndexArm::PackedPayload => SelectedStore::PackedPayload(
GitStore::<PackedPayload>::open_with_arms(root, account, hash, arms)?,
),
})
}
pub fn open_from_env(root: &Path, account: &str, hash: GitHashKind) -> Result<SelectedStore> {
let arms = StoreConfig::from_env()?;
open_selected(root, account, hash, arms)
}
impl<S: ObjectIndex> Absorber<S> {
fn resolved(&self, oid: &[u8]) -> Result<Option<(GitObjectKind, Vec<u8>)>> {
Ok(self
.resolved_costed(oid)?
.map(|(kind, payload, _)| (kind, payload)))
}
fn resolved_cached(
&self,
oid: &[u8],
cache: &mut crate::resolve::ChainCache,
) -> Result<Option<(GitObjectKind, Vec<u8>)>> {
Ok(self
.resolved_costed_with(oid, Some(cache))?
.map(|(kind, payload, _)| (kind, payload)))
}
#[allow(clippy::type_complexity)]
fn resolved_costed(
&self,
oid: &[u8],
) -> Result<Option<(GitObjectKind, Vec<u8>, Option<crate::resolve::ReadCost>)>> {
self.resolved_costed_with(oid, None)
}
#[allow(clippy::type_complexity)]
fn resolved_costed_with(
&self,
oid: &[u8],
cache: Option<&mut crate::resolve::ChainCache>,
) -> Result<Option<(GitObjectKind, Vec<u8>, Option<crate::resolve::ReadCost>)>> {
if let Some((kind, payload)) = self.exploded.content(oid)? {
return Ok(Some((kind, payload, None)));
}
let Some(row) = self.objects.lookup(oid) else {
if self.unindexed_packs() == 0 {
return Ok(None);
}
self.exploded.note_rederived();
let (_, content) = self.scan_pending(&[], Some(oid))?;
return Ok(content.map(|(k, p)| (k, p, None)));
};
self.exploded.note_rederived();
let snap = self.map.snapshot()?;
let src = ArchiveWindow { abs: self, snap: &snap };
let offset_of = |base: &[u8]| self.objects.lookup(base).map(|r| r.offset);
match crate::resolve::materialize_at_with(
&src,
row.offset,
self.hash.oid_len(),
&offset_of,
self,
cache,
) {
Ok((kind, payload, cost)) => {
self.keep_for_the_next_reader(oid, kind, &payload, cost);
Ok(Some((kind, payload, Some(cost))))
}
Err(e) => {
if self.unindexed_packs() == 0 {
return Err(e);
}
match self.scan_pending(&[], Some(oid))? {
(_, Some((k, p))) => Ok(Some((k, p, None))),
(_, None) => Err(e),
}
}
}
}
}
impl<S: ObjectIndex> Absorber<S> {
fn keep_for_the_next_reader(
&self,
oid: &[u8],
kind: GitObjectKind,
payload: &[u8],
cost: crate::resolve::ReadCost,
) {
match self.exploded.write_back(oid, kind, payload, cost) {
Ok(true) => self.exploded.kick(),
Ok(false) => {}
Err(e) => {
tracing::debug!(
target: "znippy.git",
store = %self.label,
error = %e,
"the exploded cache would not take a write-back; the read is unaffected"
);
}
}
}
}
fn tag_target_raw(payload: &[u8], oid_len: usize) -> Option<[u8; 32]> {
let line = payload.split(|b| *b == b'\n').next()?;
let rest = line.strip_prefix(b"object ")?;
if rest.len() != oid_len * 2 || oid_len > 32 {
return None;
}
let mut out = [0u8; 32];
hex::decode_to_slice(rest, &mut out[..oid_len]).ok()?;
Some(out)
}
fn peel_tag_into<S: ObjectIndex>(
abs: &Absorber<S>,
tag: u32,
view: &crate::reach::ReachView<'_>,
facts: &crate::reach::ObjectFacts<'_>,
union: &mut roaring::RoaringBitmap,
) {
const MAX_TAG_HOPS: usize = 8;
let oid_len = facts.oid_len;
let mut cursor = tag;
for _ in 0..MAX_TAG_HOPS {
let Some(raw) = facts.ordinal.oid(cursor) else { return };
let Ok(Some((GitObjectKind::Tag, payload))) = abs.resolved(raw) else { return };
let Some(target) = tag_target_raw(&payload, oid_len) else { return };
let target = &target[..oid_len];
let Some(t) = facts.ordinal.ordinal_of_raw(target) else { return };
if let Some(row) = view.graph.row_of(t) {
match view.bitmap_of_row(row) {
Some(bm) => *union |= bm,
None => crate::reach::accumulate(row, view, facts, union),
}
return;
}
match abs.resolved(target) {
Ok(Some((GitObjectKind::Tree, _))) => {
crate::reach::fill_tree(t, facts, union);
return;
}
Ok(Some((GitObjectKind::Tag, _))) => {
union.insert(t);
cursor = t;
}
_ => {
union.insert(t);
return;
}
}
}
}
struct TreesFromStore<'a, S: ObjectIndex> {
abs: &'a Absorber<S>,
chains: Option<Mutex<crate::resolve::ChainCache>>,
}
impl<S: ObjectIndex> TreesFromStore<'_, S> {
fn chain_stats(&self) -> Option<(u64, u64, usize)> {
let c = self.chains.as_ref()?.lock().ok()?;
let (h, m) = c.stats();
Some((h, m, c.bytes()))
}
}
impl<S: ObjectIndex> crate::reach::TreeSource for TreesFromStore<'_, S> {
fn tree(&self, oid: &[u8]) -> Option<std::borrow::Cow<'_, [u8]>> {
let read = match &self.chains {
Some(m) => match m.lock() {
Ok(mut c) => self.abs.resolved_cached(oid, &mut c),
Err(_) => self.abs.resolved(oid),
},
None => self.abs.resolved(oid),
};
match read {
Ok(Some((GitObjectKind::Tree, payload))) => Some(std::borrow::Cow::Owned(payload)),
Ok(_) => None,
Err(e) => {
tracing::debug!(
target: "znippy.git",
store = %self.abs.label,
oid = %hex::encode(oid),
error = %e,
"a reachability walk could not read a tree; the walk stops there"
);
None
}
}
}
}
struct ArchiveWindow<'a, S: ObjectIndex> {
abs: &'a Absorber<S>,
snap: &'a crate::archive_map::Mapped,
}
impl<S: ObjectIndex> crate::resolve::ArchiveBytes for ArchiveWindow<'_, S> {
fn at(&self, at: u64, len: u64) -> Result<std::borrow::Cow<'_, [u8]>> {
clamped_window(self.abs, self.snap, at, len)
}
}
fn clamped_window<'m, S: ObjectIndex>(
abs: &Absorber<S>,
snap: &'m crate::archive_map::Mapped,
at: u64,
len: u64,
) -> Result<std::borrow::Cow<'m, [u8]>> {
let mapped = snap.len();
if at < mapped {
let have = (mapped - at).min(len);
if let Some(b) = snap.get(at, have) {
return Ok(std::borrow::Cow::Borrowed(b));
}
}
let end = abs.reader.metadata().map(|m| m.len()).unwrap_or(mapped);
if at >= end {
return Ok(std::borrow::Cow::Owned(Vec::new()));
}
let have = (end - at).min(len);
Ok(std::borrow::Cow::Owned(abs.read_extent(at, have)?))
}
impl<S: ObjectIndex> BaseSource for Absorber<S> {
fn content(&self, oid: &[u8]) -> Result<Option<(GitObjectKind, Vec<u8>)>> {
self.resolved(oid).with_context(|| {
format!(
"reading the delta base {} out of this store — it is not absent, it could not \
be reconstructed",
hex::encode(oid)
)
})
}
}
impl<S: ObjectIndex> Drop for Absorber<S> {
fn drop(&mut self) {
if self.reach_unpersisted.load(std::sync::atomic::Ordering::Relaxed) == 0 {
return;
}
if let Ok(d) = self.derived.read() {
self.persist_reach_locked(&d);
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::archive_write::{read_journal, SafeWriter};
use crate::object::GitObjectKind;
use crate::store::tests::{one_blob_pack, corpus_pack, tmpdir};
use std::collections::HashSet;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering as AtomicOrdering};
use std::time::{Duration, Instant};
use znippy_zoomies::background::Job;
use znippy_zoomies::gatling_forkjoin::gatling_for_each;
fn loadavg() -> String {
std::fs::read_to_string("/proc/loadavg")
.unwrap_or_default()
.split_whitespace()
.take(3)
.collect::<Vec<_>>()
.join(" ")
}
#[test]
fn the_one_ordinal_space_answers_what_the_two_it_replaced_did() {
let dir = tmpdir("ordinal-space-one-representation");
let (pack, rows) = corpus_pack();
let store = GitStore::open(&dir, "alice").unwrap();
store.put_pack(&pack).unwrap();
store.wait_indexed();
let oid_len = store.hash_kind().oid_len();
let raw = {
let d = store.absorber.derived.read().unwrap();
d.ordinals.view().concat()
};
let old_oids: Vec<String> = raw.chunks_exact(oid_len).map(hex::encode).collect();
let old_ordinal: std::collections::HashMap<String, u32> = old_oids
.iter()
.enumerate()
.map(|(i, o)| (o.clone(), i as u32))
.collect();
assert!(
!old_oids.is_empty(),
"the fixture pack ({} entries) folded to an EMPTY ordinal space, so every \
comparison below would hold vacuously",
rows.len()
);
assert_eq!(
raw.len(),
old_oids.len() * oid_len,
"the raw ordinal space is {} bytes for {} oids of {oid_len} bytes",
raw.len(),
old_oids.len()
);
let space = crate::reach::OrdinalSpace::new(&raw, oid_len);
assert_eq!(space.len(), old_oids.len(), "the space holds a different count");
let mut checked = 0usize;
for (o, h) in old_oids.iter().enumerate() {
let want_ord = old_ordinal[h];
assert_eq!(
space.ordinal_of_hex(h),
Some(want_ord),
"ordinal_of_hex({h}) != the old HashMap answer {want_ord}"
);
let raw_oid = hex::decode(h).unwrap();
assert_eq!(
space.ordinal_of_raw(&raw_oid),
Some(want_ord),
"ordinal_of_raw disagrees with the old HashMap at {h}"
);
assert_eq!(
space.oid(o as u32),
Some(raw_oid.as_slice()),
"oid({o}) != the old oids_raw slice"
);
assert_eq!(
space.oid(o as u32).map(hex::encode).as_deref(),
Some(h.as_str()),
"oid({o}) hex != the old hex Vec entry {h}"
);
checked += 1;
}
assert_eq!(checked, old_oids.len());
let absent = vec![0xffu8; oid_len];
assert_eq!(space.ordinal_of_raw(&absent), None, "an absent oid resolved to an ordinal");
assert_eq!(
space.ordinal_of_hex(&hex::encode(&absent)),
None,
"an absent hex oid resolved to an ordinal"
);
assert_eq!(space.ordinal_of_raw(&absent[..oid_len - 1]), None);
assert_eq!(space.oid(old_oids.len() as u32), None, "an out-of-range ordinal named an oid");
let tip = store
.graph_snapshot()
.into_iter()
.max_by_key(|c| c.generation)
.expect("a graph");
let tip_raw = hex::decode(&tip.oid).unwrap();
let flat = store.reachable_raw(&[&tip_raw], &[]).unwrap();
let hexed = store.reachable_oids(&[&tip_raw], &[]).unwrap();
assert!(
flat.len() > 1,
"the tip's closure is {} object(s) — too small to tell an ordering bug from a \
coincidence",
flat.len()
);
assert_eq!(
flat.iter().map(hex::encode).collect::<Vec<_>>(),
hexed,
"the serving answer and the maintenance answer name different objects"
);
assert!(flat.contains(&tip_raw), "a closure without its own tip");
eprintln!(
"load {}; {} ordinals agree in both spellings, tip closure {} objects",
loadavg(),
checked,
flat.len()
);
}
fn commits_in(rows: &[crate::resolve::Resolved]) -> usize {
rows.iter()
.filter(|r| r.kind == GitObjectKind::Commit)
.map(|r| r.oid.clone())
.collect::<HashSet<Vec<u8>>>()
.len()
}
#[test]
fn a_push_ends_in_object_rows_and_no_read_asked_for_them() {
let dir = tmpdir("drain-rows");
let store = GitStore::open(&dir, "alice").unwrap();
let (pack, rows) = corpus_pack();
let t = Instant::now();
let tx = store.put_pack(&pack).unwrap();
let ack = t.elapsed();
let pack_id = tx.pack_id.expect("a pack push assigns an id");
store.wait_indexed();
let drained = t.elapsed();
assert_eq!(
store.object_count(),
rows.len(),
"the drain built no object rows: {} of {}",
store.object_count(),
rows.len()
);
assert_eq!(store.unindexed_packs(), 0, "the bit is still clear");
assert!(
store.indexer().is_indexed(pack_id),
"the pack's indexed bit is not up after the drain"
);
assert_eq!(
store.indexer().absorb_failures(),
0,
"the drain recorded an absorb failure: {:?}",
store.indexer().last_absorb_error()
);
for (i, r) in rows.iter().enumerate() {
let row = store
.index()
.lookup(&r.oid)
.unwrap_or_else(|| panic!("object {i} {} has no row", hex::encode(&r.oid)));
assert_eq!(
(row.offset, row.len),
(tx.extent.unwrap().0 + r.offset, r.len)
);
assert_eq!(row.uncompressed_size, r.uncompressed_size);
}
assert_eq!(
store.absorb_pending().unwrap(),
0,
"the drain left work for a read to do"
);
assert_eq!(
store.commit_count(),
commits_in(&rows),
"the commit graph does not match the pack"
);
let secs = (drained - ack).as_secs_f64();
eprintln!(
"load {}; {} objects in {} bytes: ack {:.0} µs, drain {:.1} ms, {:.0} object rows/s",
loadavg(),
rows.len(),
pack.len(),
ack.as_secs_f64() * 1e6,
(drained - ack).as_secs_f64() * 1e3,
rows.len() as f64 / secs.max(1e-9),
);
}
const KILL_DIR: &str = "GUNNAR_KILL_MID_ABSORB_DIR";
#[test]
#[ignore = "spawned by a_store_reopened_after_a_kill_mid_absorb_requeues_the_pack; it aborts on purpose"]
fn the_child_that_dies_between_durable_bytes_and_indexed_rows() {
let Ok(dir) = std::env::var(KILL_DIR) else {
return;
};
let dir = PathBuf::from(dir);
let pack = std::fs::read(dir.join("fixture.pack")).expect("the parent's fixture");
let store = GitStore::open(&dir, "alice").expect("open");
let _gate = store.hold_absorb_gate();
let tx = store.put_pack(&pack).expect("the ack path");
assert_eq!(tx.pack_id, Some(0), "a fresh archive starts at ordinal 0");
assert_eq!(store.unindexed_packs(), 1, "the pack is not queued");
assert_eq!(store.index().len(), 0, "a row landed before the kill");
std::fs::write(dir.join("ready"), b"durable, not indexed").expect("marker");
std::process::abort();
}
#[test]
fn a_store_reopened_after_a_kill_mid_absorb_requeues_the_pack() {
use std::os::unix::process::ExitStatusExt;
let dir = tmpdir("kill-mid-absorb");
let (pack, rows) = corpus_pack();
std::fs::write(dir.join("fixture.pack"), &pack).unwrap();
let status = std::process::Command::new(std::env::current_exe().unwrap())
.args([
"--exact",
"--ignored",
"--nocapture",
"git_ops::tests::the_child_that_dies_between_durable_bytes_and_indexed_rows",
])
.env(KILL_DIR, &dir)
.status()
.expect("spawn the child that dies");
assert_eq!(
status.signal(),
Some(6),
"the child did not die by SIGABRT — it {status:?}, so nothing was interrupted"
);
assert!(
dir.join("ready").exists(),
"the child never reached the kill point"
);
let blobs = dir.join("objects.pack");
let acked = read_journal(&SafeWriter::journal_path(&blobs)).unwrap();
assert_eq!(
acked.len(),
1,
"the pack's extent is not durable: {acked:?}"
);
assert_eq!(acked[0].1, pack.len() as u64);
let on_disk = std::fs::read(&blobs).unwrap();
assert_eq!(
&on_disk[acked[0].0 as usize..(acked[0].0 + acked[0].1) as usize],
&pack[..],
"the verbatim bytes did not survive the kill"
);
{
let tail = ObjectReadStack::<OneTableFourColumns>::open(
&dir.join("objects.tail"),
RebuildTriggers::default(),
crate::arms::DEFAULT_REDB_CACHE_BYTES,
)
.unwrap();
assert_eq!(
tail.len(),
0,
"the kill landed after the rows were indexed, not between the bytes and the rows \
— this guard would then be testing nothing"
);
}
let store = GitStore::open(&dir, "alice").unwrap();
assert!(
store.has(&rows[0].oid).unwrap(),
"a durable object read `absent` from a store reopened over the pack that holds it — \
the crash-recovery bit did not re-queue anything"
);
store.wait_indexed();
assert_eq!(
store.object_count(),
rows.len(),
"the re-queued pack's objects never reached the index — left: {}, right: {}",
store.object_count(),
rows.len()
);
assert!(
store.indexer().is_indexed(0),
"the re-queued pack's indexed bit never went up in the new process"
);
assert_eq!(store.unindexed_packs(), 0);
assert_eq!(store.absorb_pending().unwrap(), 0);
for r in rows.iter().take(512) {
let row = store
.index()
.lookup(&r.oid)
.unwrap_or_else(|| panic!("{} has no row after recovery", hex::encode(&r.oid)));
assert_eq!((row.offset, row.len), (acked[0].0 + r.offset, r.len));
}
assert_eq!(
store.commit_count(),
commits_in(&rows),
"the commit graph did not come back: {} of {}",
store.commit_count(),
commits_in(&rows)
);
let tip = store
.graph_snapshot()
.into_iter()
.max_by_key(|c| c.generation)
.expect("a graph");
let tip_raw = hex::decode(&tip.oid).unwrap();
let closure = store.reachable(&[&tip_raw], &[]).unwrap();
assert!(
!closure.is_empty(),
"reachable() came back empty after the reopen"
);
assert!(closure.contains(&tip_raw));
eprintln!(
"load {}; killed mid-absorb: {} journal extent(s) recovered {} object rows, {} \
commits, closure of the tip {} objects",
loadavg(),
acked.len(),
store.object_count(),
store.commit_count(),
closure.len(),
);
}
#[test]
fn a_clean_reopen_requeues_nothing_and_the_next_push_gets_a_fresh_ordinal() {
let dir = tmpdir("clean-reopen");
let (pack, rows) = corpus_pack();
let (second, blob_oid) = one_blob_pack(b"a push that arrives after the restart");
{
let store = GitStore::open(&dir, "alice").unwrap();
store.put_pack(&pack).unwrap();
store.wait_indexed();
assert_eq!(
store.object_count(),
rows.len(),
"the first run did not index"
);
}
let store = GitStore::open(&dir, "alice").unwrap();
store.wait_indexed();
assert!(
!store.indexer().is_indexed(0),
"a fully absorbed pack was re-queued on open — the diff is not exact"
);
assert_eq!(store.unindexed_packs(), 0);
assert_eq!(
store.absorb_pending().unwrap(),
0,
"the reopen left index work"
);
assert_eq!(
store.object_count(),
rows.len(),
"the reopened store lost rows"
);
assert!(store.has(&rows[0].oid).unwrap());
let tx = store.put_pack(&second).unwrap();
assert_eq!(
tx.pack_id,
Some(1),
"a reopened store handed a fresh pack the ordinal of an absorbed one"
);
store.wait_indexed();
assert!(
store.has(&blob_oid).unwrap(),
"the second push's object never reached the index — its ordinal collided with an \
absorbed pack's, so every writer skipped it"
);
assert_eq!(store.object_count(), rows.len() + 1);
assert!(store.indexer().is_indexed(1));
let acked = read_journal(&SafeWriter::journal_path(store.blobs_path())).unwrap();
assert_eq!(
acked.len(),
2,
"the reopen truncated the journal: {acked:?}"
);
assert_eq!(acked[1], tx.extent.unwrap());
}
#[test]
fn the_derive_on_open_diff_costs_milliseconds_at_a_realistic_pack_count() {
const PACKS: usize = 1000;
let dir = tmpdir("diff-cost");
let (real, rows) = corpus_pack();
{
let store = GitStore::open(&dir, "alice").unwrap();
store.put_pack(&real).unwrap();
for i in 0..PACKS - 1 {
let (p, _) = one_blob_pack(format!("pack number {i}").as_bytes());
store.put_pack(&p).unwrap();
}
store.wait_indexed();
assert_eq!(
store.object_count(),
rows.len() + PACKS - 1,
"the fixture did not absorb"
);
}
let acked = read_journal(&SafeWriter::journal_path(&dir.join("objects.pack"))).unwrap();
assert_eq!(acked.len(), PACKS, "one journal row per pack");
let tail = ObjectReadStack::<OneTableFourColumns>::open(
&dir.join("objects.tail"),
RebuildTriggers::default(),
crate::arms::DEFAULT_REDB_CACHE_BYTES,
)
.unwrap();
let t = Instant::now();
let hit = tail.extents_with_rows(&acked).unwrap();
let clean = t.elapsed();
assert!(
hit.iter().all(|&h| h),
"an absorbed pack was reported unabsorbed"
);
let mut in_flight = acked.clone();
let end = acked[PACKS - 1].0 + acked[PACKS - 1].1;
in_flight.push((end, 4096));
let t = Instant::now();
let hit = tail.extents_with_rows(&in_flight).unwrap();
let crashed = t.elapsed();
assert_eq!(
hit.iter().filter(|h| !**h).count(),
1,
"the pack with no rows is not the only one reported unabsorbed"
);
eprintln!(
"load {}; {PACKS} packs / {} object rows: derive-on-open diff {:.2} ms clean \
(early exit), {:.2} ms with one pack un-absorbed (full tail scan)",
loadavg(),
tail.len(),
clean.as_secs_f64() * 1e3,
crashed.as_secs_f64() * 1e3,
);
}
#[test]
fn the_indexed_bit_is_a_bitset_over_dense_pack_ordinals() {
let ids = [0u64, 1, 63, 64, 65, 4095];
let mut s = PackState::default();
for id in ids {
s.note(id, (id * 4096 + 12, 4096));
}
assert_eq!(s.unabsorbed, ids.len());
assert!(
ids.iter().all(|&i| !s.is_absorbed(i)),
"a bit is up already"
);
s.mark_absorbed(64, (64 * 4096 + 12, 4096));
assert!(s.is_absorbed(64), "the bit for ordinal 64 did not go up");
for other in [0u64, 1, 63, 65, 4095] {
assert!(
!s.is_absorbed(other),
"setting ordinal 64 also set ordinal {other} — one bit spilled onto a neighbour"
);
}
assert_eq!(s.unabsorbed, ids.len() - 1);
assert_eq!(
s.pending().iter().map(|p| p.pack_id).collect::<Vec<_>>(),
vec![0, 1, 63, 65, 4095],
"the pending list is not the clear bits"
);
assert_eq!(
s.pending()[0],
PendingPack {
pack_id: 0,
offset: 12,
len: 4096,
resumed: true,
},
"a pending pack lost the extent it must be absorbed from"
);
s.mark_absorbed(64, (64 * 4096 + 12, 4096));
s.note(64, (64 * 4096 + 12, 4096));
assert!(s.is_absorbed(64));
assert_eq!(
s.unabsorbed,
ids.len() - 1,
"an absorbed pack was queued again"
);
assert_eq!(s.words.len(), 64);
assert_eq!(s.words.len() * 8, 512, "4096 packs is 512 bytes of bits");
}
#[test]
fn a_read_before_the_drain_is_served_from_the_pack_and_never_waits() {
let dir = tmpdir("drain-before");
let store = Arc::new(GitStore::open(&dir, "alice").unwrap());
let (pack, rows) = corpus_pack();
let (second, blob_oid) = one_blob_pack(b"a second push while the drain is parked");
let gate = store.hold_absorb_gate();
let tx = store.put_pack(&pack).unwrap();
let pack_id = tx.pack_id.unwrap();
assert_eq!(store.unindexed_packs(), 1, "the pack is not queued");
assert_eq!(store.index().len(), 0, "the index already holds rows");
assert!(!store.indexer().is_indexed(pack_id), "the bit is already up");
let t = Instant::now();
let tx2 = store.put_pack(&second).unwrap();
assert_eq!(tx2.pack_id, Some(pack_id + 1));
assert!(t.elapsed() < Duration::from_secs(2), "the push waited on the parked drain");
let oid = rows[0].oid.clone();
let last = rows[rows.len() - 1].oid.clone();
let t = Instant::now();
assert!(store.has(&oid).unwrap(), "a read with the drain parked answered absent for a durable object");
assert!(store.has(&last).unwrap(), "the last object of the pack answered absent");
assert!(store.has(&blob_oid).unwrap(), "the second (still pending) pack's object answered absent");
let (kind, body) = store.content(&blob_oid).unwrap().expect("content from the pending pack");
assert_eq!(kind, GitObjectKind::Blob);
assert_eq!(body, b"a second push while the drain is parked");
let stored = store.get(&oid).unwrap().expect("get from the pending pack");
assert_eq!(stored.extent.0, rows[0].offset + tx.extent.unwrap().0);
assert_eq!(
store.size(&last).unwrap(),
Some(rows[rows.len() - 1].uncompressed_size),
"size from the pending pack"
);
let absent = vec![0xEEu8; 20];
assert!(!store.has(&absent).unwrap(), "an oid in no pack must still be absent");
let elapsed = t.elapsed();
assert!(elapsed < Duration::from_secs(20), "reads took {elapsed:?} with the drain parked");
assert_eq!(store.index().len(), 0, "a read absorbed the pack behind the parked gate");
assert!(!store.indexer().is_indexed(pack_id));
assert_eq!(store.unindexed_packs(), 2);
eprintln!("{} reads served from two pending packs with the drain parked, in {elapsed:?}", 6);
let err = store.reachable(&[oid.as_slice()], &[]).unwrap_err();
assert!(err.to_string().contains("index in progress"), "{err:#}");
drop(gate);
store.wait_indexed();
assert_eq!(store.unindexed_packs(), 0);
assert!(store.indexer().is_indexed(pack_id));
assert_eq!(store.index().len(), rows.len() + 1);
assert_eq!(store.absorb_pending().unwrap(), 0);
for r in rows.iter().take(256) {
assert!(
store.index().lookup(&r.oid).is_some(),
"{} is not in the rows after the drain",
hex::encode(&r.oid)
);
}
assert!(store.reachable(&[oid.as_slice()], &[]).is_ok(), "reachable after the drain");
}
#[test]
fn a_paused_indexer_blocks_no_push_and_no_read() {
let dir = tmpdir("paused-indexer");
let meminfo = dir.join("meminfo");
let write = |avail_kib: u64| {
std::fs::write(&meminfo, format!("MemTotal: 16000000 kB\nMemAvailable: {avail_kib} kB\n"))
.unwrap()
};
write(1_000_000);
let dog = Arc::new(crate::memguard::MemWatchdog::with(
&meminfo,
3_200_000 * 1024,
Duration::from_millis(10),
));
let store = Arc::new(GitStore::open(&dir, "alice").unwrap());
store.set_mem_watchdog(dog.clone());
let (pack, rows) = corpus_pack();
let (second, blob_oid) = one_blob_pack(b"pushed while the indexer is paused");
let t_push = Instant::now();
let tx = store.put_pack(&pack).unwrap();
let pack_id = tx.pack_id.unwrap();
let mut parked = false;
for _ in 0..200 {
if dog.is_paused() {
parked = true;
break;
}
std::thread::sleep(Duration::from_millis(10));
}
assert!(parked, "the worker never paused under a meminfo 1 GB < 3.2 GB");
assert!(!store.indexer().is_indexed(pack_id), "indexed while paused");
assert_eq!(store.index().len(), 0);
let tx2 = store.put_pack(&second).unwrap();
assert_eq!(tx2.pack_id, Some(pack_id + 1));
assert!(t_push.elapsed() < Duration::from_secs(5), "a push waited on the paused worker");
let t = Instant::now();
assert!(store.has(&rows[0].oid).unwrap());
assert!(store.has(&blob_oid).unwrap());
let (kind, body) = store.content(&blob_oid).unwrap().unwrap();
assert_eq!((kind, body.as_slice()), (GitObjectKind::Blob, &b"pushed while the indexer is paused"[..]));
assert_eq!(
store.extents(&[rows[0].oid.as_slice(), rows[1].oid.as_slice(), blob_oid.as_slice()])
.unwrap()
.iter()
.filter(|e| e.is_some())
.count(),
3,
"the batch path answered absent while paused"
);
assert!(t.elapsed() < Duration::from_secs(20), "reads waited on the paused worker");
let snap = store.index_progress();
assert_eq!(snap.root, store.root().display().to_string());
assert!(dog.is_paused(), "the worker resumed on its own — the fake still says 1 GB");
assert!(!store.indexer().is_indexed(pack_id), "indexed while still paused");
let reach = store.reachable(&[rows[0].oid.as_slice()], &[]).unwrap();
assert!(!reach.is_empty());
assert_eq!(store.unindexed_packs(), 0, "the reader did not absorb for itself");
assert!(dog.is_paused(), "the worker is still parked; the reader did the work");
assert_eq!(store.index().len(), rows.len() + 1);
write(9_000_000);
store.wait_indexed();
assert!(store.indexer().is_indexed(pack_id));
assert!(store.indexer().is_indexed(pack_id + 1));
assert_eq!(dog.pauses(), 1, "one pause, one resume");
assert!(!dog.is_paused());
}
#[test]
fn no_read_is_told_absent_while_the_drain_is_running() {
let dir = tmpdir("drain-during");
let store = GitStore::open(&dir, "alice").unwrap();
let (pack, rows) = corpus_pack();
let oids: Vec<&[u8]> = rows.iter().map(|r| r.oid.as_slice()).collect();
let tx = store.put_pack(&pack).unwrap();
let pack_id = tx.pack_id.unwrap();
let early = AtomicU64::new(0);
let absent = AtomicU64::new(0);
let reads = AtomicU64::new(0);
let workers = 4usize;
gatling_for_each(workers, workers, |w| loop {
let still_pending = store.unindexed_packs() > 0;
for oid in oids.iter().skip(w).step_by(workers * 8) {
if still_pending {
early.fetch_add(1, AtomicOrdering::Relaxed);
}
reads.fetch_add(1, AtomicOrdering::Relaxed);
if !store.has(oid).unwrap() {
absent.fetch_add(1, AtomicOrdering::Relaxed);
}
}
if store.indexer().is_indexed(pack_id) {
break;
}
});
let absent = absent.load(AtomicOrdering::Acquire);
assert_eq!(
absent,
0,
"{workers} reader(s) were told a durable object was absent — {absent} absent answers \
over {} objects",
rows.len()
);
assert_eq!(store.object_count(), rows.len());
eprintln!(
"load {}; {} reads across {workers} gatling workers, {} of them while the pack was \
still un-indexed",
loadavg(),
reads.load(AtomicOrdering::Acquire),
early.load(AtomicOrdering::Acquire),
);
}
#[test]
fn a_pack_is_absorbed_once_even_when_both_callers_race_for_it() {
let dir = tmpdir("drain-once");
let store = Arc::new(GitStore::open(&dir, "alice").unwrap());
let (pack, rows) = corpus_pack();
let gate = store.hold_absorb_gate();
store.put_pack(&pack).unwrap();
let s = store.clone();
let reader = Job::spawn(move || s.absorb_pending().unwrap());
std::thread::sleep(Duration::from_millis(100));
drop(gate);
let by_read = reader.join().unwrap();
store.wait_indexed();
assert_eq!(
store.commit_count(),
commits_in(&rows),
"the commit graph carries {} commits, the pack has {}",
store.commit_count(),
commits_in(&rows)
);
assert_eq!(store.object_count(), rows.len());
assert_eq!(store.unindexed_packs(), 0);
eprintln!(
"load {}; the read absorbed {by_read} pack(s), the drain absorbed the rest",
loadavg()
);
}
#[test]
fn a_push_produces_an_exploded_row_for_every_object_in_the_pack() {
let dir = tmpdir("exploded-every-object");
let store = GitStore::open(&dir, "alice").unwrap();
let (pack, rows) = corpus_pack();
let t = Instant::now();
store.put_pack(&pack).unwrap();
let ack = t.elapsed();
store.wait_indexed();
let drain = t.elapsed();
let distinct: HashSet<Vec<u8>> = rows.iter().map(|r| r.oid.clone()).collect();
let stats = store.exploded_stats();
let graph_rows = store.exploded_of_kind(GitObjectKind::Commit).unwrap().len();
assert_eq!(
store.exploded_eager_rows().unwrap(),
graph_rows as u64,
"the drain wrote {} eager rows and there are {} commits — the eager pool \
holds something other than the graph",
store.exploded_eager_rows().unwrap(),
graph_rows
);
assert!(
stats.rows > 0 && stats.rows < distinct.len() as u64,
"the eager pool holds {} of {} objects; lazy explode means neither 0 \
(the graph would be blind) nor all of them (that is the 22.4x tier)",
stats.rows,
distinct.len()
);
let hash = store.hash_kind();
let mut checked = 0usize;
for oid in &distinct {
let (kind, payload) = store
.content(oid)
.unwrap()
.unwrap_or_else(|| panic!("{} has no exploded row", hex::encode(oid)));
let rehashed = hash.oid_of(&crate::object::canonical(kind, &payload));
assert_eq!(
&rehashed,
oid,
"the exploded row filed under {} contains an object that hashes to {}",
hex::encode(oid),
hex::encode(&rehashed)
);
checked += 1;
}
assert_eq!(checked, distinct.len());
let after_reads = store.exploded_stats();
assert!(
after_reads.rederived > 0,
"no read re-derived anything, so either every object is eager or the \
lazy path never ran"
);
assert_eq!(
after_reads.rederived + after_reads.served,
checked as u64,
"reads {} = served {} + rederived {} must account for every object",
checked,
after_reads.served,
after_reads.rederived
);
let inflated: u64 = rows.iter().map(|r| r.uncompressed_size).sum();
let table = std::fs::metadata(store.exploded_path())
.map(|m| m.len())
.unwrap_or(0);
eprintln!(
"load {}; {} pack entries → {} exploded rows, all {} re-hashed to their own oid; \
ack {:.1} ms, drain {:.1} ms, {:.0} rows/s; disk: pack {:.1} MiB verbatim, objects \
inflate to {:.1} MiB, table file {:.1} MiB ({:.0}x the pack, {:.3}x the payload)",
loadavg(),
rows.len(),
stats.rows,
checked,
ack.as_secs_f64() * 1e3,
drain.as_secs_f64() * 1e3,
stats.rows as f64 / drain.as_secs_f64(),
pack.len() as f64 / (1 << 20) as f64,
inflated as f64 / (1 << 20) as f64,
table as f64 / (1 << 20) as f64,
table as f64 / pack.len() as f64,
table as f64 / inflated.max(1) as f64,
);
}
#[test]
#[ignore = "needs a pack: ZNIPPY_EXPLODE_BENCH_PACK=<path> cargo test … -- --ignored --nocapture"]
fn the_table_costs_what_it_holds() {
let Ok(path) = std::env::var("ZNIPPY_EXPLODE_BENCH_PACK") else {
eprintln!("no ZNIPPY_EXPLODE_BENCH_PACK named; nothing measured");
return;
};
let pack = std::fs::read(&path).unwrap_or_else(|e| panic!("reading {path}: {e}"));
let dir = tmpdir("exploded-cost");
let store = GitStore::open(&dir, "alice").unwrap();
let t = Instant::now();
store.put_pack(&pack).unwrap();
let ack = t.elapsed();
store.wait_indexed();
let drain = t.elapsed();
let stats = store.exploded_stats();
let table = std::fs::metadata(store.exploded_path())
.map(|m| m.len())
.unwrap_or(0);
let mut held = 0u64;
for kind in [
GitObjectKind::Commit,
GitObjectKind::Tree,
GitObjectKind::Blob,
GitObjectKind::Tag,
] {
held += store
.exploded_of_kind(kind)
.unwrap()
.iter()
.map(|(_, p)| p.len() as u64)
.sum::<u64>();
}
let mib = |n: u64| n as f64 / (1 << 20) as f64;
eprintln!(
"load {}; pack {} — {:.1} MiB verbatim → {} rows holding {:.1} MiB, table file \
{:.1} MiB = {:.3}x the payload and {:.1}x the pack; ack {:.0} ms, drain {:.0} ms",
loadavg(),
path,
mib(pack.len() as u64),
stats.rows,
mib(held),
mib(table),
table as f64 / held.max(1) as f64,
table as f64 / pack.len() as f64,
ack.as_secs_f64() * 1e3,
drain.as_secs_f64() * 1e3,
);
}
const CLEAN_DIR: &str = "GUNNAR_CLEAN_SHUTDOWN_DIR";
#[test]
#[ignore = "spawned by a_clean_shutdown_and_reopen_keeps_the_graph_and_reachable"]
fn the_child_that_pushes_and_shuts_down_cleanly() {
let Ok(dir) = std::env::var(CLEAN_DIR) else {
return;
};
let dir = PathBuf::from(dir);
let pack = std::fs::read(dir.join("fixture.pack")).expect("the parent's fixture");
let store = GitStore::open(&dir, "alice").expect("open");
store.put_pack(&pack).expect("the ack path");
store.wait_indexed();
assert_eq!(store.unindexed_packs(), 0, "the child shut down mid-index");
let tip = store
.graph_snapshot()
.into_iter()
.max_by_key(|c| c.generation)
.expect("the child built no graph at all");
let tip_raw = hex::decode(&tip.oid).unwrap();
let closure = store.reachable(&[&tip_raw], &[]).unwrap();
assert!(!closure.is_empty(), "the child's own reachable() was empty");
std::fs::write(dir.join("tip"), &tip.oid).expect("marker");
std::fs::write(dir.join("closure"), closure.len().to_string()).expect("marker");
std::fs::write(dir.join("commits"), store.commit_count().to_string()).expect("marker");
std::fs::write(dir.join("objects"), store.object_count().to_string()).expect("marker");
drop(store);
std::fs::write(dir.join("clean"), b"closed").expect("marker");
}
#[test]
fn an_annotated_tag_reaches_the_commit_it_points_at() {
let dir = tmpdir("tag-peel");
let store = GitStore::open(&dir, "alice").unwrap();
let (pack, rows) = corpus_pack();
store.put(&pack, &[]).unwrap();
store.wait_indexed();
let commit = rows
.iter()
.find(|r| r.kind == GitObjectKind::Commit)
.expect("the corpus has commits");
let commit_hex = hex::encode(&commit.oid);
let commit_closure = store.reachable(&[&commit.oid], &[]).unwrap();
assert!(
commit_closure.len() > 1,
"the fixture's commit closes over nothing, so this test could not tell a peel from a no-op"
);
let (tag_pack, tag_oid) = crate::store::tests::one_tag_pack(&commit_hex, "v1.0");
store.put(&tag_pack, &[]).unwrap();
store.wait_indexed();
let closure = store.reachable(&[&tag_oid], &[]).unwrap();
assert!(
closure.contains(&tag_oid),
"the tag itself is not in its own closure"
);
assert!(
closure.contains(&commit.oid),
"the tag's closure is {} object(s), so it carries the tag and nothing it points at — the commit {} it names is missing and a mirror built from this answer cannot be read",
closure.len(),
&commit_hex[..12]
);
for oid in &commit_closure {
assert!(
closure.contains(oid),
"the tag's closure is missing {} which the commit it points at reaches",
hex::encode(oid)
);
}
}
#[test]
fn a_clean_shutdown_and_reopen_keeps_the_graph_and_reachable() {
let dir = tmpdir("clean-shutdown-graph");
let (pack, rows) = corpus_pack();
std::fs::write(dir.join("fixture.pack"), &pack).unwrap();
let status = std::process::Command::new(std::env::current_exe().unwrap())
.args([
"--exact",
"--ignored",
"--nocapture",
"git_ops::tests::the_child_that_pushes_and_shuts_down_cleanly",
])
.env(CLEAN_DIR, &dir)
.status()
.expect("spawn the child that shuts down cleanly");
assert!(
status.success(),
"the child did not exit cleanly — it {status:?}, so this is a crash guard and not a \
clean-shutdown one"
);
assert!(
dir.join("clean").exists(),
"the child never reached its clean shutdown"
);
let tip_hex = std::fs::read_to_string(dir.join("tip")).unwrap();
let closure_before: usize = std::fs::read_to_string(dir.join("closure"))
.unwrap()
.parse()
.unwrap();
let commits_before: usize = std::fs::read_to_string(dir.join("commits"))
.unwrap()
.parse()
.unwrap();
assert_eq!(
commits_before,
commits_in(&rows),
"the child's own graph was already wrong, so the reopen proves nothing"
);
let store = GitStore::open(&dir, "alice").unwrap();
assert_eq!(
store.commit_count(),
commits_in(&rows),
"the commit graph did not survive a clean shutdown: {} of {} commits — the exploded \
table is a warm cache, not a durable table",
store.commit_count(),
commits_in(&rows)
);
let tip_raw = hex::decode(&tip_hex).unwrap();
let closure = store.reachable(&[&tip_raw], &[]).unwrap();
assert_eq!(
closure.len(),
closure_before,
"reachable() answers differently after a clean reopen: {} objects against the {} the \
same tip closed over before the shutdown",
closure.len(),
closure_before
);
assert!(closure.contains(&tip_raw));
store.wait_indexed();
assert!(
!store.indexer().is_indexed(0),
"the reopen re-queued the pack, so the graph above is a recovery artefact rather \
than the table's"
);
assert_eq!(store.unindexed_packs(), 0);
assert_eq!(
store.absorb_pending().unwrap(),
0,
"the reopen left index work"
);
assert_eq!(store.object_count(), rows.len());
assert_eq!(
store.exploded_stats().written,
0,
"this process wrote exploded rows, so the table was rebuilt rather than read"
);
eprintln!(
"load {}; clean shutdown → reopen: {} rows, {} commits, tip closure {} objects \
(rebuilt nothing)",
loadavg(),
store.object_count(),
store.commit_count(),
closure.len(),
);
}
#[test]
fn a_reopen_reads_the_persisted_graph_it_does_not_recompute() {
let dir = tmpdir("persisted-graph");
let (pack, rows) = corpus_pack();
let tip_hex;
{
let store = GitStore::open(&dir, "alice").unwrap();
store.put_pack(&pack).unwrap();
store.wait_indexed();
tip_hex = store
.graph_snapshot()
.into_iter()
.max_by_key(|c| c.generation)
.expect("a graph")
.oid;
assert!(
dir.join("objects.graph").exists(),
"an absorb never persisted the graph, so a reopen has nothing to read"
);
}
let tip_raw = hex::decode(&tip_hex).unwrap();
std::fs::remove_file(dir.join("objects.graph")).unwrap();
let (recomputed_graph, recomputed_closure) = {
let store = GitStore::open(&dir, "alice").unwrap();
assert_eq!(
store.graph_source_stats(),
(0, 1),
"with no objects.graph the reopen should have recomputed the graph"
);
assert_eq!(store.commit_count(), commits_in(&rows));
let g = store.graph_snapshot();
let c = store.reachable(&[&tip_raw], &[]).unwrap();
assert!(!c.is_empty());
(g, c)
};
let store = GitStore::open(&dir, "alice").unwrap();
assert_eq!(
store.graph_source_stats(),
(1, 0),
"the reopen recomputed the graph instead of reading objects.graph"
);
assert_eq!(
store.graph_snapshot(),
recomputed_graph,
"the graph READ from objects.graph differs from the one refold RECOMPUTES"
);
assert_eq!(
store.reachable(&[&tip_raw], &[]).unwrap(),
recomputed_closure,
"reachable() over the persisted graph differs from reachable() over the recomputed one"
);
assert_eq!(store.commit_count(), commits_in(&rows));
}
#[test]
fn with_the_graph_cache_off_no_objects_graph_is_written_and_the_graph_is_the_same() {
let on_dir = tmpdir("graph-cache-on");
let off_dir = tmpdir("graph-cache-off");
let (pack, rows) = corpus_pack();
let on = GitStore::open(&on_dir, "alice").unwrap();
on.put_pack(&pack).unwrap();
on.wait_indexed();
let off = GitStore::open(&off_dir, "alice").unwrap();
off.absorber.graph_cache.store(false, std::sync::atomic::Ordering::Relaxed);
std::fs::remove_file(off_dir.join("objects.graph")).ok();
off.put_pack(&pack).unwrap();
off.wait_indexed();
assert!(on_dir.join("objects.graph").exists(), "the default still persists the graph");
assert!(
!off_dir.join("objects.graph").exists(),
"the graph cache is off and objects.graph was written anyway"
);
assert_eq!(off.graph_source_stats().0, 0, "a store with the cache off never reads one");
assert_eq!(on.graph_snapshot(), off.graph_snapshot());
assert_eq!(off.commit_count(), commits_in(&rows));
}
#[test]
fn a_graph_whose_token_does_not_match_is_rejected() {
let dir = tmpdir("wrong-token-graph");
let (pack, _rows) = corpus_pack();
{
let store = GitStore::open(&dir, "alice").unwrap();
store.put_pack(&pack).unwrap();
store.wait_indexed();
}
let path = dir.join("objects.graph");
let mut bytes = std::fs::read(&path).unwrap();
assert!(bytes.len() > 37 && &bytes[..4] == b"GRPH");
bytes[6] ^= 0xff;
std::fs::write(&path, &bytes).unwrap();
let store = GitStore::open(&dir, "alice").unwrap();
assert_eq!(
store.graph_source_stats(),
(0, 1),
"a cache with the wrong token was trusted instead of being recomputed"
);
}
#[test]
fn the_ordinal_space_comes_off_the_index_not_a_scan() {
let dir = tmpdir("ordinal-off-the-index");
let (pack, rows) = corpus_pack();
let tip_hex;
let (first_oids, first_closure, first_objects, first_commits);
{
let store = GitStore::open(&dir, "alice").unwrap();
store.put_pack(&pack).unwrap();
store.wait_indexed();
tip_hex = store
.graph_snapshot()
.into_iter()
.max_by_key(|c| c.generation)
.expect("a graph")
.oid;
let tip_raw = hex::decode(&tip_hex).unwrap();
let n = store.object_count() as u64;
let (folded, tail) = store.ordinal_source_stats();
assert_eq!(folded + tail, n, "every oid is accounted to the index or the tail");
let oids = store.ordinal_oids();
let scanned: Vec<String> =
store.index().oids_in_order().unwrap().iter().map(hex::encode).collect();
assert_eq!(oids, scanned, "the ordinal space is not the repository in oid order");
assert!(
!dir.join("objects.ordinal").exists(),
"no memo may be written: the index's oid column is the ordinal space"
);
store.index().rebuild().unwrap();
first_closure = store.reachable(&[&tip_raw], &[]).unwrap();
assert!(!first_closure.is_empty());
first_oids = oids;
first_objects = store.object_count();
first_commits = store.commit_count();
}
let tip_raw = hex::decode(&tip_hex).unwrap();
let store = GitStore::open(&dir, "alice").unwrap();
let n = store.object_count() as u64;
assert_eq!(
store.ordinal_source_stats(),
(n, 0),
"after a fold the warm reopen's ordinal space must come entirely off the index"
);
assert_eq!(
store.ordinal_oids(),
first_oids,
"the ordinal space after the warm reopen is not the tail in oid order"
);
assert_eq!(
store.reachable(&[&tip_raw], &[]).unwrap(),
first_closure,
"reachable() over the index's ordinal space differs from the first session's"
);
assert_eq!(store.object_count(), first_objects);
assert_eq!(store.commit_count(), first_commits);
assert_eq!(store.commit_count(), commits_in(&rows));
assert!(!dir.join("objects.ordinal").exists(), "still no memo after the reopen");
}
#[test]
fn a_stale_objects_ordinal_from_an_older_binary_is_removed_at_open() {
let dir = tmpdir("stale-ordinal");
let (pack, _rows) = corpus_pack();
{
let store = GitStore::open(&dir, "alice").unwrap();
store.put_pack(&pack).unwrap();
store.wait_indexed();
}
let path = dir.join("objects.ordinal");
let mut bytes = b"ORDL".to_vec();
bytes.push(1);
bytes.push(20);
bytes.extend_from_slice(&7u64.to_le_bytes());
bytes.extend_from_slice(&1u64.to_le_bytes());
bytes.extend_from_slice(&[0xabu8; 20]);
std::fs::write(&path, &bytes).unwrap();
assert!(path.exists(), "precondition");
let store = GitStore::open(&dir, "alice").unwrap();
assert!(!path.exists(), "a stale objects.ordinal must be removed at open");
assert!(store.object_count() > 0, "and the store still answers");
}
#[test]
fn a_content_read_is_served_from_the_table_and_not_re_derived() {
const READS: usize = 512;
let dir = tmpdir("exploded-served");
let (pack, rows) = corpus_pack();
let oids: Vec<Vec<u8>> = rows
.iter()
.map(|r| r.oid.clone())
.collect::<HashSet<Vec<u8>>>()
.into_iter()
.take(READS)
.collect();
{
let store = GitStore::open(&dir, "alice").unwrap();
store.put_pack(&pack).unwrap();
store.wait_indexed();
let before = store.exploded_stats();
let t = Instant::now();
let mut first: Vec<(GitObjectKind, Vec<u8>)> = Vec::with_capacity(oids.len());
for oid in &oids {
first.push(
store
.content(oid)
.unwrap()
.expect("the store lost an object"),
);
}
let cold = t.elapsed();
let after = store.exploded_stats();
let rederived = after.rederived - before.rederived;
assert_eq!(
rederived + (after.served - before.served),
oids.len() as u64,
"{} reads = served {} + rederived {} must account for every one",
oids.len(),
after.served - before.served,
rederived
);
assert!(
rederived > 0,
"not one read re-derived: either every object is eager or the lazy \
path never ran, and this test then proves nothing about either"
);
store.explode_cache_maintain().unwrap();
let declined = store.explode_cache_use().direct_reads;
assert!(
declined > 0,
"not one of {} reads was declined as cheap; either every object in \
this corpus is expensive or the admission policy is not running",
oids.len()
);
let mid = store.exploded_stats();
let t = Instant::now();
for (i, oid) in oids.iter().enumerate() {
let again = store.content(oid).unwrap().expect("the store lost an object");
assert_eq!(again, first[i], "the second read of an object disagreed with the first");
}
let warm = t.elapsed();
let end = store.exploded_stats();
assert_eq!(
end.rederived - mid.rederived,
declined,
"the second pass re-derived {} objects and the cache declined {declined} — \
every EXPENSIVE object must come out of the table the second time, and \
every cheap one must be read directly again",
end.rederived - mid.rederived
);
assert_eq!(
end.served - mid.served,
oids.len() as u64 - declined,
"the expensive objects did not come out of the table on the second pass"
);
eprintln!(
"load {}; {} content reads: FIRST {:.1} ms ({:.1} µs/read, {} re-derived from \
the pack, {declined} of them too cheap to cache) → SECOND {:.1} ms \
({:.1} µs/read) = {:.1}x",
loadavg(),
oids.len(),
cold.as_secs_f64() * 1e3,
cold.as_secs_f64() * 1e6 / oids.len() as f64,
rederived,
warm.as_secs_f64() * 1e3,
warm.as_secs_f64() * 1e6 / oids.len() as f64,
cold.as_secs_f64() / warm.as_secs_f64().max(f64::MIN_POSITIVE),
);
}
std::fs::remove_file(dir.join("objects.exploded")).unwrap();
for e in std::fs::read_dir(&dir).unwrap().flatten() {
if e.file_name().to_string_lossy().starts_with("objects.exploded.c") {
std::fs::remove_file(e.path()).unwrap();
}
}
let store = GitStore::open(&dir, "alice").unwrap();
let gate = store.hold_absorb_gate();
let before = store.exploded_stats();
let mut rederived = 0u64;
for oid in oids.iter().take(8) {
assert!(
store.absorber.resolved(oid).unwrap().is_some(),
"the verbatim truth could not re-derive {}",
hex::encode(oid)
);
rederived += 1;
}
let after = store.exploded_stats();
let drain_won = store.unindexed_packs() == 0;
drop(gate);
if drain_won {
eprintln!("the drain rebuilt the table before the gate was held; the fallback was not observed this run");
return;
}
assert_eq!(
after.served - before.served,
0,
"a dropped table still reported {} reads served",
after.served - before.served
);
assert_eq!(
after.rederived - before.rederived,
rederived,
"the fallback ran but was not counted"
);
}
#[test]
fn every_read_path_answers_a_just_acked_push_with_the_indexer_paused() {
use crate::serve::GitServe as _;
let dir = tmpdir("read-after-ack");
let store = Arc::new(GitStore::open(&dir, "alice").unwrap());
let (pack, rows) = corpus_pack();
store.put_pack(&pack).unwrap();
let gate = store.hold_absorb_gate();
assert!(
store.unindexed_packs() > 0,
"the drain absorbed the pack before the gate was taken, so this run \
read an INDEXED store and proves nothing about the window"
);
let distinct: Vec<Vec<u8>> = rows
.iter()
.map(|r| r.oid.clone())
.collect::<HashSet<Vec<u8>>>()
.into_iter()
.collect();
let hash = store.hash_kind();
let n = distinct.len();
let (tx, rx) = std::sync::mpsc::channel::<Result<(usize, usize, usize)>>();
let reader = {
let store = store.clone();
let distinct = distinct.clone();
std::thread::spawn(move || {
let mut had = 0usize;
let mut headed = 0usize;
let mut readback = 0usize;
let r = (|| -> Result<()> {
for oid in &distinct {
if store.has(oid)? {
had += 1;
}
if crate::serve::GitServe::header(&*store, oid)?.is_some() {
headed += 1;
}
let Some((kind, payload)) = crate::serve::GitServe::read(&*store, oid)? else {
continue;
};
let k = match kind {
crate::index_layout::ObjType::Commit => GitObjectKind::Commit,
crate::index_layout::ObjType::Tree => GitObjectKind::Tree,
crate::index_layout::ObjType::Blob => GitObjectKind::Blob,
crate::index_layout::ObjType::Tag => GitObjectKind::Tag,
other => anyhow::bail!("`read` answered with a {other:?}"),
};
let rehashed = hash.oid_of(&crate::object::canonical(k, &payload));
if &rehashed == oid {
readback += 1;
}
}
Ok(())
})();
let _ = tx.send(r.map(|_| (had, headed, readback)));
})
};
let out = rx
.recv_timeout(Duration::from_secs(20))
.expect(
"a read path blocked on the paused indexer for 20 s: 'nobody waits' is the \
design note's rule and a read that waits for the drain is a bug by definition",
);
let (had, headed, readback) = out.expect("a read path errored inside the index window");
drop(gate);
let _ = reader.join();
assert_eq!(
(had, headed, readback),
(n, n, n),
"the store answered for {had}/{n} `has`, {headed}/{n} `header` and {readback}/{n} \
byte-correct `read`s of a pack it had already ACKED — an object the client was \
told had landed must never read as absent"
);
}
#[test]
fn an_account_at_its_budget_serves_every_object_only_slower() {
let dir = tmpdir("explode-budget");
let store = GitStore::open(&dir, "alice").unwrap();
let (pack, rows) = corpus_pack();
store.put_pack(&pack).unwrap();
store.wait_indexed();
let oids: Vec<Vec<u8>> = rows
.iter()
.map(|r| r.oid.clone())
.collect::<HashSet<Vec<u8>>>()
.into_iter()
.collect();
let mut truth: Vec<(GitObjectKind, Vec<u8>)> = Vec::with_capacity(oids.len());
for oid in &oids {
truth.push(store.content(oid).unwrap().expect("the store lost an object"));
}
store.explode_cache_maintain().unwrap();
store.set_explode_budget(64 << 10);
let before = store.exploded_stats();
for _ in 0..3 {
for (i, oid) in oids.iter().enumerate() {
let got = store
.content(oid)
.unwrap()
.expect("an object went absent when the budget was lowered");
assert_eq!(
got, truth[i],
"the store returned the wrong bytes for an object under a squeezed budget"
);
}
store.explode_cache_maintain().unwrap();
}
let after = store.exploded_stats();
let band = store.explode_cache_use();
assert!(
band.evicted_bytes > 0 || band.lazy_bytes <= 64 << 10,
"the budget was lowered to 64 KiB and the lazy pool holds {} B with nothing \\
evicted — the quota is not enforced, so this run proves nothing about reading \\
under one: {band:?}",
band.lazy_bytes
);
assert!(
after.rederived > before.rederived,
"not one read re-derived from the pack under a 64 KiB budget, so the cache was \\
never squeezed and 'only slower' was never tested"
);
eprintln!(
"load {}; {} objects read 3x under a 64 KiB budget: {} re-derived, {} served, \
{} evicted bytes over {} segments; every byte identical",
loadavg(),
oids.len(),
after.rederived - before.rederived,
after.served - before.served,
band.evicted_bytes,
band.evicted_segments,
);
}
#[test]
fn a_thin_push_on_top_of_an_unindexed_one_is_accepted() {
let dir = tmpdir("thin-on-unindexed");
let store = GitStore::open(&dir, "alice").unwrap();
let (pack, rows) = corpus_pack();
store.put_pack(&pack).unwrap();
let gate = store.hold_absorb_gate();
assert!(
store.unindexed_packs() > 0,
"the drain absorbed the first pack before the gate was taken; this run \
pushed onto an INDEXED store and proves nothing"
);
let base = rows
.iter()
.find(|r| r.kind == GitObjectKind::Blob && r.uncompressed_size > 0)
.expect("the corpus has a blob");
let thin = crate::pack_fixture::thin_ref_delta_pack(
&base.oid,
base.uncompressed_size,
b"lane Q was here\n",
);
let tx = store.put_pack(&thin).unwrap_or_else(|e| {
panic!("the thin push was refused: {e:#}");
});
assert!(tx.pack_id.is_some(), "the thin pack was not stored");
let hash = store.hash_kind();
let oid = hash.oid_of(&crate::object::canonical(
GitObjectKind::Blob,
b"lane Q was here\n",
));
let got = store.content(&oid);
drop(gate);
let (kind, payload) = got
.expect("reading the thin pack's object errored inside the index window")
.expect("the thin pack's own object is absent after its push was acked");
assert_eq!(kind, GitObjectKind::Blob);
assert_eq!(payload, b"lane Q was here\n");
}
#[test]
fn dropping_the_exploded_table_and_reopening_still_answers() {
let dir = tmpdir("exploded-dropped");
let (pack, rows) = corpus_pack();
let tip_hex;
let closure_before;
{
let store = GitStore::open(&dir, "alice").unwrap();
store.put_pack(&pack).unwrap();
store.wait_indexed();
let tip = store
.graph_snapshot()
.into_iter()
.max_by_key(|c| c.generation)
.expect("a graph");
let tip_raw = hex::decode(&tip.oid).unwrap();
closure_before = store.reachable(&[&tip_raw], &[]).unwrap().len();
tip_hex = tip.oid;
}
let table = dir.join("objects.exploded");
assert!(table.exists(), "the store never built an exploded table");
std::fs::remove_file(&table).unwrap();
{
let fresh = crate::exploded_arrow::ExplodedArchive::open(&table).unwrap();
assert_eq!(fresh.rows().unwrap(), 0, "the drop did not drop anything");
}
let _ = std::fs::remove_file(&table);
let store = GitStore::open(&dir, "alice").unwrap();
let tip_raw = hex::decode(&tip_hex).unwrap();
let closure = store.reachable(&[&tip_raw], &[]).unwrap();
assert_eq!(
closure.len(),
closure_before,
"reachable() after dropping the table: {} objects, against {} before",
closure.len(),
closure_before
);
assert!(store.has(&rows[0].oid).unwrap());
store.wait_indexed();
let stats = store.exploded_stats();
let distinct: HashSet<Vec<u8>> = rows.iter().map(|r| r.oid.clone()).collect();
let graph_rows = store.exploded_of_kind(GitObjectKind::Commit).unwrap().len();
let eager = store.exploded_eager_rows().unwrap();
assert_eq!(
eager, graph_rows as u64,
"the dropped table came back holding {eager} rows against a {}-object graph",
graph_rows
);
assert!(
eager > 0 && eager < distinct.len() as u64,
"the rebuilt pool holds {eager} of {} objects — neither 0 nor all of them is \
the lane Q shape",
distinct.len()
);
assert!(
stats.written > 0,
"the table came back without this process writing a row — it was never dropped"
);
assert_eq!(
store.commit_count(),
commits_in(&rows),
"the graph did not come back with the table: {} of {}",
store.commit_count(),
commits_in(&rows)
);
assert_eq!(
store.object_count(),
rows.len(),
"the rebuild lost index rows"
);
eprintln!(
"load {}; dropped and rebuilt: {} rows written, {} commits, tip closure {} objects",
loadavg(),
stats.written,
store.commit_count(),
closure.len(),
);
}
fn linear_history_pack_from(
n: usize,
start_parent: Option<&[u8]>,
) -> (Vec<u8>, Vec<Vec<u8>>) {
use crate::object::canonical;
use crate::pack_fixture::{build_pack, Item};
let hash = GitHashKind::Sha1;
let mut items: Vec<Item> = Vec::with_capacity(n * 3);
let mut commits: Vec<Vec<u8>> = Vec::with_capacity(n);
let mut prev: Option<Vec<u8>> = start_parent.map(|p| p.to_vec());
for c in 0..n {
let body = format!("blob for commit {c} of {n}, parent {prev:?}\n").into_bytes();
let blob_oid = hash.oid_of(&canonical(GitObjectKind::Blob, &body));
items.push(Item::Whole(GitObjectKind::Blob, body));
let mut tpayload = Vec::new();
tpayload.extend_from_slice(b"100644 f.bin\0");
tpayload.extend_from_slice(&blob_oid);
let tree_oid = hash.oid_of(&canonical(GitObjectKind::Tree, &tpayload));
items.push(Item::Whole(GitObjectKind::Tree, tpayload));
let mut cbody = format!("tree {}\n", hex::encode(&tree_oid));
if let Some(parent) = &prev {
cbody.push_str(&format!("parent {}\n", hex::encode(parent)));
}
cbody.push_str(
"author A U Thor <a@example.com> 1700000000 +0000\n\
committer A U Thor <a@example.com> 1700000000 +0000\n\n\
commit\n",
);
let cbody = cbody.into_bytes();
let commit_oid = hash.oid_of(&canonical(GitObjectKind::Commit, &cbody));
items.push(Item::Whole(GitObjectKind::Commit, cbody));
prev = Some(commit_oid.clone());
commits.push(commit_oid);
}
(build_pack(&items), commits)
}
#[test]
#[ignore = "perf measurement; run with --release --ignored --nocapture"]
fn bench_reachable_negotiation() {
for &n in &[48usize, 2_000, 20_000] {
let dir = tmpdir(&format!("bench-reach-{n}"));
let (pack, commits) = linear_history_pack_from(n, None);
let store = GitStore::open(&dir, "alice").unwrap();
store.put_pack(&pack).unwrap();
store.wait_indexed();
let root = commits.first().unwrap().clone();
let tip = commits.last().unwrap().clone();
let _ = store.reachable(&[&tip], &[]).unwrap();
let reps = 9usize;
let (mut root_t, mut tip_t) = (Vec::new(), Vec::new());
for _ in 0..reps {
let t = Instant::now();
let _ = store.reachable(&[&root], &[]).unwrap();
root_t.push(t.elapsed());
let t = Instant::now();
let _ = store.reachable(&[&tip], &[]).unwrap();
tip_t.push(t.elapsed());
}
root_t.sort();
tip_t.sort();
let tip_closure = store.reachable(&[&tip], &[]).unwrap().len();
eprintln!(
"load {}; n={n}: reachable(root, closure~1) med {:?} [{:?}..{:?}] | \
reachable(tip, closure {}) med {:?} [{:?}..{:?}]",
loadavg(),
root_t[reps / 2],
root_t[0],
root_t[reps - 1],
tip_closure,
tip_t[reps / 2],
tip_t[0],
tip_t[reps - 1],
);
}
}
#[test]
fn reachable_memo_matches_a_cold_fold_across_a_push() {
let dir = tmpdir("reach-memo-across-push");
let (pack_a, commits_a) = linear_history_pack_from(40, None);
let store = GitStore::open(&dir, "alice").unwrap();
store.put_pack(&pack_a).unwrap();
store.wait_indexed();
let tip_a = commits_a.last().unwrap().clone();
let warm_a = store.reachable(&[&tip_a], &[]).unwrap();
assert!(
warm_a.len() > 10,
"A's closure is too small to tell a stale memo from a coincidence: {}",
warm_a.len()
);
let (pack_b, commits_b) = linear_history_pack_from(40, Some(&tip_a));
store.put_pack(&pack_b).unwrap();
store.wait_indexed();
let tip_b = commits_b.last().unwrap().clone();
let warm_b_full = store.reachable(&[&tip_b], &[]).unwrap();
let warm_b_minus_a = store.reachable(&[&tip_b], &[&tip_a]).unwrap();
assert!(
warm_b_full.len() > warm_a.len(),
"the push did not grow the closure ({} !> {}), so the memo could be \
stale and this test would not know",
warm_b_full.len(),
warm_a.len()
);
drop(store);
let cold = GitStore::open(&dir, "alice").unwrap();
cold.wait_indexed();
let cold_b_full = cold.reachable(&[&tip_b], &[]).unwrap();
let cold_b_minus_a = cold.reachable(&[&tip_b], &[&tip_a]).unwrap();
assert_eq!(
warm_b_full, cold_b_full,
"warm memo != cold fold for want=tip_b (a memo that survived the push)"
);
assert_eq!(
warm_b_minus_a, cold_b_minus_a,
"warm memo != cold fold for want=tip_b have=tip_a"
);
let cold_a = cold.reachable(&[&tip_a], &[]).unwrap();
assert_eq!(
warm_a, cold_a,
"tip_a's closure changed across the push"
);
eprintln!(
"load {}; A closure {}, B closure {}, B−A {}",
loadavg(),
warm_a.len(),
warm_b_full.len(),
warm_b_minus_a.len(),
);
}
#[test]
fn a_push_carries_the_reach_table_across_the_fold() {
let dir = tmpdir("reach-carry-across-push");
let (pack_a, commits_a) = linear_history_pack_from(40, None);
let store = GitStore::open(&dir, "alice").unwrap();
store.put_pack(&pack_a).unwrap();
store.wait_indexed();
let tip_a = commits_a.last().unwrap().clone();
let warm_a = store.reachable(&[&tip_a], &[]).unwrap();
assert_eq!(store.reach_source_stats(), (0, 1), "the first read builds the table once");
let (pack_b, commits_b) = linear_history_pack_from(30, Some(&tip_a));
store.put_pack(&pack_b).unwrap();
store.wait_indexed();
let tip_b = commits_b.last().unwrap().clone();
let warm_b = store.reachable(&[&tip_b], &[]).unwrap();
let warm_b_minus_a = store.reachable(&[&tip_b], &[&tip_a]).unwrap();
assert_eq!(
store.reach_source_stats(),
(1, 1),
"the fold carried the table; no read rebuilt it"
);
assert!(warm_b.len() > warm_a.len(), "the push grew the closure");
assert_eq!(warm_b_minus_a.len(), warm_b.len() - warm_a.len(), "tip_b − tip_a is exactly the pushed objects");
assert_eq!(store.reachable(&[&tip_a], &[]).unwrap(), warm_a, "tip_a's closure survived the renumbering");
drop(store);
let cold = GitStore::open(&dir, "alice").unwrap();
cold.wait_indexed();
assert_eq!(cold.reachable(&[&tip_b], &[]).unwrap(), warm_b, "carried != rebuilt for want=tip_b");
assert_eq!(
cold.reachable(&[&tip_b], &[&tip_a]).unwrap(),
warm_b_minus_a,
"carried != rebuilt for want=tip_b have=tip_a"
);
assert_eq!(cold.reach_source_stats(), (1, 0), "the cold handle kept the persisted table and rebuilt nothing");
let (loaded, extended, _) = cold.reach_stats();
assert_eq!((loaded, extended), (1, 0), "loaded once, nothing left to extend");
}
#[test]
fn a_push_of_any_size_extends_the_reach_table_and_rebuilds_nothing() {
let dir = tmpdir("reach-extend-not-rebuild");
let (pack_a, commits_a) = linear_history_pack_from(20, None);
let store = GitStore::open(&dir, "alice").unwrap();
store.put_pack(&pack_a).unwrap();
store.wait_indexed();
let tip_a = commits_a.last().unwrap().clone();
let warm_a = store.reachable(&[&tip_a], &[]).unwrap();
assert_eq!(store.reach_source_stats(), (0, 1));
let (pack_b, commits_b) = linear_history_pack_from(100, Some(&tip_a));
store.put_pack(&pack_b).unwrap();
store.wait_indexed();
let tip_b = commits_b.last().unwrap().clone();
let warm_b = store.reachable(&[&tip_b], &[]).unwrap();
assert_eq!(store.reach_source_stats(), (1, 1), "the fold kept the table; no read rebuilt it");
let (_, extended, unpersisted) = store.reach_stats();
assert_eq!(extended, 1, "the fold extended the table for the pushed commits");
assert_eq!(unpersisted, 1, "one extension since the last write-back");
assert_eq!(warm_b.len(), warm_a.len() + 300, "100 commits × (commit, tree, blob)");
assert_eq!(store.reachable(&[&tip_a], &[]).unwrap(), warm_a, "tip_a's closure is untouched");
drop(store);
let cold = GitStore::open(&dir, "alice").unwrap();
cold.wait_indexed();
assert_eq!(cold.reachable(&[&tip_b], &[]).unwrap(), warm_b, "extended warm != cold");
let (loaded, extended, _) = cold.reach_stats();
assert_eq!((loaded, extended), (1, 0), "read back the table the drop persisted; nothing to extend");
assert_eq!(cold.reach_source_stats(), (1, 0));
}
#[test]
fn a_reopen_reads_the_persisted_reach_table_and_rebuilds_nothing() {
let dir = tmpdir("reach-persisted-reopen");
let (pack, commits) = linear_history_pack_from(12, None);
let store = GitStore::open(&dir, "alice").unwrap();
store.put_pack(&pack).unwrap();
store.wait_indexed();
let tip = commits.last().unwrap().clone();
let root = commits.first().unwrap().clone();
let warm = store.reachable(&[&tip], &[]).unwrap();
let warm_delta = store.reachable(&[&tip], &[&root]).unwrap();
assert_eq!(store.reach_source_stats(), (0, 1));
drop(store);
assert!(dir.join("objects.reach").is_file(), "the build persisted objects.reach");
let cold = GitStore::open(&dir, "alice").unwrap();
cold.wait_indexed();
assert_eq!(cold.reachable(&[&tip], &[]).unwrap(), warm);
assert_eq!(cold.reachable(&[&tip], &[&root]).unwrap(), warm_delta);
assert_eq!(cold.reach_source_stats(), (1, 0), "read back, nothing rebuilt");
}
#[test]
fn a_hostile_persisted_reach_table_is_rebuilt_not_trusted() {
let dir = tmpdir("reach-hostile-persisted");
let (pack, commits) = linear_history_pack_from(12, None);
let store = GitStore::open(&dir, "alice").unwrap();
store.put_pack(&pack).unwrap();
store.wait_indexed();
let tip = commits.last().unwrap().clone();
let warm = store.reachable(&[&tip], &[]).unwrap();
drop(store);
let path = dir.join("objects.reach");
let bytes = std::fs::read(&path).expect("the build persisted objects.reach");
for (name, hostile) in [
("truncated", bytes[..bytes.len() / 2].to_vec()),
("garbage", b"RECH\x01not a table at all, not even a header".to_vec()),
("bitflipped", {
let mut b = bytes.clone();
let at = b.len() - 7;
b[at] ^= 0xff;
b
}),
] {
std::fs::write(&path, &hostile).unwrap();
let cold = GitStore::open(&dir, "alice").unwrap();
cold.wait_indexed();
assert_eq!(cold.reachable(&[&tip], &[]).unwrap(), warm, "{name}: the answer must not change");
let (loaded, _, _) = cold.reach_stats();
let (_, rebuilt) = cold.reach_source_stats();
assert!(
loaded == 0 || rebuilt == 0,
"{name}: either refused and rebuilt (loaded 0) or decoded exactly (rebuilt 0); got loaded {loaded} rebuilt {rebuilt}"
);
drop(cold);
}
}
#[test]
fn a_push_past_the_disk_line_evicts_the_oldest_cache_without_a_read() {
let dir = tmpdir("push-disk-line");
let store = GitStore::open(&dir, "rickard").unwrap();
let c = store.absorber.exploded.clone();
c.set_global_line(0);
c.set_segment_bytes_for_test(256 << 10);
c.set_disk_reading_for_test(10);
let mut x: u32 = 0x9E37_79B9;
let big: Vec<u8> = (0..64 << 10)
.map(|_| {
x = x.wrapping_mul(1_664_525).wrapping_add(1_013_904_223);
(x >> 24) as u8
})
.collect();
let cost = crate::resolve::ReadCost {
depth: crate::exploded_cache::CHEAP_DEPTH + 8,
inflated: big.len() as u64,
external_base: false,
};
for i in 0..24u32 {
let mut oid = vec![0u8; 20];
oid[..4].copy_from_slice(&i.to_be_bytes());
c.write_back(&oid, GitObjectKind::Blob, &big, cost).unwrap();
c.maintain().unwrap();
}
let before = c.use_now();
assert!(before.segments >= 4, "precondition: rotated into segments, has {}", before.segments);
assert_eq!(before.evicted_segments, 0, "precondition: nothing evicted under a 10 % disk");
c.set_disk_reading_for_test(85);
let (pack, _) = one_blob_pack(b"pushed after the disk crossed its line\n");
store.put(&pack, &[]).unwrap();
store.wait_indexed();
let deadline = Instant::now() + Duration::from_secs(10);
while c.use_now().evicted_segments == 0 && Instant::now() < deadline {
std::thread::sleep(Duration::from_millis(20));
}
let after = c.use_now();
assert!(after.evicted_segments > 0, "the push did not make the cache give anything back");
assert_eq!(after.segments, 1, "past the line the pool keeps only its live segment");
}
}