use std::io::BufRead;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use anyhow::{anyhow, bail, Context, Result};
use gunnar_store::git_storage_trait::GitServe;
use gunnar_store::RepoStore;
#[cfg(not(any(feature = "engine-znippy", feature = "engine-gix")))]
compile_error!(
"gunnar-server must be built with at least one storage engine: `engine-znippy` (the \
default) or `engine-gix`. A server binary with no engine is not a smaller server — it \
is the defect this seam replaced: it built, it shipped, it answered every push with \
`this binary carries no storage engine`, and the build matrix could not tell it from \
the real one."
);
#[cfg(feature = "engine-gix")]
mod gix_engine;
mod pack_end;
mod store_cache;
#[cfg(feature = "engine-znippy")]
pub mod writer;
#[cfg(feature = "engine-znippy")]
pub mod znippy_engine;
#[cfg(feature = "engine-znippy")]
pub mod encrypted_quarantine;
pub fn writer_arm() -> Option<&'static str> {
#[cfg(feature = "engine-znippy")]
{
writer::selected().map(|s| s.arm.as_str())
}
#[cfg(not(feature = "engine-znippy"))]
{
None
}
}
pub fn shutdown() {
#[cfg(feature = "engine-znippy")]
znippy_engine::shutdown();
#[cfg(feature = "engine-gix")]
gix_engine::shutdown();
}
pub const ENGINE_ENV: &str = "GUNNAR_ENGINE";
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Engine {
#[cfg(feature = "engine-znippy")]
Znippy,
#[cfg(feature = "engine-gix")]
Gix,
}
#[cfg(feature = "engine-znippy")]
pub const DEFAULT_ENGINE: Engine = Engine::Znippy;
#[cfg(all(not(feature = "engine-znippy"), feature = "engine-gix"))]
pub const DEFAULT_ENGINE: Engine = Engine::Gix;
impl std::fmt::Display for Engine {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(self.name())
}
}
pub const TWIN_RESOLVE_WORKERS: usize = 2;
pub const TWIN_RESOLVE_LOOKAHEAD_BYTES: u64 = 16 << 20;
pub const TWIN_EXPLODE_BUDGET_BYTES: u64 = 16 << 20;
pub const NORDISK_REPOS: &str = "https://git.gunnar.rs/nordisk";
impl Engine {
pub const COMPILED: &'static [Engine] = &[
#[cfg(feature = "engine-znippy")]
Engine::Znippy,
#[cfg(feature = "engine-gix")]
Engine::Gix,
];
pub fn name(self) -> &'static str {
match self {
#[cfg(feature = "engine-znippy")]
Engine::Znippy => "znippy",
#[cfg(feature = "engine-gix")]
Engine::Gix => "gix",
}
}
pub fn replication_can_read_repositories(self) -> bool {
match self {
#[cfg(feature = "engine-znippy")]
Engine::Znippy => false,
#[cfg(feature = "engine-gix")]
Engine::Gix => true,
}
}
pub fn plugin(self) -> &'static str {
match self {
#[cfg(feature = "engine-znippy")]
Engine::Znippy => "znippy-plugin-git",
#[cfg(feature = "engine-gix")]
Engine::Gix => "storage-git-gix",
}
}
pub fn summary(self) -> &'static str {
match self {
#[cfg(feature = "engine-znippy")]
Engine::Znippy => "Arrow-IPC archive; objects + refs",
#[cfg(feature = "engine-gix")]
Engine::Gix => "gix-odb packs + gix-ref (gitoxide); objects + refs",
}
}
pub fn repo_url(self) -> String {
match self {
#[cfg(feature = "engine-znippy")]
Engine::Znippy => format!("{NORDISK_REPOS}/znippy"),
#[cfg(feature = "engine-gix")]
Engine::Gix => format!("{NORDISK_REPOS}/edda"),
}
}
pub fn report(self) -> gunnar_proto::StorageEngine {
gunnar_proto::StorageEngine {
name: self.plugin().to_owned(),
description: self.summary().to_owned(),
repo_url: self.repo_url(),
}
}
pub fn describe(self) -> String {
format!("{} ({})", self.plugin(), self.summary())
}
pub fn compiled_list() -> String {
Self::COMPILED
.iter()
.map(|e| e.name())
.collect::<Vec<_>>()
.join(", ")
}
pub fn from_name(name: &str) -> Result<Engine> {
match name.trim().to_ascii_lowercase().as_str() {
#[cfg(feature = "engine-znippy")]
"znippy" => Ok(Engine::Znippy),
#[cfg(not(feature = "engine-znippy"))]
"znippy" => Err(not_compiled_in("znippy")),
#[cfg(feature = "engine-gix")]
"gix" => Ok(Engine::Gix),
#[cfg(not(feature = "engine-gix"))]
"gix" => Err(not_compiled_in("gix")),
other => Err(anyhow!(
"unknown storage engine {other:?}; the engine names are `znippy` and `gix`, and \
this binary carries {}",
Self::compiled_list()
)),
}
}
pub fn selected() -> Result<Engine> {
match std::env::var(ENGINE_ENV) {
Ok(raw) if !raw.trim().is_empty() => Self::from_name(&raw).with_context(|| {
format!("{ENGINE_ENV}={raw:?} selects the storage engine for this process")
}),
_ => Ok(DEFAULT_ENGINE),
}
}
pub fn open_repo(self, git_dir: &Path, hash_kind: gix_hash::Kind) -> Result<Arc<OpenRepo>> {
match self {
#[cfg(feature = "engine-znippy")]
Engine::Znippy => znippy_engine::open(git_dir, hash_kind),
#[cfg(feature = "engine-gix")]
Engine::Gix => gix_engine::open(git_dir, hash_kind),
}
}
pub fn cancel_index(self, git_dir: &Path) {
match self {
#[cfg(feature = "engine-znippy")]
Engine::Znippy => znippy_engine::cancel_index(git_dir),
#[cfg(feature = "engine-gix")]
Engine::Gix => {
let _ = git_dir;
}
}
}
pub fn index_progress(self) -> Vec<(PathBuf, IndexSnapshot)> {
match self {
#[cfg(feature = "engine-znippy")]
Engine::Znippy => znippy_engine::index_progress()
.into_iter()
.map(|(dir, s)| {
(
dir,
IndexSnapshot {
active: s.active,
resumed: s.resumed,
pack_id: s.pack_id,
objects_done: s.objects_done,
objects_total: s.objects_total,
bytes_done: s.bytes_done,
bytes_total: s.bytes_total,
phase: s.phase.as_str(),
started_at_ms: s.started_at_ms,
reinflations: s.reinflations,
pauses: s.pauses,
},
)
})
.collect(),
#[cfg(feature = "engine-gix")]
Engine::Gix => Vec::new(),
}
}
pub fn explode_use(self) -> Vec<(PathBuf, ExplodeBand)> {
match self {
#[cfg(feature = "engine-znippy")]
Engine::Znippy => znippy_engine::explode_use(),
#[cfg(feature = "engine-gix")]
Engine::Gix => Vec::new(),
}
}
pub fn open_object_counts(self) -> Vec<(PathBuf, u64)> {
match self {
#[cfg(feature = "engine-znippy")]
Engine::Znippy => znippy_engine::open_object_counts(),
#[cfg(feature = "engine-gix")]
Engine::Gix => Vec::new(),
}
}
pub fn set_explode_budget(self, git_dir: &Path, bytes: u64) -> usize {
match self {
#[cfg(feature = "engine-znippy")]
Engine::Znippy => znippy_engine::set_explode_budget(git_dir, bytes),
#[cfg(feature = "engine-gix")]
Engine::Gix => {
let _ = (git_dir, bytes);
0
}
}
}
pub fn hold_no_graph_memo(self) {
match self {
#[cfg(feature = "engine-znippy")]
Engine::Znippy => znippy_engine::set_graph_cache_default(false),
#[cfg(feature = "engine-gix")]
Engine::Gix => {}
}
}
pub fn hold_twin_memory_bound(self) {
match self {
#[cfg(feature = "engine-znippy")]
Engine::Znippy => znippy_engine::hold_twin_memory_bound(),
#[cfg(feature = "engine-gix")]
Engine::Gix => {}
}
}
pub fn bound_open_stores_to_twin(self) -> u64 {
match self {
#[cfg(feature = "engine-znippy")]
Engine::Znippy => znippy_engine::bound_open_stores_to_twin(),
#[cfg(feature = "engine-gix")]
Engine::Gix => 0,
}
}
pub fn hold_index_memory(self, git_dir: &Path, meminfo: Option<(&Path, u64)>) -> bool {
match self {
#[cfg(feature = "engine-znippy")]
Engine::Znippy => znippy_engine::hold_index_memory(git_dir, meminfo),
#[cfg(feature = "engine-gix")]
Engine::Gix => {
let _ = (git_dir, meminfo);
false
}
}
}
pub fn drop_explode_cache(self, git_dir: &Path) -> u64 {
match self {
#[cfg(feature = "engine-znippy")]
Engine::Znippy => znippy_engine::drop_explode_cache(git_dir),
#[cfg(feature = "engine-gix")]
Engine::Gix => {
let _ = git_dir;
0
}
}
}
pub fn ingest_push(
self,
git_dir: &Path,
hash_kind: gix_hash::Kind,
pack: &mut dyn BufRead,
objects: ObjectSet,
progress: Option<&dyn git_storage_trait::PackProgress>,
before_store: Option<BeforeStore<'_>>,
) -> Result<PushReceipt> {
match self {
#[cfg(feature = "engine-znippy")]
Engine::Znippy => {
znippy_engine::ingest(git_dir, hash_kind, pack, objects, progress, before_store)
}
#[cfg(feature = "engine-gix")]
Engine::Gix => {
gix_engine::ingest(git_dir, hash_kind, pack, objects, progress, before_store)
}
}
}
pub fn maintain(self, git_dir: &Path, hash_kind: gix_hash::Kind) -> Result<MaintenanceReport> {
match self {
#[cfg(feature = "engine-znippy")]
Engine::Znippy => znippy_engine::maintain(git_dir, hash_kind),
#[cfg(feature = "engine-gix")]
Engine::Gix => gix_engine::maintain(git_dir, hash_kind),
}
}
}
#[allow(dead_code)]
fn not_compiled_in(name: &str) -> anyhow::Error {
anyhow!(
"this binary was built without the `{name}` storage engine; it carries {}. Rebuild with \
`--features engine-{name}`, or select an engine that is here. Refusing rather than \
serving from a different engine: a server that quietly answers out of the wrong store \
is worse than one that will not start.",
Engine::compiled_list()
)
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct IndexSnapshot {
pub active: bool,
pub resumed: bool,
pub pack_id: u64,
pub objects_done: u64,
pub objects_total: u64,
pub bytes_done: u64,
pub bytes_total: u64,
pub phase: &'static str,
pub started_at_ms: u64,
pub reinflations: u64,
pub pauses: u64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct ExplodeBand {
pub eager_bytes: u64,
pub lazy_bytes: u64,
pub budget: u64,
pub segments: u64,
pub evicted_segments: u64,
pub evicted_bytes: u64,
pub written_back: u64,
}
impl ExplodeBand {
pub fn used(&self) -> u64 {
self.eager_bytes + self.lazy_bytes
}
pub fn at_limit(&self) -> bool {
self.budget != 0 && self.used() >= self.budget
}
}
impl IndexSnapshot {
pub fn percent(&self) -> u64 {
(self.objects_done * 100)
.checked_div(self.objects_total)
.map_or(0, |percent| percent.min(100))
}
}
pub struct OpenRepo {
git_dir: PathBuf,
hash_kind: gix_hash::Kind,
serve: Arc<dyn GitServe>,
raw: Option<Arc<dyn GitServe>>,
}
impl OpenRepo {
pub(crate) fn new<S: GitServe + 'static>(
git_dir: &Path,
hash_kind: gix_hash::Kind,
store: Arc<S>,
) -> Arc<Self> {
Arc::new(OpenRepo {
git_dir: git_dir.to_path_buf(),
hash_kind,
serve: store as Arc<dyn GitServe>,
raw: None,
})
}
pub fn serve(&self) -> &dyn GitServe {
self.serve.as_ref()
}
pub(crate) fn logged(
self: &Arc<Self>,
log: Arc<gunnar_core::txlog::Log>,
repo: &str,
hash: &str,
) -> Arc<Self> {
let tee = gunnar_core::txlog::tee::LoggedGit::new(Arc::clone(&self.serve), log, repo, hash);
Arc::new(OpenRepo {
git_dir: self.git_dir.clone(),
hash_kind: self.hash_kind,
serve: Arc::new(tee) as Arc<dyn GitServe>,
raw: Some(Arc::clone(&self.serve)),
})
}
pub fn unlogged(&self) -> &dyn GitServe {
self.raw.as_deref().unwrap_or(self.serve.as_ref())
}
pub fn git_dir(&self) -> &Path {
&self.git_dir
}
pub fn hash_kind(&self) -> gix_hash::Kind {
self.hash_kind
}
}
impl RepoStore for OpenRepo {
fn git_dir(&self) -> &Path {
&self.git_dir
}
fn hash_kind(&self) -> gix_hash::Kind {
self.hash_kind
}
fn git(&self) -> Arc<dyn GitServe> {
Arc::clone(&self.serve)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ObjectSet {
Required,
Wanted,
NotNeeded,
}
pub type BeforeStore<'a> = &'a dyn Fn(&[u8]) -> Result<()>;
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct PushReceipt {
pub num_objects: u32,
pub version: u32,
pub pack_id: Vec<u8>,
pub pack_bytes: u64,
pub new_oids: Option<Vec<Vec<u8>>>,
pub elapsed_ms: u64,
}
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct MaintenanceReport {
pub strategy: String,
pub bytes_before: u64,
pub bytes_after: u64,
pub rows: u64,
pub retired_packs: u64,
pub verified: bool,
pub sections_sealed: usize,
}
pub(crate) struct PushedPack {
pub bytes: PackBytes,
pub num_objects: u32,
pub version: u32,
}
pub(crate) const SPOOL_ABOVE: usize = 64 << 20;
pub(crate) fn spooled_on_disk(consumed: u64, spool_above: u64) -> u64 {
if consumed >= spool_above {
consumed
} else {
0
}
}
pub(crate) enum PackBytes {
Heap(Vec<u8>),
Spooled {
_file: tempfile::NamedTempFile,
map: memmap2::Mmap,
},
}
impl std::ops::Deref for PackBytes {
type Target = [u8];
fn deref(&self) -> &[u8] {
match self {
PackBytes::Heap(v) => v,
PackBytes::Spooled { map, .. } => map,
}
}
}
impl PartialEq<Vec<u8>> for PackBytes {
fn eq(&self, other: &Vec<u8>) -> bool {
&**self == other.as_slice()
}
}
impl std::fmt::Debug for PackBytes {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
PackBytes::Heap(v) => write!(f, "{} byte(s) on the heap", v.len()),
PackBytes::Spooled { map, .. } => write!(f, "{} byte(s) spooled", map.len()),
}
}
}
impl std::fmt::Debug for PushedPack {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("PushedPack")
.field("bytes", &format_args!("{} byte(s)", self.bytes.len()))
.field("num_objects", &self.num_objects)
.field("version", &self.version)
.finish()
}
}
pub(crate) fn read_pushed_pack_spooling(
pack: &mut dyn BufRead,
hash_kind: gix_hash::Kind,
spool_dir: Option<&Path>,
) -> Result<PushedPack> {
read_pushed_pack_spooling_at(pack, hash_kind, spool_dir, SPOOL_ABOVE)
}
pub(crate) fn read_pushed_pack_spooling_at(
pack: &mut dyn BufRead,
hash_kind: gix_hash::Kind,
spool_dir: Option<&Path>,
spool_above: usize,
) -> Result<PushedPack> {
use std::io::Write as _;
let mut scanner = pack_end::PackScanner::new(hash_kind.len_in_bytes());
let mut bytes = Vec::new();
let mut spool: Option<(tempfile::NamedTempFile, u64)> = None;
while !scanner.done() {
let used = {
let available = pack
.fill_buf()
.context("reading the pushed packfile off the wire")?;
if available.is_empty() {
if bytes.is_empty() {
bail!("an empty push is not a pack");
}
bail!(
"the pushed pack ended after {} byte(s), in the middle of {}. The client \
hung up mid-push; nothing was stored",
bytes.len(),
scanner.waiting_for()
);
}
let used = scanner.feed(available)?;
match &mut spool {
Some((f, n)) => {
f.write_all(&available[..used])
.context("spooling the pushed pack")?;
*n += used as u64;
}
None => {
bytes.extend_from_slice(&available[..used]);
if let (Some(dir), true) = (spool_dir, bytes.len() >= spool_above) {
let mut f = tempfile::Builder::new()
.prefix("incoming-")
.suffix(".pack")
.tempfile_in(dir)
.with_context(|| {
format!("spooling the pushed pack under {}", dir.display())
})?;
f.write_all(&bytes).context("spooling the pushed pack")?;
let n = bytes.len() as u64;
bytes = Vec::new();
tracing::info!(
target: "gunnar::store",
event = "push.spooled",
dir = %dir.display(),
threshold = spool_above,
"a pushed pack over {} MiB is spooled to disk while it arrives, not held in RAM",
spool_above >> 20
);
spool = Some((f, n));
}
}
}
used
};
pack.consume(used);
}
let (version, num_objects) = scanner.header();
let bytes = match spool {
None => PackBytes::Heap(bytes),
Some((f, n)) => {
f.as_file().sync_data().ok();
let map =
unsafe { memmap2::Mmap::map(f.as_file()) }.context("mapping the spooled pack")?;
tracing::info!(
target: "gunnar::store",
event = "push.spooled.mapped",
bytes = n,
"the spooled pack is complete and mapped for the walk and the append"
);
PackBytes::Spooled { _file: f, map }
}
};
Ok(PushedPack {
bytes,
num_objects,
version,
})
}
pub(crate) fn pack_trailer(bytes: &[u8], hash_kind: gix_hash::Kind) -> Result<Vec<u8>> {
let len = hash_kind.len_in_bytes();
bytes
.len()
.checked_sub(len)
.and_then(|at| bytes.get(at..))
.map(<[u8]>::to_vec)
.ok_or_else(|| anyhow!("the pushed pack has no trailing {hash_kind} checksum"))
}
#[cfg(test)]
mod tests {
use super::pack_end::fixture::fixture_pack;
use super::*;
#[test]
fn the_engine_is_reported_as_name_description_and_source_repository() {
for engine in Engine::COMPILED {
let r = engine.report();
assert!(
!r.name.is_empty() && !r.name.contains(' '),
"{engine:?}: the name is one word: {r:?}"
);
assert!(
!r.description.is_empty() && !r.description.contains(&r.name),
"{engine:?}: {r:?}"
);
assert!(r.repo_url.starts_with("https://"), "{engine:?}: {r:?}");
assert_eq!(
engine.describe(),
format!("{} ({})", r.name, r.description),
"the CLI's line is the same parts"
);
}
#[cfg(feature = "engine-znippy")]
{
let z = Engine::Znippy.report();
assert_eq!(
(z.name.as_str(), z.description.as_str()),
("znippy-plugin-git", "Arrow-IPC archive; objects + refs")
);
assert_eq!(z.repo_url, format!("{NORDISK_REPOS}/znippy"));
assert_eq!(
z.repo_url, "https://git.gunnar.rs/nordisk/znippy",
"znippy's home is nordisk on git.gunnar.rs"
);
}
#[cfg(feature = "engine-gix")]
{
let g = Engine::Gix.report();
assert_eq!(g.name, "storage-git-gix");
assert_eq!(
g.repo_url,
format!("{NORDISK_REPOS}/edda"),
"storage-git-gix's source is edda"
);
}
}
fn read_pushed_pack(pack: &mut dyn BufRead, hash_kind: gix_hash::Kind) -> Result<PushedPack> {
read_pushed_pack_spooling(pack, hash_kind, None)
}
struct ClientWaitingForItsReport {
stream: Vec<u8>,
at: usize,
pack_len: usize,
reads_past_the_pack: usize,
}
impl ClientWaitingForItsReport {
fn holding(pack: &[u8]) -> Self {
ClientWaitingForItsReport {
stream: pack.to_vec(),
at: 0,
pack_len: pack.len(),
reads_past_the_pack: 0,
}
}
fn with_a_report_behind_the_pack(pack: &[u8], report: &[u8]) -> Self {
let mut stream = pack.to_vec();
stream.extend_from_slice(report);
ClientWaitingForItsReport {
stream,
at: 0,
pack_len: pack.len(),
reads_past_the_pack: 0,
}
}
fn what_is_left(&self) -> &[u8] {
&self.stream[self.at..]
}
}
impl std::io::Read for ClientWaitingForItsReport {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
let available = self.fill_buf()?.len().min(buf.len());
buf[..available].copy_from_slice(&self.stream[self.at..self.at + available]);
self.consume(available);
Ok(available)
}
}
impl BufRead for ClientWaitingForItsReport {
fn fill_buf(&mut self) -> std::io::Result<&[u8]> {
if self.at >= self.pack_len && self.at >= self.stream.len() {
self.reads_past_the_pack += 1;
return Err(std::io::Error::new(
std::io::ErrorKind::WouldBlock,
"the client sent its whole pack and is waiting for the status report; this \
read would block for ever",
));
}
let end = self.stream.len().min(self.at + 997);
Ok(&self.stream[self.at..end])
}
fn consume(&mut self, amount: usize) {
self.at += amount;
}
}
#[test]
fn a_pack_over_the_threshold_is_spooled_to_disk_and_mapped() {
let pack = fixture_pack();
let dir = tempfile::tempdir().expect("tempdir");
let mut client = std::io::Cursor::new(pack.clone());
let pushed =
read_pushed_pack_spooling_at(&mut client, gix_hash::Kind::Sha1, Some(dir.path()), 0)
.expect("a well-formed pack");
assert!(
matches!(pushed.bytes, PackBytes::Spooled { .. }),
"{:?}",
pushed.bytes
);
assert_eq!(pushed.bytes, pack, "the mapped pack is not the pack sent");
assert_eq!(pushed.bytes.len(), pack.len());
let spooled: Vec<_> = std::fs::read_dir(dir.path()).unwrap().flatten().collect();
assert_eq!(
spooled.len(),
1,
"one incoming-*.pack while the pack is held"
);
assert!(spooled[0]
.file_name()
.to_string_lossy()
.starts_with("incoming-"));
assert_eq!(
pack_trailer(&pushed.bytes, gix_hash::Kind::Sha1)
.unwrap()
.len(),
20
);
drop(pushed);
assert_eq!(
std::fs::read_dir(dir.path()).unwrap().count(),
0,
"the spool file outlived the pack"
);
let mut client = std::io::Cursor::new(pack.clone());
let pushed = read_pushed_pack_spooling_at(
&mut client,
gix_hash::Kind::Sha1,
Some(dir.path()),
usize::MAX,
)
.unwrap();
assert!(matches!(pushed.bytes, PackBytes::Heap(_)));
assert_eq!(std::fs::read_dir(dir.path()).unwrap().count(), 0);
}
#[test]
fn the_spool_holds_what_spooled_on_disk_says_at_every_consume() {
struct Probe<'a> {
inner: std::io::Cursor<Vec<u8>>,
dir: &'a Path,
consumed: u64,
threshold: u64,
checked: u32,
}
impl std::io::Read for Probe<'_> {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
std::io::Read::read(&mut self.inner, buf)
}
}
impl BufRead for Probe<'_> {
fn fill_buf(&mut self) -> std::io::Result<&[u8]> {
let b = self.inner.fill_buf()?;
Ok(&b[..b.len().min(37)])
}
fn consume(&mut self, n: usize) {
self.inner.consume(n);
self.consumed += n as u64;
let on_disk: u64 = std::fs::read_dir(self.dir)
.unwrap()
.flatten()
.map(|e| e.metadata().unwrap().len())
.sum();
assert_eq!(
on_disk,
spooled_on_disk(self.consumed, self.threshold),
"after {} byte(s) consumed",
self.consumed
);
self.checked += 1;
}
}
let pack = fixture_pack();
let threshold = (pack.len() / 3) as u64;
let dir = tempfile::tempdir().unwrap();
let mut probe = Probe {
inner: std::io::Cursor::new(pack.clone()),
dir: dir.path(),
consumed: 0,
threshold,
checked: 0,
};
let pushed = read_pushed_pack_spooling_at(
&mut probe,
gix_hash::Kind::Sha1,
Some(dir.path()),
threshold as usize,
)
.unwrap();
assert_eq!(pushed.bytes, pack);
assert!(probe.checked > 3, "{} consume(s)", probe.checked);
}
#[test]
fn the_pack_read_stops_at_the_trailer_and_never_waits_for_eof() {
let pack = fixture_pack();
let mut client = ClientWaitingForItsReport::holding(&pack);
let pushed = read_pushed_pack(&mut client, gix_hash::Kind::Sha1)
.expect("the pack is complete; nothing more is owed before the report");
assert_eq!(
client.reads_past_the_pack, 0,
"the server made {} read(s) past the last byte of the pack. There is nobody to \
answer them: the client is waiting for the status report those reads come before, \
and on a socket this is not an error, it is a deadlock",
client.reads_past_the_pack
);
assert_eq!(pushed.bytes, pack, "the stored pack is not the pack sent");
assert_eq!(pushed.num_objects, 4);
assert_eq!(pushed.version, 2);
}
#[test]
fn what_follows_the_pack_is_left_in_the_stream() {
let pack = fixture_pack();
let report = b"0009done\n0000";
let mut client = ClientWaitingForItsReport::with_a_report_behind_the_pack(&pack, report);
let pushed =
read_pushed_pack(&mut client, gix_hash::Kind::Sha1).expect("a well-formed pack");
assert_eq!(pushed.bytes.len(), pack.len());
assert_eq!(
client.what_is_left(),
report,
"the bytes after the pack were swallowed, so whatever spoke next would be read \
out of the middle of the pack"
);
assert_eq!(client.reads_past_the_pack, 0);
}
#[test]
fn a_client_that_hangs_up_mid_pack_is_refused_and_says_where() {
let pack = fixture_pack();
let truncated = &pack[..pack.len() - 4];
let mut client = std::io::Cursor::new(truncated.to_vec());
let err = read_pushed_pack(&mut client, gix_hash::Kind::Sha1)
.expect_err("half a pack must not be accepted as a push");
let message = format!("{err:#}");
assert!(
message.contains("trailing checksum") && message.contains("nothing was stored"),
"the refusal must name what never arrived: {message}"
);
}
#[test]
fn an_empty_push_is_still_not_a_pack() {
let mut nothing = std::io::Cursor::new(Vec::new());
let err = read_pushed_pack(&mut nothing, gix_hash::Kind::Sha1).expect_err("no bytes");
assert!(format!("{err:#}").contains("an empty push is not a pack"));
}
#[test]
#[cfg(feature = "engine-znippy")]
fn the_default_is_znippy() {
assert_eq!(DEFAULT_ENGINE, Engine::Znippy);
assert_eq!(Engine::COMPILED[0], Engine::Znippy);
}
#[test]
fn every_compiled_engine_answers_to_its_name() {
for engine in Engine::COMPILED {
assert_eq!(
Engine::from_name(engine.name()).expect("a compiled engine resolves"),
*engine
);
}
}
#[test]
fn an_unknown_engine_is_refused() {
let err = Engine::from_name("postgres").expect_err("not an engine");
let msg = format!("{err:#}");
assert!(msg.contains("unknown storage engine"), "{msg}");
assert!(msg.contains(&Engine::compiled_list()), "{msg}");
}
#[test]
#[cfg(all(feature = "engine-znippy", not(feature = "engine-gix")))]
fn an_engine_that_was_not_compiled_in_refuses_rather_than_falling_back() {
let err = Engine::from_name("gix").expect_err("gix is not in this build");
let msg = format!("{err:#}");
assert!(
msg.contains("built without the `gix` storage engine"),
"{msg}"
);
assert!(msg.contains("--features engine-gix"), "{msg}");
}
#[test]
#[cfg(all(feature = "engine-gix", not(feature = "engine-znippy")))]
fn znippy_is_refused_when_it_was_not_compiled_in() {
let err = Engine::from_name("znippy").expect_err("znippy is not in this build");
assert!(
format!("{err:#}").contains("built without the `znippy` storage engine"),
"{err:#}"
);
}
}