use std::collections::{BinaryHeap, HashMap, HashSet, VecDeque};
use gix_hash::{Kind, ObjectId};
use gunnar_store::git_storage_trait::{GitServe, ObjType};
use gunnar_store::object;
use crate::filter::Filter;
use crate::limits::Limits;
use crate::{Error, Result};
fn resolved_kind(id: &ObjectId, kind: ObjType) -> Result<object::Kind> {
object::Kind::resolved(kind).ok_or_else(|| Error::CorruptObject {
oid: *id,
reason: format!(
"the store answered a resolved read with {}, which is a delta entry \
and not an object type",
kind.as_str()
),
})
}
pub(crate) fn backend(err: impl std::fmt::Display) -> Error {
Error::Store(gunnar_store::Error::backend(format!("{err:#}")))
}
pub(crate) fn oid(bytes: &[u8]) -> Result<ObjectId> {
ObjectId::try_from(bytes).map_err(|err| {
backend(format_args!(
"the store answered with a {}-byte object id: {err}",
bytes.len()
))
})
}
pub(crate) fn has_for_negotiation(store: &dyn GitServe, id: &gix_hash::oid) -> bool {
match store.has(id.as_bytes()) {
Ok(present) => present,
Err(err) => {
tracing::warn!(
target: crate::receive_pack::GIT_TARGET,
event = "git.have.store_error",
oid = %id,
error = %format_args!("{err:#}"),
"the object store failed answering a `have`; treating it as absent, which \
widens the pack and cannot corrupt it"
);
false
}
}
}
const MAX_TAG_DEPTH: usize = 16;
pub const DEFAULT_BUDGET: usize = 10_000_000;
pub const CONNECTIVITY_BUDGET: usize = 64_000_000;
#[derive(Debug, Default)]
pub(crate) struct OidSet {
ids: HashSet<ObjectId>,
}
impl OidSet {
pub(crate) fn new() -> Self {
OidSet::default()
}
pub(crate) fn insert(&mut self, id: ObjectId) -> bool {
let fresh = self.ids.insert(id);
if fresh {
meter::acquire(1);
}
fresh
}
pub(crate) fn into_inner(mut self) -> HashSet<ObjectId> {
let ids = std::mem::take(&mut self.ids);
meter::release(ids.len());
ids
}
pub(crate) fn extend(&mut self, ids: impl IntoIterator<Item = ObjectId>) {
let before = self.ids.len();
self.ids.extend(ids);
meter::acquire(self.ids.len() - before);
}
pub(crate) fn remove(&mut self, id: &ObjectId) -> bool {
let gone = self.ids.remove(id);
if gone {
meter::release(1);
}
gone
}
}
impl std::ops::Deref for OidSet {
type Target = HashSet<ObjectId>;
fn deref(&self) -> &HashSet<ObjectId> {
&self.ids
}
}
impl Drop for OidSet {
fn drop(&mut self) {
meter::release(self.ids.len());
}
}
impl FromIterator<ObjectId> for OidSet {
fn from_iter<I: IntoIterator<Item = ObjectId>>(ids: I) -> Self {
let mut set = OidSet::new();
set.extend(ids);
set
}
}
pub(crate) mod meter {
use std::cell::Cell;
thread_local! {
static LIVE: Cell<usize> = const { Cell::new(0) };
static PEAK: Cell<usize> = const { Cell::new(0) };
}
pub(crate) fn acquire(count: usize) {
if count == 0 {
return;
}
let live = LIVE.with(|live| {
let now = live.get() + count;
live.set(now);
now
});
PEAK.with(|peak| {
if live > peak.get() {
peak.set(live);
}
});
}
pub(crate) fn release(count: usize) {
LIVE.with(|live| live.set(live.get().saturating_sub(count)));
}
pub(crate) fn reset_peak() {
PEAK.with(|peak| peak.set(LIVE.with(|live| live.get())));
}
#[cfg(test)]
pub(crate) fn peak() -> usize {
PEAK.with(|peak| peak.get())
}
pub(crate) fn live() -> usize {
LIVE.with(|live| live.get())
}
pub(crate) fn restore(to: usize) {
LIVE.with(|live| live.set(to));
}
}
pub(crate) enum Emit<'v> {
Keep(&'v mut Vec<ObjectId>),
Tally(usize),
}
impl Emit<'_> {
fn push(&mut self, id: ObjectId) {
match self {
Emit::Keep(out) => out.push(id),
Emit::Tally(count) => *count += 1,
}
}
fn len(&self) -> usize {
match self {
Emit::Keep(out) => out.len(),
Emit::Tally(count) => *count,
}
}
}
pub(crate) trait Seen {
fn insert(&mut self, id: &ObjectId) -> bool;
}
impl Seen for OidSet {
fn insert(&mut self, id: &ObjectId) -> bool {
OidSet::insert(self, *id)
}
}
struct Carried<'a> {
seen: &'a mut OidSet,
carried: &'a mut Vec<ObjectId>,
}
impl Seen for Carried<'_> {
fn insert(&mut self, id: &ObjectId) -> bool {
let fresh = self.seen.insert(*id);
if fresh {
self.carried.push(*id);
}
fresh
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub(crate) enum Missing {
#[default]
Fail,
Skip,
}
pub(crate) struct Graph<'a> {
store: &'a dyn GitServe,
hash_kind: Kind,
budget: Option<usize>,
walk_bytes: Option<usize>,
pool: Option<std::sync::Arc<crate::WalkPool>>,
reserved: std::cell::Cell<usize>,
entered_at: usize,
}
impl Drop for Graph<'_> {
fn drop(&mut self) {
if let Some(pool) = &self.pool {
pool.release(self.reserved.get());
}
meter::restore(self.entered_at);
}
}
impl<'a> Graph<'a> {
pub(crate) fn new(store: &'a dyn GitServe, hash_kind: Kind) -> Self {
let entered_at = meter::live();
meter::reset_peak();
Graph {
hash_kind,
store,
budget: None,
walk_bytes: None,
pool: None,
reserved: std::cell::Cell::new(0),
entered_at,
}
}
pub(crate) fn with_budget(mut self, budget: Option<usize>) -> Self {
self.budget = budget;
self
}
pub(crate) fn with_walk_limit(
mut self,
walk_bytes: Option<usize>,
pool: Option<std::sync::Arc<crate::WalkPool>>,
) -> Self {
self.walk_bytes = walk_bytes;
self.pool = pool;
self
}
fn charge_walk(&self) -> Result<()> {
let held = meter::live().saturating_sub(self.entered_at);
self.charge_budget(held)?;
let bytes = held.saturating_mul(crate::WALK_BYTES_PER_OBJECT);
if let Some(limit) = self.walk_bytes {
if bytes > limit {
return Err(Error::WalkTooLarge { held: bytes, limit });
}
}
if let Some(pool) = &self.pool {
if bytes >= crate::WalkPool::FLOOR_BYTES && bytes > self.reserved.get() {
let want = bytes.next_multiple_of(crate::WalkPool::CHUNK_BYTES);
pool.try_reserve(want - self.reserved.get())?;
self.reserved.set(want);
}
}
Ok(())
}
fn charge_budget(&self, held: usize) -> Result<()> {
if let Some(limit) = self.budget {
if held >= limit {
return Err(Error::WalkBudget { limit });
}
}
Ok(())
}
pub(crate) fn charge_answer(&self, objects: usize) -> Result<()> {
self.charge_budget(objects)
}
pub(crate) fn contains(&self, id: &gix_hash::oid) -> Result<bool> {
self.store.has(id.as_bytes()).map_err(backend)
}
pub(crate) fn contains_have(&self, id: &gix_hash::oid) -> bool {
has_for_negotiation(self.store, id)
}
pub(crate) fn store(&self) -> &'a dyn GitServe {
self.store
}
fn load(&self, id: &ObjectId, what: &str, buf: &mut Vec<u8>) -> Result<object::Kind> {
match self.store.read(id.as_bytes()).map_err(backend)? {
Some((kind, bytes)) => {
*buf = bytes;
resolved_kind(id, kind)
}
None => Err(Error::missing(*id, what.to_owned())),
}
}
fn corrupt(&self, id: &ObjectId, err: impl std::fmt::Display) -> Error {
Error::CorruptObject {
oid: *id,
reason: err.to_string(),
}
}
pub(crate) fn kind(&self, id: &ObjectId) -> Result<Option<object::Kind>> {
match self.store.header(id.as_bytes()).map_err(backend)? {
Some((kind, _size)) => Ok(Some(resolved_kind(id, kind)?)),
None => Ok(None),
}
}
pub(crate) fn peel(&self, id: &ObjectId) -> Result<Peeled> {
let mut tags = Vec::new();
let mut current = *id;
let mut buf = Vec::new();
for depth in 0..MAX_TAG_DEPTH {
let what = if depth == 0 {
"resolving an object"
} else {
"peeling a tag"
};
let Some(kind) = self.kind(¤t)? else {
return Err(Error::missing(current, what.to_owned()));
};
if kind != object::Kind::Tag {
return Ok(Peeled {
id: current,
kind,
tags,
});
}
let loaded = self.load(¤t, "peeling a tag", &mut buf)?;
debug_assert_eq!(loaded, object::Kind::Tag);
let target = object::tag_target(&buf, self.hash_kind)
.map_err(|err| self.corrupt(¤t, err))?;
tags.push(current);
current = target;
}
Err(self.corrupt(id, format!("tag chain deeper than {MAX_TAG_DEPTH}")))
}
pub(crate) fn commit_links(
&self,
id: &ObjectId,
buf: &mut Vec<u8>,
parents: &mut Vec<ObjectId>,
) -> Result<ObjectId> {
let kind = self.load(id, "reading a commit", buf)?;
if kind != object::Kind::Commit {
return Err(self.corrupt(id, format!("expected a commit, found a {kind}")));
}
object::commit_links(buf, self.hash_kind, parents).map_err(|err| self.corrupt(id, err))
}
pub(crate) fn commit(&self, id: &ObjectId) -> Result<CommitFacts> {
let mut buf = Vec::new();
self.commit_facts(id, &mut buf)
}
fn commit_facts(&self, id: &ObjectId, buf: &mut Vec<u8>) -> Result<CommitFacts> {
let mut parents = Vec::new();
let (tree, committed_at) = self.commit_facts_into(id, buf, &mut parents)?;
Ok(CommitFacts {
tree,
parents,
committed_at,
})
}
fn commit_facts_into(
&self,
id: &ObjectId,
buf: &mut Vec<u8>,
parents: &mut Vec<ObjectId>,
) -> Result<(ObjectId, i64)> {
let kind = self.load(id, "reading a commit", buf)?;
if kind != object::Kind::Commit {
return Err(self.corrupt(id, format!("expected a commit, found a {kind}")));
}
object::commit_facts(buf, self.hash_kind, parents).map_err(|err| self.corrupt(id, err))
}
pub(crate) fn commit_into(
&self,
id: &ObjectId,
buf: &mut Vec<u8>,
parents: &mut Vec<ObjectId>,
) -> Result<i64> {
self.commit_facts_into(id, buf, parents).map(|(_, at)| at)
}
pub(crate) fn commit_closure(
&self,
tips: &[ObjectId],
stop: &HashSet<ObjectId>,
boundary: &HashSet<ObjectId>,
restrict: Option<&HashSet<ObjectId>>,
missing: Missing,
) -> Result<CommitClosure> {
let mut seen = OidSet::new();
let mut out = CommitClosure::default();
let mut queue: VecDeque<ObjectId> = VecDeque::new();
let mut edges: HashSet<ObjectId> = HashSet::new();
let mut buf = Vec::new();
let mut parents = Vec::new();
for tip in tips {
if !stop.contains(tip) && seen.insert(*tip) {
queue.push_back(*tip);
}
}
while let Some(id) = queue.pop_front() {
self.charge_walk()?;
if missing == Missing::Skip && !self.contains(&id)? {
continue;
}
let tree = match self.commit_links(&id, &mut buf, &mut parents) {
Ok(tree) => tree,
Err(Error::ObjectMissing { .. }) if missing == Missing::Skip => continue,
Err(err) => return Err(err),
};
out.ids.push(id);
out.trees.push(tree);
if boundary.contains(&id) {
continue;
}
for parent in parents.drain(..) {
if stop.contains(&parent) {
if edges.insert(parent) {
out.edges.push(parent);
}
continue;
}
if restrict.is_some_and(|allowed| !allowed.contains(&parent)) {
continue;
}
if seen.insert(parent) {
queue.push_back(parent);
}
}
}
Ok(out)
}
pub(crate) fn limited_closure(
&self,
tips: &[ObjectId],
haves: &[ObjectId],
boundary: &HashSet<ObjectId>,
have_boundary: &HashSet<ObjectId>,
restrict: Option<&HashSet<ObjectId>>,
missing: Missing,
) -> Result<LimitedClosure> {
let mut walk = Limiter::default();
let mut buf = Vec::new();
for id in haves {
walk.admit(self, *id, true, Missing::Skip, &mut buf)?;
}
for id in tips {
walk.admit(self, *id, false, missing, &mut buf)?;
}
let mut out = LimitedClosure::default();
let mut slop = SLOP;
let mut date = i64::MAX;
let mut cache: Option<ObjectId> = None;
while let Some(Pending { id, .. }) = walk.queue.pop() {
if cache == Some(id) {
cache = None;
}
self.charge_walk()?;
let uninteresting = walk.is_uninteresting(&id);
let facts = walk
.parsed
.get(&id)
.expect("only a parsed commit is queued");
let (tree, commit_date) = (facts.tree, facts.committed_at);
let parents = facts.parents.clone();
if uninteresting {
out.uninteresting.push(id);
if !have_boundary.contains(&id) {
for parent in &parents {
walk.admit(self, *parent, true, Missing::Skip, &mut buf)?;
if have_boundary.contains(parent) {
continue;
}
walk.mark_parents_uninteresting(parent, have_boundary);
}
}
slop = walk.still_interesting(date, slop, &mut cache);
if slop > 0 {
continue;
}
break;
}
if !boundary.contains(&id) {
for parent in &parents {
if restrict.is_some_and(|allowed| !allowed.contains(parent)) {
continue;
}
walk.admit(self, *parent, false, missing, &mut buf)?;
}
}
date = commit_date;
out.ids.push(id);
out.trees.push(tree);
}
let mut edges: HashSet<ObjectId> = HashSet::new();
for id in &out.ids {
if boundary.contains(id) {
continue;
}
let Some(facts) = walk.parsed.get(id) else {
continue;
};
for parent in &facts.parents {
if !walk.is_uninteresting(parent) {
continue;
}
if edges.insert(*parent) {
out.edges.push(*parent);
}
}
}
Ok(out)
}
#[allow(clippy::too_many_arguments)]
pub(crate) fn tree_closure<S: Seen + ?Sized>(
&self,
root: ObjectId,
seen: &mut S,
exclude: &HashSet<ObjectId>,
filter: Option<&Filter>,
missing: Missing,
max_blob_bytes: Option<u64>,
out: &mut Emit<'_>,
) -> Result<()> {
let mut stack = vec![(root, 1usize)];
let mut buf = Vec::new();
let mut children: Vec<(ObjectId, usize)> = Vec::new();
let mut blobs: Vec<ObjectId> = Vec::new();
while let Some((id, depth)) = stack.pop() {
self.charge_walk()?;
if exclude.contains(&id) || !seen.insert(&id) {
continue;
}
if missing == Missing::Skip && !self.contains(&id)? {
continue;
}
let keep_tree = filter.is_none_or(|f| f.keeps_tree(depth));
let kind = match self.load(&id, "reading a tree", &mut buf) {
Ok(kind) => kind,
Err(Error::ObjectMissing { .. }) if missing == Missing::Skip => continue,
Err(err) => return Err(err),
};
if kind != object::Kind::Tree {
return Err(self.corrupt(&id, format!("expected a tree, found a {kind}")));
}
if keep_tree {
out.push(id);
}
let entries = object::TreeIter::from_bytes(&buf, self.hash_kind);
children.clear();
blobs.clear();
for entry in entries {
let entry = entry.map_err(|err| self.corrupt(&id, err))?;
match entry.kind {
object::EntryKind::Tree => children.push((entry.oid, depth + 1)),
object::EntryKind::Commit => {}
_ => blobs.push(entry.oid),
}
}
for blob in blobs.drain(..) {
self.charge_walk()?;
if exclude.contains(&blob) || !seen.insert(&blob) {
continue;
}
if !self.keeps_blob(&blob, filter, missing, max_blob_bytes)? {
continue;
}
out.push(blob);
}
stack.append(&mut children);
}
Ok(())
}
fn keeps_blob(
&self,
id: &ObjectId,
filter: Option<&Filter>,
missing: Missing,
max_blob_bytes: Option<u64>,
) -> Result<bool> {
let filter_limit = match filter {
None => None,
Some(filter) => match filter.blob_limit() {
Some(limit) => Some(limit),
None if filter.keeps_any_blob() => None,
None => return Ok(false),
},
};
if filter_limit.is_none() && max_blob_bytes.is_none() {
if missing == Missing::Fail && !self.contains(id)? {
return Err(Error::missing(*id, "packing a blob"));
}
return Ok(true);
}
let Some((_kind, size)) = self.store.header(id.as_bytes()).map_err(backend)? else {
return match missing {
Missing::Skip => Ok(false),
Missing::Fail => Err(Error::missing(*id, "sizing a blob")),
};
};
if let Some(limit) = max_blob_bytes {
if size > limit {
return Err(Error::ObjectTooLarge {
oid: *id,
size,
limit,
});
}
}
Ok(filter_limit.is_none_or(|limit| size < limit))
}
}
#[derive(Debug, Clone)]
pub(crate) struct Peeled {
pub id: ObjectId,
pub kind: object::Kind,
pub tags: Vec<ObjectId>,
}
#[derive(Debug, Clone)]
pub(crate) struct CommitFacts {
pub tree: ObjectId,
pub parents: Vec<ObjectId>,
pub committed_at: i64,
}
#[derive(Debug, Default)]
pub(crate) struct CommitClosure {
pub ids: Vec<ObjectId>,
pub trees: Vec<ObjectId>,
pub edges: Vec<ObjectId>,
}
#[derive(Debug, Default)]
pub(crate) struct LimitedClosure {
pub ids: Vec<ObjectId>,
pub trees: Vec<ObjectId>,
pub edges: Vec<ObjectId>,
pub uninteresting: Vec<ObjectId>,
}
const SLOP: usize = 5;
mod flag {
pub const SEEN: u8 = 1 << 0;
pub const UNINTERESTING: u8 = 1 << 1;
pub const ABSENT: u8 = 1 << 2;
}
#[derive(Debug, PartialEq, Eq)]
struct Pending {
date: i64,
seq: std::cmp::Reverse<u64>,
id: ObjectId,
}
impl Ord for Pending {
fn cmp(&self, other: &Self) -> std::cmp::Ordering {
self.date
.cmp(&other.date)
.then_with(|| self.seq.cmp(&other.seq))
}
}
impl PartialOrd for Pending {
fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
Some(self.cmp(other))
}
}
#[derive(Default)]
struct Limiter {
flags: HashMap<ObjectId, u8>,
parsed: HashMap<ObjectId, CommitFacts>,
queue: BinaryHeap<Pending>,
seq: u64,
}
impl Limiter {
fn admit(
&mut self,
graph: &Graph<'_>,
id: ObjectId,
uninteresting: bool,
missing: Missing,
buf: &mut Vec<u8>,
) -> Result<()> {
{
let flags = self.flags.entry(id).or_insert(0);
if uninteresting {
*flags |= flag::UNINTERESTING;
}
if *flags & (flag::SEEN | flag::ABSENT) != 0 {
return Ok(());
}
}
let facts = match graph.commit_facts(&id, buf) {
Ok(facts) => facts,
Err(Error::ObjectMissing { .. }) if missing == Missing::Skip => {
*self.flags.entry(id).or_insert(0) |= flag::ABSENT;
return Ok(());
}
Err(err) => return Err(err),
};
*self.flags.entry(id).or_insert(0) |= flag::SEEN;
self.queue.push(Pending {
date: facts.committed_at,
seq: std::cmp::Reverse(self.seq),
id,
});
self.seq += 1;
self.parsed.insert(id, facts);
Ok(())
}
fn mark_parents_uninteresting(&mut self, start: &ObjectId, have_boundary: &HashSet<ObjectId>) {
let Some(facts) = self.parsed.get(start) else {
return;
};
let mut stack: Vec<ObjectId> = facts.parents.clone();
while let Some(mut id) = stack.pop() {
loop {
let flags = self.flags.entry(id).or_insert(0);
if *flags & flag::UNINTERESTING != 0 {
break;
}
*flags |= flag::UNINTERESTING;
if have_boundary.contains(&id) {
break;
}
let Some(facts) = self.parsed.get(&id) else {
break;
};
let Some((first, rest)) = facts.parents.split_first() else {
break;
};
stack.extend_from_slice(rest);
id = *first;
}
}
}
fn is_uninteresting(&self, id: &ObjectId) -> bool {
self.flags
.get(id)
.is_some_and(|flags| flags & flag::UNINTERESTING != 0)
}
fn everybody_uninteresting(&self, cache: &mut Option<ObjectId>) -> bool {
if let Some(id) = *cache {
if !self.is_uninteresting(&id) {
return false;
}
*cache = None;
}
for pending in &self.queue {
if self.is_uninteresting(&pending.id) {
continue;
}
*cache = Some(pending.id);
return false;
}
true
}
fn still_interesting(&self, date: i64, slop: usize, cache: &mut Option<ObjectId>) -> usize {
let Some(head) = self.queue.peek() else {
return 0;
};
if date <= head.date {
return SLOP;
}
if !self.everybody_uninteresting(cache) {
return SLOP;
}
slop.saturating_sub(1)
}
}
#[derive(Debug, Default)]
pub(crate) struct Selection {
pub objects: Vec<ObjectId>,
pub commits: Vec<ObjectId>,
pub client_has: ClientHas,
pub answered_from_bitmap: bool,
pub bitmap_commits_walked: usize,
}
#[derive(Debug)]
pub(crate) enum ClientHas {
Ids(OidSet),
}
impl Default for ClientHas {
fn default() -> Self {
ClientHas::Ids(OidSet::new())
}
}
impl ClientHas {
pub(crate) fn len(&self) -> usize {
match self {
ClientHas::Ids(ids) => ids.len(),
}
}
}
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub(crate) enum Exclusion {
Full,
#[default]
Edges,
}
impl Exclusion {
pub(crate) fn from_env() -> Self {
match std::env::var("GUNNAR_EXCLUSION").as_deref() {
Ok("full") => Exclusion::Full,
_ => Exclusion::Edges,
}
}
}
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub(crate) enum Limiting {
TwoPass,
#[default]
Merged,
}
impl Limiting {
pub(crate) fn from_env() -> Self {
match std::env::var("GUNNAR_LIMIT").as_deref() {
Ok("two-pass") => Limiting::TwoPass,
_ => Limiting::Merged,
}
}
}
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub(crate) enum Bitmaps {
#[default]
Read,
Off,
}
impl Bitmaps {
pub(crate) fn from_env() -> Self {
match std::env::var("GUNNAR_BITMAP").as_deref() {
Ok("off") => Bitmaps::Off,
_ => Bitmaps::Read,
}
}
}
#[derive(Debug, Default, Clone)]
pub(crate) struct SelectOptions<'f> {
pub boundary: HashSet<ObjectId>,
pub have_boundary: HashSet<ObjectId>,
pub filter: Option<&'f Filter>,
pub restrict: Option<&'f HashSet<ObjectId>>,
pub missing: Missing,
pub exclusion: Exclusion,
pub limiting: Limiting,
pub max_blob_bytes: Option<u64>,
pub bitmaps: Bitmaps,
}
impl SelectOptions<'_> {
pub(crate) fn for_serving(limits: &Limits) -> Self {
SelectOptions {
missing: Missing::Fail,
exclusion: Exclusion::from_env(),
limiting: Limiting::from_env(),
bitmaps: Bitmaps::from_env(),
max_blob_bytes: limits.object_bytes.map(|n| n as u64),
..Default::default()
}
}
pub(crate) fn for_mirroring() -> Self {
SelectOptions {
missing: Missing::Fail,
exclusion: Exclusion::from_env(),
limiting: Limiting::from_env(),
bitmaps: Bitmaps::from_env(),
max_blob_bytes: None,
..Default::default()
}
}
}
pub fn objects_to_send(
store: &dyn GitServe,
hash_kind: Kind,
want: &[&[u8]],
have: &[&[u8]],
) -> Result<Vec<Vec<u8>>> {
let tips: Vec<ObjectId> = want.iter().map(|raw| oid(raw)).collect::<Result<_>>()?;
let excludes: Vec<ObjectId> = have.iter().map(|raw| oid(raw)).collect::<Result<_>>()?;
let graph = Graph::new(store, hash_kind);
let selection = select_objects(&graph, &tips, &excludes, &SelectOptions::for_mirroring())?;
Ok(selection
.objects
.into_iter()
.map(|id| id.as_slice().to_vec())
.collect())
}
pub(crate) fn select_objects(
graph: &Graph<'_>,
tips: &[ObjectId],
excludes: &[ObjectId],
options: &SelectOptions<'_>,
) -> Result<Selection> {
if options.bitmaps == Bitmaps::Read {
let _s = crate::stage::enter("select.bitmap");
if let Some(selection) = crate::reachable::select(graph, tips, excludes, options)? {
if let Some(limit) = graph.budget {
if selection.objects.len() > limit {
return Err(Error::WalkBudget { limit });
}
}
return Ok(selection);
}
}
let mut selection = Selection::default();
let mut emitted = OidSet::new();
let _peel = crate::stage::enter("select.peel");
let mut commit_tips = Vec::new();
let mut direct: Vec<(ObjectId, object::Kind)> = Vec::new();
for tip in tips {
let peeled = graph.peel(tip)?;
for tag in peeled.tags {
if emitted.insert(tag) {
selection.objects.push(tag);
}
}
match peeled.kind {
object::Kind::Commit => commit_tips.push(peeled.id),
other => direct.push((peeled.id, other)),
}
}
let mut exclude_commits = Vec::new();
for id in excludes {
if !graph.contains(id)? {
continue;
}
match graph.peel(id) {
Ok(peeled) if peeled.kind == object::Kind::Commit => exclude_commits.push(peeled.id),
Ok(_) => {}
Err(Error::ObjectMissing { .. }) => {}
Err(err) => return Err(err),
}
}
let limiting = match options.exclusion {
Exclusion::Full => Limiting::TwoPass,
Exclusion::Edges => options.limiting,
};
let mut excluded_trees: Vec<ObjectId> = Vec::new();
crate::stage::end(_peel);
let _walk = crate::stage::enter("select.commit_walk");
let (commits, excluded_commits) = match limiting {
Limiting::TwoPass => {
let held = graph.commit_closure(
&exclude_commits,
&HashSet::new(),
&options.have_boundary,
None,
Missing::Skip,
)?;
let excluded: OidSet = held.ids.iter().copied().collect();
excluded_trees = held.trees;
let commits = graph.commit_closure(
&commit_tips,
&excluded,
&options.boundary,
options.restrict,
options.missing,
)?;
(commits, excluded)
}
Limiting::Merged => {
let limited = graph.limited_closure(
&commit_tips,
&exclude_commits,
&options.boundary,
&options.have_boundary,
options.restrict,
options.missing,
)?;
let excluded: OidSet = limited.uninteresting.iter().copied().collect();
let commits = CommitClosure {
ids: limited.ids,
trees: limited.trees,
edges: limited.edges,
};
(commits, excluded)
}
};
let CommitClosure {
ids: commit_ids,
trees: commit_trees,
edges: commit_edges,
} = commits;
for commit in &commit_ids {
if emitted.insert(*commit) {
selection.objects.push(*commit);
}
}
selection.commits = commit_ids;
crate::stage::end(_walk);
let mut excluded_objects = excluded_commits;
{
let _s = crate::stage::enter("select.exclude_trees");
let mut tally = Emit::Tally(0);
let mut buf = Vec::new();
let mut parents = Vec::new();
let roots: Vec<ObjectId> = match options.exclusion {
Exclusion::Full => excluded_trees,
Exclusion::Edges => {
let mut roots = Vec::with_capacity(exclude_commits.len() + commit_edges.len());
for id in exclude_commits.iter().chain(commit_edges.iter()) {
match graph.commit_links(id, &mut buf, &mut parents) {
Ok(tree) => roots.push(tree),
Err(Error::ObjectMissing { .. }) => {}
Err(err) => return Err(err),
}
}
roots
}
};
for tree in &roots {
graph.tree_closure(
*tree,
&mut excluded_objects,
&HashSet::new(),
None,
Missing::Skip,
None,
&mut tally,
)?;
}
}
let _tree = crate::stage::enter("select.tree_closure");
let mut seen = excluded_objects;
let mut carried: Vec<ObjectId> = Vec::with_capacity(emitted.len());
let mut payload = Vec::new();
let mut keep = Emit::Keep(&mut payload);
{
let mut visited = Carried {
seen: &mut seen,
carried: &mut carried,
};
for id in emitted.iter() {
visited.insert(id);
}
for tree in &commit_trees {
graph.tree_closure(
*tree,
&mut visited,
&HashSet::new(),
options.filter,
options.missing,
options.max_blob_bytes,
&mut keep,
)?;
}
for (id, kind) in direct {
match kind {
object::Kind::Tree => graph.tree_closure(
id,
&mut visited,
&HashSet::new(),
options.filter,
options.missing,
options.max_blob_bytes,
&mut keep,
)?,
_ => {
if let (Some(limit), object::Kind::Blob) = (options.max_blob_bytes, kind) {
if let Some((_kind, size)) =
graph.store.header(id.as_bytes()).map_err(backend)?
{
if size > limit {
return Err(Error::ObjectTooLarge {
oid: id,
size,
limit,
});
}
}
}
if visited.insert(&id) {
keep.push(id);
}
}
}
}
}
selection.objects.extend(payload);
for id in &carried {
seen.remove(id);
}
selection.client_has = ClientHas::Ids(seen);
crate::stage::end(_tree);
Ok(selection)
}
pub(crate) fn reachable_closure(graph: &Graph<'_>, tips: &[ObjectId]) -> Result<OidSet> {
let mut seen = OidSet::new();
let mut commit_tips = Vec::new();
let mut direct_trees = Vec::new();
for tip in tips {
if !graph.contains(tip)? {
continue;
}
let peeled = match graph.peel(tip) {
Ok(peeled) => peeled,
Err(Error::ObjectMissing { .. }) => continue,
Err(err) => return Err(err),
};
for tag in peeled.tags {
seen.insert(tag);
}
match peeled.kind {
object::Kind::Commit => commit_tips.push(peeled.id),
object::Kind::Tree => direct_trees.push(peeled.id),
_ => {
seen.insert(peeled.id);
}
}
}
let commits = graph.commit_closure(
&commit_tips,
&HashSet::new(),
&HashSet::new(),
None,
Missing::Skip,
)?;
let mut tally = Emit::Tally(0);
for (commit, tree) in commits.ids.iter().zip(&commits.trees) {
seen.insert(*commit);
graph.tree_closure(
*tree,
&mut seen,
&HashSet::new(),
None,
Missing::Skip,
None,
&mut tally,
)?;
}
for tree in direct_trees {
graph.tree_closure(
tree,
&mut seen,
&HashSet::new(),
None,
Missing::Skip,
None,
&mut tally,
)?;
}
Ok(seen)
}
pub(crate) struct Frontier {
commits: HashSet<ObjectId>,
present: HashSet<ObjectId>,
}
impl Frontier {
pub(crate) fn new(graph: &Graph<'_>, values: &[ObjectId]) -> Result<Self> {
let mut commits = HashSet::new();
let mut present = HashSet::new();
for id in values {
if !graph.contains(id)? {
continue;
}
match graph.peel(id) {
Ok(peeled) if peeled.kind == object::Kind::Commit => {
commits.insert(peeled.id);
}
Ok(peeled) => {
present.insert(peeled.id);
}
Err(Error::ObjectMissing { .. }) => {}
Err(err) => return Err(err),
}
}
for commit in &commits {
present.insert(graph.commit(commit)?.tree);
}
Ok(Frontier { commits, present })
}
}
pub(crate) fn verify_connectivity(
graph: &Graph<'_>,
tips: &[ObjectId],
frontier: &Frontier,
) -> Result<usize> {
let mut tags: HashSet<ObjectId> = HashSet::new();
let mut commit_tips = Vec::new();
let mut direct: Vec<(ObjectId, object::Kind)> = Vec::new();
for tip in tips {
let peeled = graph.peel(tip)?;
tags.extend(peeled.tags);
match peeled.kind {
object::Kind::Commit => commit_tips.push(peeled.id),
other => direct.push((peeled.id, other)),
}
}
let commits = graph.commit_closure(
&commit_tips,
&frontier.commits,
&HashSet::new(),
None,
Missing::Fail,
)?;
let mut seen = OidSet::new();
let mut keep = Emit::Tally(0);
for tree in &commits.trees {
graph.tree_closure(
*tree,
&mut seen,
&frontier.present,
None,
Missing::Fail,
None,
&mut keep,
)?;
}
for (id, kind) in direct {
match kind {
object::Kind::Tree => graph.tree_closure(
id,
&mut seen,
&frontier.present,
None,
Missing::Fail,
None,
&mut keep,
)?,
_ => {
if seen.insert(id) {
keep.push(id);
}
}
}
}
Ok(tags.len() + commits.ids.len() + keep.len())
}
pub(crate) struct Reachability<'g, 'a> {
graph: &'g Graph<'a>,
common: OidSet,
known: HashMap<ObjectId, bool>,
cutoff: i64,
visited: usize,
budget: Option<usize>,
}
impl<'g, 'a> Reachability<'g, 'a> {
pub(crate) fn new(graph: &'g Graph<'a>, common: &[ObjectId], cutoff: i64) -> Self {
Reachability {
graph,
common: common.iter().copied().collect(),
known: HashMap::new(),
cutoff,
visited: 0,
budget: graph.budget,
}
}
pub(crate) fn reaches_common(&mut self, start: ObjectId) -> Result<bool> {
if self.common.is_empty() {
return Ok(false);
}
if let Some(&answer) = self.known.get(&start) {
return Ok(answer);
}
let mut queue: VecDeque<ObjectId> = VecDeque::new();
let mut queued = OidSet::new();
let mut exhausted: Vec<ObjectId> = Vec::new();
queue.push_back(start);
queued.insert(start);
let mut found = false;
while let Some(id) = queue.pop_front() {
if self.common.contains(&id) {
found = true;
break;
}
match self.known.get(&id) {
Some(true) => {
found = true;
break;
}
Some(false) => continue,
None => {}
}
self.visited += 1;
if let Some(limit) = self.budget {
if self.visited > limit {
return Err(Error::WalkBudget { limit });
}
}
self.graph.charge_walk()?;
if !self.graph.contains(&id)? {
self.known.insert(id, false);
continue;
}
let facts = self.graph.commit(&id)?;
if facts.committed_at < self.cutoff {
self.known.insert(id, false);
continue;
}
exhausted.push(id);
for parent in facts.parents {
if queued.insert(parent) {
queue.push_back(parent);
}
}
}
if found {
self.known.insert(start, true);
} else {
for id in exhausted {
self.known.insert(id, false);
}
}
Ok(found)
}
}
#[cfg(test)]
mod tests {
fn walked() -> SelectOptions<'static> {
SelectOptions {
bitmaps: Bitmaps::Off,
..SelectOptions::default()
}
}
use super::*;
use gix_hash::Kind;
use crate::common::Fixture;
#[test]
fn a_fetch_holds_the_clients_object_closure_once() {
const HISTORY: usize = 300;
let fixture = Fixture::build(Kind::Sha1);
let client_tip = fixture.extend(fixture.four, HISTORY, "client");
let server_tip = fixture.extend(client_tip, 1, "server");
let graph = Graph::new(&fixture.git, fixture.kind);
let options = SelectOptions {
exclusion: Exclusion::Full,
..walked()
};
let closure = select_objects(&graph, &[client_tip], &[], &options)
.expect("the client's own closure")
.objects
.len();
assert!(
closure > 3 * HISTORY,
"the fixture is not the shape this test needs: {HISTORY} commits \
should close over more than {} objects, not {closure}",
3 * HISTORY
);
meter::reset_peak();
assert_eq!(
meter::live(),
0,
"a set from the sizing call above outlived it"
);
let selection = select_objects(&graph, &[server_tip], &[client_tip], &options)
.expect("the incremental fetch");
let peak = meter::peak();
assert_eq!(
selection.objects.len(),
3,
"an incremental fetch one commit behind selected {} objects",
selection.objects.len()
);
assert!(
peak >= closure,
"the walk peaked at {peak} live object ids against a client closure \
of {closure}: the exclusion set was never built, so this fetch is \
about to send objects the client already has"
);
let ceiling = closure + 4 * selection.objects.len() + 16;
assert!(
peak <= ceiling,
"the walk held {peak} object ids at once for a client closure of \
{closure}, {:.2}× one copy, ceiling {ceiling}. The exclusion \
closure is being materialised more than once; at ten \
million objects each extra copy is ~450 MB per in-flight fetch",
peak as f64 / closure as f64
);
let full_selected = selection.objects.len();
let vouched = selection.client_has.len();
assert!(
vouched >= closure,
"the walk handed back {vouched} ids to vouch with against a client \
closure of {closure}: the set being carried is not the closure, so \
a thin delta's base would be refused for want of evidence that was \
computed and thrown away"
);
assert_eq!(
meter::live(),
vouched,
"the ids alive after the walk are not exactly the set it handed \
back, so something else is being retained too"
);
drop(selection);
assert_eq!(
meter::live(),
0,
"the client's closure outlived the Selection that owns it"
);
meter::reset_peak();
let edge_selection = select_objects(&graph, &[server_tip], &[client_tip], &walked())
.expect("the incremental fetch, edge exclusion");
let edge_peak = meter::peak();
assert_eq!(
edge_selection.objects.len(),
full_selected,
"edge exclusion selected {} objects where the full closure selected \
{full_selected}: the frontier walk is missing objects the client \
demonstrably has",
edge_selection.objects.len()
);
assert!(
edge_selection.client_has.len() < vouched,
"the voucher carried under edge exclusion is {} ids against the full \
closure's {vouched}, so the frontier walk is handing the pack writer \
the client's whole history after all",
edge_selection.client_has.len()
);
assert!(
edge_peak < peak,
"edge exclusion held {edge_peak} ids at once against the full \
closure's {peak} over a client history of {HISTORY} commits. It is \
walking the client's whole history after all, and the read-side \
win is not there"
);
}
#[test]
fn the_budget_bounds_the_exclusion_walk_not_only_the_commit_walk() {
const HISTORY: usize = 300;
const BUDGET: usize = 400;
let fixture = Fixture::build(Kind::Sha1);
let client_tip = fixture.extend(fixture.four, HISTORY, "client");
let server_tip = fixture.extend(client_tip, 1, "server");
let graph = Graph::new(&fixture.git, fixture.kind);
let commits = graph
.commit_closure(
&[client_tip],
&HashSet::new(),
&HashSet::new(),
None,
Missing::Skip,
)
.expect("the client's commits");
assert!(
commits.ids.len() < BUDGET,
"the {} commits alone reach a budget of {BUDGET}, so this test cannot tell the two walks apart",
commits.ids.len()
);
let graph = Graph::new(&fixture.git, fixture.kind).with_budget(Some(BUDGET));
let outcome = select_objects(
&graph,
&[server_tip],
&[client_tip],
&SelectOptions {
exclusion: Exclusion::Full,
..walked()
},
);
match outcome {
Err(Error::WalkBudget { limit }) => assert_eq!(limit, BUDGET),
Err(other) => panic!("the budget should have stopped this walk, not {other}"),
Ok(selection) => panic!(
"the exclusion walk ran to completion under a budget of \
{BUDGET} and selected {} objects",
selection.objects.len()
),
}
let edges = select_objects(&graph, &[server_tip], &[client_tip], &walked()).expect(
"edge exclusion tripped a budget of 400 on a 305-commit history; \
it is supposed to open up the frontier, not the history",
);
assert_eq!(
edges.objects.len(),
3,
"edge exclusion selected {} objects where the full closure selects 3",
edges.objects.len()
);
}
#[test]
fn a_fetch_does_not_read_the_history_the_client_already_told_it_about() {
use crate::common::Counting;
const HISTORY: usize = 300;
const DAY: usize = 5;
let fixture = Fixture::build(Kind::Sha1);
let client_tip = fixture.extend_dated(fixture.four, HISTORY, "client", |i| {
1_700_200_000 + i as i64
});
let server_tip = fixture.extend_dated(client_tip, DAY, "server", |i| {
1_700_200_000 + (HISTORY + i) as i64
});
let measure = |limiting: Limiting| -> (usize, Vec<ObjectId>) {
let store = Counting::new(fixture.git.inner.clone());
let graph = Graph::new(&store, Kind::Sha1);
let options = SelectOptions {
limiting,
..walked()
};
let selection = select_objects(&graph, &[server_tip], &[client_tip], &options)
.expect("the incremental fetch");
let mut ids = selection.objects;
ids.sort();
(store.touches(), ids)
};
let (two_pass_reads, two_pass_objects) = measure(Limiting::TwoPass);
let (merged_reads, merged_objects) = measure(Limiting::Merged);
assert!(
two_pass_reads > HISTORY,
"the two-pass walk touched the store {two_pass_reads} times over a \
client history of {HISTORY} commits, so it is not walking that \
history and this test cannot tell the two walks apart"
);
assert_eq!(
merged_objects,
two_pass_objects,
"the merged walk selected {} objects where the two-pass walk \
selected {}. Fewer is a broken fetch, not a win",
merged_objects.len(),
two_pass_objects.len()
);
assert!(
merged_objects.len() >= 3 * DAY,
"an incremental fetch of {DAY} commits selected only {} objects, so \
the comparison above is between two empty sets",
merged_objects.len()
);
assert!(
merged_reads * 10 < two_pass_reads,
"the merged walk touched the store {merged_reads} times against the \
two-pass walk's {two_pass_reads}, {:.2}x. It is still enumerating \
the client's history, which is the whole cost this removes",
two_pass_reads as f64 / merged_reads as f64
);
}
#[test]
fn the_two_walks_agree_on_the_shapes_that_could_split_them() {
let fixture = Fixture::build(Kind::Sha1);
let old_line = fixture.extend_dated(fixture.three, 8, "old", |i| 1_700_300_000 + i as i64);
let new_line = fixture.extend_dated(fixture.two, 8, "new", |i| 1_700_400_000 + i as i64);
let merged_tip = crate::common::commit(
&fixture.git,
fixture.kind,
"merged.txt",
"merged",
&[new_line, old_line],
1_700_400_100,
);
let skewed = fixture.extend_dated(new_line, 4, "skew", |_| 1_700_000_000);
let cases: [(&str, &[ObjectId], &[ObjectId], bool); 5] = [
(
"merge across the frontier",
&[merged_tip],
&[old_line],
true,
),
(
"a want that is also a have",
&[merged_tip],
&[merged_tip],
false,
),
(
"one have behind another",
&[merged_tip],
&[new_line, fixture.two],
true,
),
("skewed wants", &[skewed], &[new_line], true),
("no haves at all", &[merged_tip], &[], true),
];
for (name, tips, haves, expect_objects) in cases {
let pick = |limiting: Limiting| -> Vec<ObjectId> {
let graph = Graph::new(&fixture.git, fixture.kind);
let mut ids = select_objects(
&graph,
tips,
haves,
&SelectOptions {
limiting,
..walked()
},
)
.unwrap_or_else(|err| panic!("{name}: {err}"))
.objects;
ids.sort();
ids
};
let merged = pick(Limiting::Merged);
let two_pass = pick(Limiting::TwoPass);
assert_eq!(
!two_pass.is_empty(),
expect_objects,
"{name}: the two-pass walk selected {} objects, which is not the \
shape this case is here to compare",
two_pass.len()
);
assert_eq!(
merged,
two_pass,
"{name}: merged selected {}, two-pass {}",
merged.len(),
two_pass.len()
);
}
}
#[test]
fn a_have_older_than_its_own_parent_still_stops_the_walk_behind_it() {
use crate::common::commit;
use gunnar_store::mem::MemGitServe;
let store = MemGitServe::new(Kind::Sha1);
let a = commit(&store, Kind::Sha1, "a.txt", "a", &[], 1_000);
let b = commit(&store, Kind::Sha1, "b.txt", "b", &[a], 2_000);
let u = commit(&store, Kind::Sha1, "u.txt", "u", &[b], 1_500);
let w = commit(&store, Kind::Sha1, "w.txt", "w", &[b], 3_000);
let graph = Graph::new(&store, Kind::Sha1);
let pick = |limiting: Limiting| -> Vec<ObjectId> {
select_objects(
&graph,
&[w],
&[u],
&SelectOptions {
limiting,
..walked()
},
)
.expect("the skewed fetch")
.commits
};
let merged = pick(Limiting::Merged);
assert!(
!merged.contains(&a),
"the merged walk sent {} commits, including A, which sits behind a \
commit the client holds: the client's side of the graph stopped \
propagating at the first commit already taken for interesting",
merged.len()
);
assert_eq!(
merged,
vec![w, b],
"the merged walk sent {merged:?}; W and B is the documented \
superset on this shape, and anything else has moved it"
);
assert_eq!(
pick(Limiting::TwoPass),
vec![w],
"the two-pass walk is the exact answer here and must stay it, or \
the superset above is being measured against a moving baseline"
);
}
#[test]
fn the_exclusion_opens_up_the_parent_the_two_histories_share() {
use crate::common::commit;
use gunnar_store::mem::MemGitServe;
let store = MemGitServe::new(Kind::Sha1);
let a = commit(&store, Kind::Sha1, "shared.txt", "shared", &[], 1_000);
let c1 = commit(&store, Kind::Sha1, "c1.txt", "c1", &[a], 2_000);
let c2 = commit(&store, Kind::Sha1, "c2.txt", "c2", &[c1], 3_000);
let s1 = commit(&store, Kind::Sha1, "shared.txt", "shared", &[a], 4_000);
assert_ne!(
s1, a,
"the two commits must differ, only their content may not"
);
let graph = Graph::new(&store, Kind::Sha1);
for limiting in [Limiting::Merged, Limiting::TwoPass] {
let selection = select_objects(
&graph,
&[s1],
&[c2],
&SelectOptions {
limiting,
..walked()
},
)
.expect("the branch fetch");
assert_eq!(
selection.objects,
vec![s1],
"{limiting:?}: a fetch sharing a tree and a blob with the client \
packed {} objects, not 1. The edge behind the client's named \
tip was never opened up, so content the client demonstrably \
holds is going back over the wire",
selection.objects.len()
);
}
}
#[test]
fn the_commit_walk_hands_back_each_commits_own_root_tree() {
const HISTORY: usize = 40;
let fixture = Fixture::build(Kind::Sha1);
let tip = fixture.extend(fixture.four, HISTORY, "p010");
let graph = Graph::new(&fixture.git, fixture.kind);
let closure = graph
.commit_closure(
&[tip],
&HashSet::new(),
&HashSet::new(),
None,
Missing::Fail,
)
.expect("the commit closure");
assert_eq!(
closure.ids.len(),
closure.trees.len(),
"the two halves of a CommitClosure must be the same length or the \
pairing is not a pairing"
);
assert!(
closure.ids.len() > HISTORY,
"the fixture is not the shape this test needs: {HISTORY} commits on \
top of the base should close over more than {HISTORY} commits, not \
{}. A zip over a shorter list would assert nothing.",
closure.ids.len()
);
for (i, (id, tree)) in closure.ids.iter().zip(&closure.trees).enumerate() {
let independent = graph
.commit(id)
.expect("re-reading a commit the walk just read")
.tree;
assert_eq!(
*tree, independent,
"commit {i} ({id}) was paired with tree {tree}, but its own root \
tree is {independent}"
);
}
}
fn ceiling_for(objects: usize) -> usize {
objects * crate::WALK_BYTES_PER_OBJECT
}
#[test]
fn the_ceiling_charges_what_the_walk_holds_not_what_it_sends() {
const HISTORY: usize = 300;
let fixture = Fixture::build(Kind::Sha1);
let tip = fixture.extend(fixture.four, HISTORY, "held");
let blobless = Filter::parse("blob:none").expect("blob:none parses");
let options = |filter| SelectOptions { filter, ..walked() };
let (sent, held) = {
let graph = Graph::new(&fixture.git, fixture.kind);
let sent = select_objects(&graph, &[tip], &[], &options(Some(&blobless)))
.expect("the blob:none clone")
.objects
.len();
(sent, meter::peak())
};
assert!(
held > sent + 100,
"the fixture is not the shape this test needs: a blob:none clone \
sent {sent} objects and held {held}, and the two have to be far \
enough apart that a ceiling can sit between them"
);
let graph =
Graph::new(&fixture.git, fixture.kind).with_walk_limit(Some(ceiling_for(sent)), None);
let refused = select_objects(&graph, &[tip], &[], &options(Some(&blobless)))
.expect_err("a ceiling the size of what this fetch sends must refuse it");
match refused {
Error::WalkTooLarge { held: bytes, limit } => {
assert_eq!(
limit,
ceiling_for(sent),
"the refusal named another ceiling"
);
assert!(
bytes > limit,
"the refusal said it was holding {bytes} bytes against a \
ceiling of {limit}, which is not over it"
);
}
other => panic!(
"a blob:none fetch over the byte ceiling failed with {other}, not WalkTooLarge — \
so the ceiling is not what stopped it and this test is measuring another fault"
),
}
let graph =
Graph::new(&fixture.git, fixture.kind).with_walk_limit(Some(ceiling_for(held)), None);
let served = select_objects(&graph, &[tip], &[], &options(Some(&blobless)))
.expect("a ceiling above what this fetch holds must serve it");
assert_eq!(
served.objects.len(),
sent,
"the served arm sent {} objects where the unbounded one sent {sent}: a ceiling that \
changes the ANSWER rather than refusing is worse than the fault being measured",
served.objects.len()
);
}
#[test]
fn no_ceiling_means_no_ceiling() {
let fixture = Fixture::build(Kind::Sha1);
let tip = fixture.extend(fixture.four, 50, "off");
let graph = Graph::new(&fixture.git, fixture.kind).with_walk_limit(None, None);
let all = select_objects(&graph, &[tip], &[], &walked())
.expect("an unbounded walk serves")
.objects
.len();
assert!(
all > 50,
"the fixture holds {all} objects, too few to mean anything"
);
}
#[test]
fn the_pool_refuses_a_walk_because_of_what_another_is_holding() {
let history = crate::WalkPool::FLOOR_BYTES / crate::WALK_BYTES_PER_OBJECT / 3 + 500;
let fixture = Fixture::build(Kind::Sha1);
let tip = fixture.extend(fixture.four, history, "pool");
let held = {
let graph = Graph::new(&fixture.git, fixture.kind);
select_objects(&graph, &[tip], &[], &walked()).expect("the whole clone");
meter::peak()
};
assert!(
ceiling_for(held) >= crate::WalkPool::FLOOR_BYTES,
"this fixture holds {held} ids, {} bytes, which is under the pool's floor of {} \
— the walk would never touch the pool and every arm below would pass against \
nothing",
ceiling_for(held),
crate::WalkPool::FLOOR_BYTES
);
let reservation = ceiling_for(held).next_multiple_of(crate::WalkPool::CHUNK_BYTES);
let pool = crate::WalkPool::new(reservation);
let outer =
Graph::new(&fixture.git, fixture.kind).with_walk_limit(None, Some(pool.clone()));
let holding = select_objects(&outer, &[tip], &[], &walked())
.expect("the first walk fits the empty pool");
assert!(!holding.objects.is_empty());
assert_eq!(
pool.in_flight_bytes(),
reservation,
"the first walk reserved {} of the {reservation} it holds, so there is nothing for \
the second to be refused by",
pool.in_flight_bytes()
);
let inner =
Graph::new(&fixture.git, fixture.kind).with_walk_limit(None, Some(pool.clone()));
let refused =
select_objects(&inner, &[tip], &[], &walked()).expect_err("a full pool must refuse");
match refused {
Error::WalkPoolFull { limit, .. } => assert_eq!(limit, reservation),
other => panic!(
"a walk against a full pool failed with {other}, not WalkPoolFull — a client \
cannot tell a busy server from a request it must never send again"
),
}
assert_eq!(pool.refusals(), 1, "the refusal was not counted");
drop(inner);
drop(outer);
drop(holding);
assert_eq!(
pool.in_flight_bytes(),
0,
"the pool did not get its bytes back, so this server refuses everything after enough \
clones"
);
let again =
Graph::new(&fixture.git, fixture.kind).with_walk_limit(None, Some(pool.clone()));
select_objects(&again, &[tip], &[], &walked())
.expect("the same walk must serve once the first has finished");
}
#[test]
fn walk_pool_never_goes_negative() {
let pool = crate::WalkPool::new(1024);
pool.release(4096);
assert_eq!(pool.in_flight_bytes(), 0);
pool.try_reserve(1024)
.expect("a pool at zero has all of itself");
assert_eq!(pool.in_flight_bytes(), 1024);
}
#[test]
fn a_pool_for_this_machine_reads_the_machine() {
let pool = crate::WalkPool::for_this_machine();
assert!(
pool.limit_bytes() > 0,
"a pool of zero bytes refuses every transfer"
);
if let Ok(text) = std::fs::read_to_string("/proc/meminfo") {
let kb: usize = text
.lines()
.find_map(|l| l.strip_prefix("MemTotal:")?.trim().strip_suffix(" kB"))
.and_then(|v| v.trim().parse().ok())
.expect("MemTotal is in /proc/meminfo");
assert_eq!(
pool.limit_bytes(),
kb * 1024 / 2,
"the pool is not half of this machine's memory, so it is a constant wearing a \
measurement's name"
);
}
}
}