use std::io::BufRead;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use anyhow::{Context, Result};
use git_storage_trait::GitOps;
use znippy_plugin_git::arms::StoreConfig;
use znippy_plugin_git::git_ops::{open_selected, SelectedStore};
use znippy_plugin_git::GitHashKind;
use super::store_cache::{self, Cache};
use super::{BeforeStore, MaintenanceReport, OpenRepo, PushReceipt};
pub fn store_root(git_dir: &Path) -> PathBuf {
git_dir.join("gunnar-git.znippy")
}
fn to_znippy_kind(kind: gix_hash::Kind) -> GitHashKind {
match kind {
gix_hash::Kind::Sha256 => GitHashKind::Sha256,
_ => GitHashKind::Sha1,
}
}
static STORES: Cache<SelectedStore> = Cache::new();
fn arms() -> Result<StoreConfig> {
use std::sync::OnceLock;
static ARMS: OnceLock<std::result::Result<StoreConfig, String>> = OnceLock::new();
ARMS.get_or_init(|| {
let resolved = StoreConfig::from_env()
.map(super::writer::select)
.map_err(|e| format!("{e:#}"));
match &resolved {
Ok(cfg) => tracing::info!(
target: "gunnar::store",
event = "store.znippy_env_selected",
config = %cfg,
"the znippy selector, resolved once per process from the environment"
),
Err(e) => tracing::error!(
target: "gunnar::store",
event = "store.znippy_env_refused",
error = %e,
"the znippy selector could not be resolved from the environment"
),
}
resolved
})
.clone()
.map_err(|e| anyhow::anyhow!(e))
}
pub fn shutdown() {
store_cache::clear(&STORES);
}
pub fn cancel_index(git_dir: &Path) {
if let Some(store) = store_cache::peek(&STORES, &store_root(git_dir)) {
store.cancel_index();
}
}
pub fn index_progress() -> Vec<(PathBuf, znippy_plugin_git::progress::Snapshot)> {
znippy_plugin_git::progress::snapshot_all(false)
.into_iter()
.filter_map(|s| {
let root = PathBuf::from(&s.root);
let git_dir = root.parent()?.to_path_buf();
Some((git_dir, s))
})
.collect()
}
#[must_use = "the suppression lasts only as long as the guard is held"]
pub fn validation_reads() -> znippy_plugin_git::SuppressWriteBack {
znippy_plugin_git::suppress_write_back()
}
pub fn explode_use() -> Vec<(PathBuf, crate::storage::ExplodeBand)> {
znippy_plugin_git::exploded_cache::use_all()
.into_iter()
.filter_map(|(root, u)| {
let git_dir = PathBuf::from(&root).parent()?.to_path_buf();
Some((
git_dir,
crate::storage::ExplodeBand {
eager_bytes: u.eager_bytes,
lazy_bytes: u.lazy_bytes,
budget: u.budget,
segments: u.segments,
evicted_segments: u.evicted_segments,
evicted_bytes: u.evicted_bytes,
written_back: u.written_back,
},
))
})
.collect()
}
pub fn open_object_counts() -> Vec<(PathBuf, u64)> {
store_cache::live(&STORES)
.into_iter()
.filter_map(|(root, store)| {
let git_dir = root.parent()?.to_path_buf();
Some((git_dir, store.object_count() as u64))
})
.collect()
}
pub fn set_explode_budget(git_dir: &Path, bytes: u64) -> usize {
znippy_plugin_git::exploded_cache::set_budget_under(
&store_root(git_dir).display().to_string(),
bytes,
)
}
pub fn drop_explode_cache(git_dir: &Path) -> u64 {
match store_cache::peek(&STORES, &store_root(git_dir)) {
Some(store) => store.drop_explode_cache().unwrap_or(0),
None => 0,
}
}
fn open_shared(git_dir: &Path, hash_kind: gix_hash::Kind) -> Result<Arc<SelectedStore>> {
let root = store_root(git_dir);
let account = git_dir.to_string_lossy().into_owned();
let make_in = root.clone();
store_cache::shared(&STORES, root, move || {
let store = open_selected(&make_in, &account, to_znippy_kind(hash_kind), arms()?)
.with_context(|| format!("opening the git store at {}", make_in.display()))?;
tracing::info!(
target: "gunnar::store",
event = "store.znippy_arms_selected",
writer = store.writer_name(),
index = store.index_name(),
durability = store.writer_durability(),
root = %make_in.display(),
"the archive writer and index layout this store was built on"
);
Ok(store)
})
}
pub fn open(git_dir: &Path, hash_kind: gix_hash::Kind) -> Result<Arc<OpenRepo>> {
Ok(OpenRepo::new(
git_dir,
hash_kind,
open_shared(git_dir, hash_kind)?,
))
}
pub fn ingest(
git_dir: &Path,
hash_kind: gix_hash::Kind,
pack: &mut dyn BufRead,
objects: super::ObjectSet,
progress: Option<&dyn git_storage_trait::PackProgress>,
before_store: Option<BeforeStore<'_>>,
) -> Result<PushReceipt> {
let started = std::time::Instant::now();
let store = open_shared(git_dir, hash_kind)?;
let pushed = super::read_pushed_pack_spooling(pack, hash_kind, Some(&store_root(git_dir)))?;
if let Some(hook) = before_store {
hook(&pushed.bytes).context("framing the pushed pack in the transaction log")?;
}
let tx = match progress {
Some(sink) => store.put_pack_observed(&pushed.bytes, sink),
None => store.put_pack(&pushed.bytes),
}
.context("storing the pushed pack verbatim")?;
let pack_id = super::pack_trailer(&pushed.bytes, hash_kind)?;
let want_objects = objects != super::ObjectSet::NotNeeded;
let new_oids = match (want_objects, tx.extent) {
(true, Some(extent)) => Some(
store
.oids_in_extent(extent)
.context("enumerating the objects this push introduced")?,
),
(true, None) => Some(Vec::new()),
(false, _) => None,
};
Ok(PushReceipt {
num_objects: pushed.num_objects,
version: pushed.version,
pack_id,
pack_bytes: pushed.bytes.len() as u64,
new_oids,
elapsed_ms: started.elapsed().as_millis() as u64,
})
}
pub fn maintain(git_dir: &Path, hash_kind: gix_hash::Kind) -> Result<MaintenanceReport> {
let store = open_shared(git_dir, hash_kind)?;
let sections = store
.seal()
.context("sealing the live logs into reserved sections")?;
let report = store.gc().context("compacting the archive")?;
let cache = store.drop_explode_cache().unwrap_or(0);
if cache > 0 {
tracing::info!(
target: "gunnar::store",
git_dir = %git_dir.display(),
bytes = cache,
"gc reclaimed the exploded read cache"
);
}
Ok(MaintenanceReport {
strategy: report.strategy.to_owned(),
bytes_before: report.bytes_before + cache,
bytes_after: report.bytes_after,
rows: report.rows,
retired_packs: report.retired_packs,
verified: report.verified,
sections_sealed: sections.len(),
})
}
pub use znippy_plugin_git::archive_file::{
ArchiveInReceipt, ArchiveOutReceipt, Segment, OBJECT_FORMAT,
};
pub type Cut = ArchiveOutReceipt;
pub type Restored = ArchiveInReceipt;
pub fn archive_in(file: &Path, dir: &Path) -> Result<ArchiveInReceipt> {
znippy_plugin_git::archive_file::archive_in(file, dir)
}
pub fn archive_out(dir: &Path, to: &Path, from_generation: u64) -> Result<ArchiveOutReceipt> {
znippy_plugin_git::archive_file::archive_out(dir, to, from_generation)
}
pub fn archive_out_to<W: std::io::Write>(
dir: &Path,
sink: &mut W,
from_generation: u64,
) -> Result<ArchiveOutReceipt> {
znippy_plugin_git::archive_file::archive_out_to(dir, sink, from_generation)
}
pub fn parse_explode_budget(spec: &str) -> Result<u64> {
znippy_plugin_git::exploded_cache::parse_budget(spec)
}
pub fn human_bytes(bytes: u64) -> String {
znippy_plugin_git::exploded_cache::human(bytes)
}
pub fn set_graph_cache_default(on: bool) {
znippy_plugin_git::arms::set_graph_cache_default(on);
}
pub fn hold_twin_memory_bound() {
znippy_plugin_git::set_explode_default(
znippy_plugin_git::exploded_arrow::ExplodePolicy::Off,
);
znippy_plugin_git::set_explode_budget_default(Some(
crate::storage::TWIN_EXPLODE_BUDGET_BYTES,
));
znippy_plugin_git::set_resolve_workers_default(crate::storage::TWIN_RESOLVE_WORKERS);
znippy_plugin_git::set_resolve_lookahead_default(
crate::storage::TWIN_RESOLVE_LOOKAHEAD_BYTES,
);
tracing::info!(
target: gunnar_core::observe::TARGET_TWIN,
event = "twin.memory_bound",
explode = ?znippy_plugin_git::explode_default(),
explode_budget_bytes = ?znippy_plugin_git::budget_default(),
resolve_workers = znippy_plugin_git::resolve::workers_from_env(),
resolve_lookahead_bytes = znippy_plugin_git::resolve::lookahead_from_env(),
"the twin's memory bound, as the engine now resolves it for every store opened from here on"
);
}
pub fn bound_open_stores_to_twin() -> u64 {
let mut freed = 0;
for (root, _) in store_cache::live(&STORES) {
let git_dir = match root.parent() {
Some(d) => d.to_path_buf(),
None => continue,
};
set_explode_budget(&git_dir, crate::storage::TWIN_EXPLODE_BUDGET_BYTES);
freed += drop_explode_cache(&git_dir);
}
freed
}
#[cfg(test)]
mod tests {
use std::sync::Mutex;
use super::*;
static SERIAL: Mutex<()> = Mutex::new(());
#[test]
fn the_twin_memory_bound_reaches_the_resolver_and_the_next_store_open() {
let _g = SERIAL.lock().unwrap_or_else(|e| e.into_inner());
hold_twin_memory_bound();
assert_eq!(
znippy_plugin_git::exploded_arrow::ExplodePolicy::from_env().unwrap(),
znippy_plugin_git::exploded_arrow::ExplodePolicy::Off,
"a store opened now would still explode objects — a twin holds no read cache"
);
assert_eq!(
znippy_plugin_git::exploded_cache::budget_from_env(std::path::Path::new("/"))
.unwrap(),
crate::storage::TWIN_EXPLODE_BUDGET_BYTES,
"the exploded pool is still sized from the filesystem, not from the twin's cap"
);
let cancel = std::sync::atomic::AtomicBool::new(false);
let progress = znippy_plugin_git::progress::IndexProgress::new("twin-bound-guard");
let dog = Arc::new(znippy_plugin_git::memguard::MemWatchdog::from_env());
let spill = || 0u64;
let limits = znippy_plugin_git::resolve::ResolveLimits::for_indexer(
"twin", &cancel, &progress, dog, &spill,
);
assert_eq!(
limits.workers,
crate::storage::TWIN_RESOLVE_WORKERS,
"the indexer would still resolve one worker per core: +133 MiB measured"
);
assert_eq!(
limits.lookahead_bytes,
crate::storage::TWIN_RESOLVE_LOOKAHEAD_BYTES,
"the resolver's workers may still run 32 MiB ahead of the stream"
);
}
#[test]
fn the_cache_does_not_keep_a_store_alive() {
let _serial = SERIAL.lock().unwrap_or_else(|p| p.into_inner());
shutdown();
let dir = tempfile::tempdir().expect("tempdir");
let git_dir = dir.path().join("repo.git");
std::fs::create_dir_all(&git_dir).expect("git dir");
let store = open_shared(&git_dir, gix_hash::Kind::Sha1).expect("opening the store");
assert_eq!(
Arc::strong_count(&store),
1,
"something other than this test holds the store. A strong entry in the cache is \
what made every repository the process ever opened immortal."
);
let again = open_shared(&git_dir, gix_hash::Kind::Sha1).expect("second open");
assert!(
Arc::ptr_eq(&store, &again),
"two live opens of one root produced two stores, which is the defect this cache \
exists to prevent"
);
drop(again);
let weak = Arc::downgrade(&store);
drop(store);
assert!(
weak.upgrade().is_none(),
"the store outlived its last handle, so the LRU above it still frees nothing"
);
shutdown();
}
#[test]
fn shutdown_is_safe_with_nothing_open() {
let _serial = SERIAL.lock().unwrap_or_else(|p| p.into_inner());
shutdown();
shutdown();
}
}