use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use anyhow::{Result, anyhow, bail};
use crate::exploded::{NoSink, PayloadSink};
use crate::forest::{DeltaForest, EntrySet};
use crate::index_layout::{IndexEntry, ObjType};
use crate::memguard::MemWatchdog;
use crate::object::{GitHashKind, GitObjectKind};
use crate::pack_walk::{DeltaBase, PackEntry, PackWalk, walk};
use crate::progress::{IndexProgress, Phase};
pub trait BaseSource: Sync {
fn content(&self, oid: &[u8]) -> Result<Option<(GitObjectKind, Vec<u8>)>>;
}
pub struct NoBases;
impl BaseSource for NoBases {
fn content(&self, _oid: &[u8]) -> Result<Option<(GitObjectKind, Vec<u8>)>> {
Ok(None)
}
}
#[derive(Debug, Clone)]
pub struct Resolved {
pub oid: Vec<u8>,
pub offset: u64,
pub len: u64,
pub stored_type: ObjType,
pub kind: GitObjectKind,
pub uncompressed_size: u64,
pub delta_base: u64,
}
impl Resolved {
pub fn index_entry(&self) -> IndexEntry {
IndexEntry {
oid: self.oid.clone(),
offset: self.offset,
len: self.len,
obj_type: self.stored_type,
uncompressed_size: self.uncompressed_size,
delta_base: self.delta_base,
}
}
}
#[derive(Debug)]
pub struct ResolveError;
impl ResolveError {
pub fn missing_base(oid: &[u8]) -> anyhow::Error {
anyhow!(
"this pack is thin: it deltas against {} which is not in it, and no base source could \
produce that object's content — §14's exploded table does not hold it and it could \
not be re-derived from the verbatim packs either, so this repository genuinely does \
not have that object. The pack is refused by name rather than half-resolved",
hex::encode(oid)
)
}
pub fn cancelled() -> anyhow::Error {
anyhow!("index work cancelled: the repository is being deleted")
}
pub fn out_of_memory(reason: &str, available: u64, budget: u64, done: u64, total: u64) -> anyhow::Error {
anyhow!(
"this box cannot index this pack: the memory watchdog gave up ({reason}) with {} \
available against a {} floor, after {done} of {total} objects. Everything droppable \
was dropped and the floor was still breached — what holds the memory is the \
resolver's own per-object state, which is proportional to the pack. Give the box \
more RAM, lower ZNIPPY_GIT_MEM_BUDGET_PCT, or push a smaller pack. The pack's bytes \
are durable and untouched; nothing partial was indexed.",
crate::memguard::fmt_bytes(available),
crate::memguard::fmt_bytes(budget)
)
}
}
pub const DEFAULT_BASE_BUDGET_PCT: u64 = 25;
pub const DEFAULT_ROW_BATCH: usize = 262_144;
pub const DEFAULT_LOOKAHEAD_BYTES: u64 = 32 << 20;
pub const MAX_WORKERS: usize = 256;
#[derive(Clone)]
pub struct ResolveLimits<'a> {
pub base_budget: u64,
pub row_batch: usize,
pub workers: usize,
pub lookahead_bytes: u64,
pub check_every: u64,
pub watchdog: Option<Arc<MemWatchdog>>,
pub cancel: Option<&'a AtomicBool>,
pub progress: Option<&'a IndexProgress>,
pub label: &'a str,
pub on_pause: Option<&'a (dyn Fn() -> u64 + Sync)>,
}
impl<'a> ResolveLimits<'a> {
pub fn none() -> Self {
Self {
base_budget: 0,
row_batch: DEFAULT_ROW_BATCH,
workers: 1,
lookahead_bytes: DEFAULT_LOOKAHEAD_BYTES,
check_every: crate::memguard::CHECK_EVERY_OBJECTS,
watchdog: None,
cancel: None,
progress: None,
label: "",
on_pause: None,
}
}
pub fn for_indexer(
label: &'a str,
cancel: &'a AtomicBool,
progress: &'a IndexProgress,
watchdog: Arc<MemWatchdog>,
on_pause: &'a (dyn Fn() -> u64 + Sync),
) -> Self {
Self {
base_budget: base_budget_from_env(),
row_batch: row_batch_from_env(),
workers: workers_from_env(),
lookahead_bytes: lookahead_from_env(),
check_every: crate::memguard::CHECK_EVERY_OBJECTS,
watchdog: Some(watchdog),
cancel: Some(cancel),
progress: Some(progress),
label,
on_pause: Some(on_pause),
}
}
}
pub fn base_budget_from_env() -> u64 {
use std::sync::OnceLock;
static B: OnceLock<u64> = OnceLock::new();
*B.get_or_init(|| {
if let Some(b) = crate::arms::read_env(crate::arms::ENV_BASE_BUDGET_BYTES)
.and_then(|v| v.parse::<u64>().ok())
{
return b;
}
let total = crate::memguard::mem_total();
if total == 0 { 1 << 30 } else { total / 100 * DEFAULT_BASE_BUDGET_PCT }
})
}
pub fn row_batch_from_env() -> usize {
use std::sync::OnceLock;
static B: OnceLock<usize> = OnceLock::new();
*B.get_or_init(|| {
crate::arms::read_env(crate::arms::ENV_INDEX_BATCH_ROWS)
.and_then(|v| v.parse::<usize>().ok())
.filter(|n| *n > 0)
.unwrap_or(DEFAULT_ROW_BATCH)
})
}
pub fn workers_from_env() -> usize {
use std::sync::OnceLock;
static W: OnceLock<Option<usize>> = OnceLock::new();
let from_env = *W.get_or_init(|| {
crate::arms::read_env(crate::arms::ENV_RESOLVE_WORKERS)
.and_then(|v| v.parse::<usize>().ok())
.filter(|n| *n > 0)
});
from_env
.unwrap_or_else(|| match RESOLVE_WORKERS_DEFAULT.load(Ordering::Relaxed) {
0 => std::thread::available_parallelism().map(|x| x.get()).unwrap_or(1),
n => n,
})
.clamp(1, MAX_WORKERS)
}
static RESOLVE_WORKERS_DEFAULT: AtomicUsize = AtomicUsize::new(0);
pub fn set_resolve_workers_default(workers: usize) {
RESOLVE_WORKERS_DEFAULT.store(workers, Ordering::Relaxed);
}
pub fn lookahead_from_env() -> u64 {
use std::sync::OnceLock;
static L: OnceLock<Option<u64>> = OnceLock::new();
let from_env = *L.get_or_init(|| {
crate::arms::read_env(crate::arms::ENV_RESOLVE_LOOKAHEAD_BYTES)
.and_then(|v| v.parse::<u64>().ok())
.filter(|n| *n > 0)
});
from_env.unwrap_or_else(|| match RESOLVE_LOOKAHEAD_DEFAULT.load(Ordering::Relaxed) {
0 => DEFAULT_LOOKAHEAD_BYTES,
n => n,
})
}
static RESOLVE_LOOKAHEAD_DEFAULT: AtomicU64 = AtomicU64::new(0);
pub fn set_resolve_lookahead_default(bytes: u64) {
RESOLVE_LOOKAHEAD_DEFAULT.store(bytes, Ordering::Relaxed);
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct ResolveReport {
pub rows: u64,
pub reinflations: u64,
pub evictions: u64,
pub watchdog_drops: u64,
pub peak_base_bytes: u64,
pub pauses: u64,
pub paused_ms: u64,
pub batches: u64,
pub stalls: u64,
pub workers: u32,
pub shed: u64,
pub gave_up: bool,
}
impl ResolveReport {
pub fn absorb(&mut self, r: ResolveReport) {
self.rows += r.rows;
self.reinflations += r.reinflations;
self.evictions += r.evictions;
self.watchdog_drops += r.watchdog_drops;
self.peak_base_bytes = self.peak_base_bytes.max(r.peak_base_bytes);
self.pauses += r.pauses;
self.paused_ms += r.paused_ms;
self.batches += r.batches;
self.stalls += r.stalls;
self.shed += r.shed;
self.workers = self.workers.max(r.workers);
self.gave_up |= r.gave_up;
}
}
pub fn resolve(
pack: &[u8],
hash: GitHashKind,
archive_offset: u64,
bases: &dyn BaseSource,
) -> Result<Vec<Resolved>> {
let w = walk(pack, hash.oid_len())?;
resolve_walked(pack, &w, hash, archive_offset, bases, &NoSink)
}
pub fn resolve_walked(
pack: &[u8],
w: &PackWalk,
hash: GitHashKind,
archive_offset: u64,
bases: &dyn BaseSource,
sink: &dyn PayloadSink,
) -> Result<Vec<Resolved>> {
let mut out: Vec<Resolved> = Vec::with_capacity(w.entries.len());
let limits = ResolveLimits {
row_batch: w.entries.len().max(1),
..ResolveLimits::none()
};
resolve_streaming(pack, w, hash, archive_offset, bases, sink, limits, &mut |rows| {
out.extend(rows.drain(..));
Ok(())
})?;
out.sort_unstable_by_key(|r| r.offset);
Ok(out)
}
#[allow(clippy::too_many_arguments)]
pub fn resolve_streaming(
pack: &[u8],
w: &PackWalk,
hash: GitHashKind,
archive_offset: u64,
bases: &dyn BaseSource,
sink: &dyn PayloadSink,
limits: ResolveLimits<'_>,
rows: &mut (dyn FnMut(&mut Vec<Resolved>) -> Result<()> + Send),
) -> Result<ResolveReport> {
let mut r = Resolver::new(pack, w, hash, archive_offset, bases, sink, limits)?;
r.run(rows)?;
Ok(r.report)
}
const HANDOVER_ROWS: usize = 4096;
const HANDOVER_BYTES: u64 = 1 << 20;
const SCRATCH_KEEP_BYTES: usize = 1 << 20;
const ALLOC_OVERHEAD: usize = 32;
const DRAIN_RUN: usize = 8192;
struct Frame {
entry: u32,
kind: GitObjectKind,
payload: Vec<u8>,
evicted: bool,
next_child: u32,
}
struct Emit {
kind: GitObjectKind,
row: Resolved,
payload: Option<Vec<u8>>,
}
impl Emit {
fn weight(&self) -> u64 {
(std::mem::size_of::<Emit>()
+ ALLOC_OVERHEAD
+ self.row.oid.len()
+ self.payload.as_ref().map_or(0, |p| p.len() + ALLOC_OVERHEAD)) as u64
}
}
#[derive(Default)]
struct Scratch {
unit: usize,
delta: Vec<u8>,
leaf: Vec<u8>,
stack: Vec<Frame>,
out: Vec<Emit>,
out_bytes: u64,
found_refs: Vec<(Vec<u8>, u64)>,
}
std::thread_local! {
static SCRATCH: std::cell::RefCell<Scratch> = std::cell::RefCell::new(Scratch::default());
}
#[derive(Default)]
struct RootOut {
rows: Vec<Emit>,
done: bool,
}
#[derive(Default)]
struct Pending {
map: std::collections::BTreeMap<usize, RootOut>,
bytes: u64,
in_flight: std::collections::BTreeSet<usize>,
parked: usize,
}
struct Drain<'r> {
cursor: usize,
batch: Vec<Resolved>,
rows: &'r mut (dyn FnMut(&mut Vec<Resolved>) -> Result<()> + Send),
}
struct RoundBases<'r, 'w, 'a> {
external: &'r HashMap<&'w [u8], (GitObjectKind, Vec<u8>)>,
inner: &'a dyn BaseSource,
}
impl BaseSource for RoundBases<'_, '_, '_> {
fn content(&self, oid: &[u8]) -> Result<Option<(GitObjectKind, Vec<u8>)>> {
if let Some((k, p)) = self.external.get(oid) {
return Ok(Some((*k, p.clone())));
}
self.inner.content(oid)
}
}
struct Round<'p, 'w, 'a, 'r> {
pack: &'p [u8],
w: &'w PackWalk,
forest: &'r DeltaForest,
hash: GitHashKind,
oid_len: usize,
archive_offset: u64,
bases: &'a dyn BaseSource,
sink: &'a dyn PayloadSink,
limits: &'r ResolveLimits<'a>,
ref_wanted: &'r std::collections::HashSet<&'w [u8]>,
oid_to_offset: &'r HashMap<Vec<u8>, u64>,
external: &'r HashMap<&'w [u8], (GitObjectKind, Vec<u8>)>,
roots: &'r [u32],
in_round: &'r EntrySet,
workers: usize,
lookahead: u64,
pend: std::sync::Mutex<Pending>,
cv: std::sync::Condvar,
drain: std::sync::Mutex<Drain<'r>>,
pause_gate: std::sync::Mutex<()>,
stop: AtomicBool,
err: std::sync::Mutex<Option<(usize, anyhow::Error)>>,
since_check: std::sync::atomic::AtomicU64,
rows: std::sync::atomic::AtomicU64,
reinflations: std::sync::atomic::AtomicU64,
evictions: std::sync::atomic::AtomicU64,
watchdog_drops: std::sync::atomic::AtomicU64,
live_path: std::sync::atomic::AtomicU64,
peak_path: std::sync::atomic::AtomicU64,
pauses: std::sync::atomic::AtomicU64,
paused_ms: std::sync::atomic::AtomicU64,
batches: std::sync::atomic::AtomicU64,
stalls: std::sync::atomic::AtomicU64,
shed: std::sync::atomic::AtomicU64,
gave_up: AtomicBool,
found_refs: std::sync::Mutex<Vec<(Vec<u8>, u64)>>,
}
impl<'p, 'w, 'a, 'r> Round<'p, 'w, 'a, 'r> {
fn run_root(&self, ri: usize, s: &mut Scratch) -> Result<()> {
if !s.stack.is_empty() {
self.drop_path(s);
s.stack.clear();
}
s.unit = ri;
let root = self.roots[ri];
let r = (|| {
let (kind, payload) = self.root_content(root)?;
self.emit(root, kind, &payload, s)?;
if self.forest.children(root).is_empty() {
drop(payload);
} else {
self.push_frame(root, kind, payload, s);
self.walk(s)?;
}
Ok(())
})();
self.flush_scratch(s, true)?;
r
}
fn root_content(&self, root: u32) -> Result<(GitObjectKind, Vec<u8>)> {
let e = &self.w.entries[root as usize];
match &e.delta_base {
DeltaBase::None => {
let kind = whole_kind(e.obj_type)?;
let mut out = Vec::with_capacity(e.uncompressed_size as usize);
inflate_into(self.pack, e, self.oid_len, &mut out)?;
Ok((kind, out))
}
DeltaBase::Ref(oid) => {
let (kind, base) = match self.oid_to_offset.get(oid.as_slice()).copied() {
Some(o) => {
let be = self
.w
.entries
.binary_search_by_key(&o, |x| x.offset)
.map_err(|_| {
anyhow!("no entry starts at offset {o} — the pack is corrupt")
})?;
self.materialize(be as u32)?
}
None => match self.external.get(oid.as_slice()) {
Some((k, p)) => (*k, p.clone()),
None => return Err(ResolveError::missing_base(oid)),
},
};
let mut delta = Vec::with_capacity(e.uncompressed_size as usize);
inflate_into(self.pack, e, self.oid_len, &mut delta)?;
let mut out = Vec::new();
apply_delta_into(&base, &delta, &mut out)?;
Ok((kind, out))
}
DeltaBase::Offset(_) => bail!(
"the entry at offset {} deltas against an offset, so it is not a root of a \
resolve round — the forest is inconsistent with the walk",
e.offset
),
}
}
fn walk(&self, s: &mut Scratch) -> Result<()> {
while !s.stack.is_empty() {
let fi = s.stack.len() - 1;
let entry = s.stack[fi].entry;
let nkids = self.forest.children(entry).len();
let nc = s.stack[fi].next_child as usize;
if nc == nkids {
self.pop_frame(s);
continue;
}
let c = self.forest.children(entry)[nc];
s.stack[fi].next_child += 1;
let last = nc + 1 == nkids;
let kind = s.stack[fi].kind;
let base: Vec<u8> = if s.stack[fi].evicted {
let (_, p) = self.materialize(entry)?;
self.reinflations.fetch_add(1, Ordering::Relaxed);
if let Some(pr) = self.limits.progress {
pr.add_reinflations(1);
}
if !last {
let n = p.len() as u64;
let budget = self.limits.base_budget;
if budget == 0 || self.live_path.load(Ordering::Relaxed) + n <= budget {
s.stack[fi].payload = p.clone();
s.stack[fi].evicted = false;
let live = self.live_path.fetch_add(n, Ordering::Relaxed) + n;
self.peak_path.fetch_max(live, Ordering::Relaxed);
}
}
p
} else if last {
let p = std::mem::take(&mut s.stack[fi].payload);
self.live_path
.fetch_sub(p.len() as u64, Ordering::Relaxed);
s.stack[fi].evicted = true;
p
} else {
s.stack[fi].payload.clone()
};
let ce = &self.w.entries[c as usize];
let mut delta = std::mem::take(&mut s.delta);
let mut leaf = std::mem::take(&mut s.leaf);
let leafy = self.forest.children(c).is_empty();
let mut carried: Option<Vec<u8>> = None;
let r = (|| -> Result<()> {
inflate_into(self.pack, ce, self.oid_len, &mut delta)?;
if leafy {
apply_delta_into(&base, &delta, &mut leaf)?;
drop(base);
self.emit(c, kind, &leaf, s)
} else {
let mut out = Vec::with_capacity(ce.uncompressed_size as usize);
apply_delta_into(&base, &delta, &mut out)?;
drop(base);
self.emit(c, kind, &out, s)?;
carried = Some(out);
Ok(())
}
})();
delta.shrink_to(SCRATCH_KEEP_BYTES);
leaf.shrink_to(SCRATCH_KEEP_BYTES);
s.delta = delta;
s.leaf = leaf;
r?;
if let Some(out) = carried {
self.push_frame(c, kind, out, s);
}
}
Ok(())
}
fn push_frame(&self, entry: u32, kind: GitObjectKind, payload: Vec<u8>, s: &mut Scratch) {
let len = payload.len() as u64;
s.stack.push(Frame {
entry,
kind,
payload,
evicted: false,
next_child: 0,
});
self.live_path.fetch_add(len, Ordering::Relaxed);
self.enforce_budget(s);
self.peak_path
.fetch_max(self.live_path.load(Ordering::Relaxed), Ordering::Relaxed);
}
fn pop_frame(&self, s: &mut Scratch) {
if let Some(f) = s.stack.pop()
&& !f.evicted
{
self.live_path
.fetch_sub(f.payload.len() as u64, Ordering::Relaxed);
}
}
fn enforce_budget(&self, s: &mut Scratch) {
let budget = self.limits.base_budget;
if budget == 0 {
return;
}
let mut i = 0usize;
while i + 1 < s.stack.len() && self.live_path.load(Ordering::Relaxed) > budget {
if !s.stack[i].evicted {
let n = s.stack[i].payload.len() as u64;
s.stack[i].payload = Vec::new();
s.stack[i].evicted = true;
self.live_path.fetch_sub(n, Ordering::Relaxed);
self.evictions.fetch_add(1, Ordering::Relaxed);
}
i += 1;
}
}
fn drop_path(&self, s: &mut Scratch) -> u64 {
let mut freed = 0u64;
for f in s.stack.iter_mut() {
if !f.evicted {
freed += f.payload.len() as u64;
f.payload = Vec::new();
f.evicted = true;
}
}
self.live_path.fetch_sub(freed, Ordering::Relaxed);
freed
}
fn materialize(&self, entry: u32) -> Result<(GitObjectKind, Vec<u8>)> {
let rb = RoundBases {
external: self.external,
inner: self.bases,
};
let of = |oid: &[u8]| self.oid_to_offset.get(oid).copied();
materialize_entry(self.pack, self.w, entry as usize, self.oid_len, &of, &rb)
}
fn emit(&self, index: u32, kind: GitObjectKind, payload: &[u8], s: &mut Scratch) -> Result<()> {
let e = &self.w.entries[index as usize];
let oid = self.hash.oid_of_object(kind, payload);
if self.ref_wanted.contains(&oid.as_slice()) {
s.found_refs.push((oid.clone(), e.offset));
}
let want = self.sink.wants(kind);
let em = Emit {
kind,
row: Resolved {
oid,
offset: e.offset + self.archive_offset,
len: e.len,
stored_type: e.obj_type,
kind,
uncompressed_size: payload.len() as u64,
delta_base: match &e.delta_base {
DeltaBase::Offset(o) => o + self.archive_offset,
_ => 0,
},
},
payload: if want { Some(payload.to_vec()) } else { None },
};
s.out_bytes += em.weight();
s.out.push(em);
self.rows.fetch_add(1, Ordering::Relaxed);
if let Some(p) = self.limits.progress {
p.object_done(e.len);
}
if s.out.len() >= HANDOVER_ROWS || s.out_bytes >= HANDOVER_BYTES {
self.flush_scratch(s, false)?;
}
let n = self.since_check.fetch_add(1, Ordering::Relaxed) + 1;
if n >= self.limits.check_every {
self.since_check.store(0, Ordering::Relaxed);
self.checkpoint(s)?;
}
Ok(())
}
fn flush_scratch(&self, s: &mut Scratch, done: bool) -> Result<()> {
if !s.out.is_empty() || done {
let mut bytes = 0u64;
for em in s.out.iter() {
bytes += em.weight();
}
let unit = s.unit;
{
let mut p = self.lock_pend()?;
let e = p.map.entry(unit).or_default();
if e.rows.is_empty() {
e.rows = std::mem::take(&mut s.out);
} else {
e.rows.append(&mut s.out);
}
e.done |= done;
p.bytes += bytes;
}
s.out.clear();
s.out.shrink_to(HANDOVER_ROWS);
s.out_bytes = 0;
}
self.try_drain(false)
}
fn lock_pend(&self) -> Result<std::sync::MutexGuard<'_, Pending>> {
self.pend
.lock()
.map_err(|_| anyhow!("the resolver's emit buffer is poisoned"))
}
fn try_drain(&self, force: bool) -> Result<()> {
let mut guard = match self.drain.try_lock() {
Ok(g) => g,
Err(std::sync::TryLockError::WouldBlock) => {
if !force {
return Ok(());
}
self.drain
.lock()
.map_err(|_| anyhow!("the resolver's row drain is poisoned"))?
}
Err(std::sync::TryLockError::Poisoned(_)) => {
bail!("the resolver's row drain is poisoned")
}
};
let d = &mut *guard;
loop {
let mut run: Vec<Vec<Emit>> = Vec::new();
let mut held = 0usize;
let mut moved = false;
{
let mut p = self.lock_pend()?;
while d.cursor < self.roots.len() {
let (rows, done) = match p.map.get_mut(&d.cursor) {
Some(ro) => (std::mem::take(&mut ro.rows), ro.done),
None => break,
};
if !rows.is_empty() {
let n: u64 = rows.iter().map(Emit::weight).sum();
p.bytes = p.bytes.saturating_sub(n);
held += rows.len();
run.push(rows);
moved = true;
}
if !done {
break;
}
p.map.remove(&d.cursor);
d.cursor += 1;
moved = true;
if held >= DRAIN_RUN {
break;
}
}
if moved {
self.cv.notify_all();
}
}
if !moved {
return Ok(());
}
for chunk in run {
for em in chunk {
if let Some(pay) = &em.payload {
self.sink.explode(&em.row.oid, em.kind, pay)?;
}
d.batch.push(em.row);
if d.batch.len() >= self.limits.row_batch {
(d.rows)(&mut d.batch)?;
self.batches.fetch_add(1, Ordering::Relaxed);
d.batch.clear();
}
}
}
}
}
fn finish(&self) -> Result<()> {
self.try_drain(true)?;
let mut d = self
.drain
.lock()
.map_err(|_| anyhow!("the resolver's row drain is poisoned"))?;
if !d.batch.is_empty() {
let d = &mut *d;
(d.rows)(&mut d.batch)?;
self.batches.fetch_add(1, Ordering::Relaxed);
d.batch.clear();
}
let p = self.lock_pend()?;
if !p.map.is_empty() || d.cursor != self.roots.len() {
bail!(
"the resolver finished a round at root {} of {} with {} roots still unemitted — \
the round's roots and the walk disagree",
d.cursor,
self.roots.len(),
p.map.len()
);
}
Ok(())
}
fn backpressure(&self, ri: usize) -> Result<()> {
self.try_drain(false)?;
let mut p = self.lock_pend()?;
while p.bytes > self.lookahead
&& !self.stop.load(Ordering::Relaxed)
&& p.in_flight.first().is_some_and(|lowest| *lowest < ri)
{
self.stalls.fetch_add(1, Ordering::Relaxed);
p = self
.cv
.wait(p)
.map_err(|_| anyhow!("the resolver's emit buffer is poisoned"))?;
}
Ok(())
}
fn enter(&self, ri: usize) -> Result<()> {
self.lock_pend()?.in_flight.insert(ri);
Ok(())
}
fn leave(&self, ri: usize) {
if let Ok(mut p) = self.pend.lock() {
p.in_flight.remove(&ri);
}
self.cv.notify_all();
}
fn checkpoint(&self, s: &mut Scratch) -> Result<()> {
if self.stop.load(Ordering::Relaxed) {
bail!("this pack is being failed by another worker");
}
if self.limits.cancel.is_some_and(|c| c.load(Ordering::Relaxed)) {
return Err(ResolveError::cancelled());
}
let Some(dog) = self.limits.watchdog.clone() else {
return Ok(());
};
if dog.breached().is_none() {
return Ok(());
}
let mut freed = self.drop_path(s);
{
let mut p = self.lock_pend()?;
let mut waited = false;
loop {
if self.stop.load(Ordering::Relaxed) {
break;
}
let lowest = p.in_flight.first().copied();
if lowest.is_some_and(|l| l < s.unit) {
if !waited {
waited = true;
p.parked += 1;
self.shed.fetch_add(1, Ordering::Relaxed);
if let Some(pr) = self.limits.progress {
pr.add_shed();
pr.set_phase(Phase::Paused);
}
self.cv.notify_all();
}
p = self
.cv
.wait(p)
.map_err(|_| anyhow!("the resolver's emit buffer is poisoned"))?;
continue;
}
if p.in_flight.len() > p.parked + 1 {
if waited {
waited = false;
p.parked -= 1;
}
drop(p);
if let Some(pr) = self.limits.progress {
pr.set_phase(Phase::Resolve);
}
return Ok(());
}
break;
}
if waited {
p.parked -= 1;
}
drop(p);
if waited {
if let Some(pr) = self.limits.progress {
pr.set_phase(Phase::Resolve);
}
if self.stop.load(Ordering::Relaxed) {
bail!("this pack is being failed by another worker");
}
if dog.breached().is_none() {
return Ok(());
}
}
}
if let Some(p) = self.limits.progress {
p.set_phase(Phase::Paused);
p.add_pause();
}
let gate = self
.pause_gate
.lock()
.map_err(|_| anyhow!("the resolver's pause gate is poisoned"))?;
if self.stop.load(Ordering::Relaxed) {
drop(gate);
bail!("this pack is being failed by another worker");
}
if dog.breached().is_none() {
drop(gate);
if let Some(p) = self.limits.progress {
p.set_phase(Phase::Resolve);
}
return Ok(());
}
let (cancel, on_pause, label) = (self.limits.cancel, self.limits.on_pause, self.limits.label);
let mut flush = || -> Result<u64> {
let mut f = std::mem::take(&mut freed);
self.watchdog_drops.fetch_add(1, Ordering::Relaxed);
self.sink.flush()?;
self.flush_scratch(s, false)?;
self.try_drain(true)?;
if let Some(g) = on_pause {
f = f.saturating_add(g());
}
Ok(f)
};
let paused = dog.pause_until_ok(label, cancel, &mut flush);
let ms = match &paused {
Ok(p) => p.ms,
Err(_) => 0,
};
self.pauses.fetch_add(1, Ordering::Relaxed);
self.paused_ms.fetch_add(ms, Ordering::Relaxed);
if let Some(p) = self.limits.progress {
p.set_phase(Phase::Resolve);
}
drop(gate);
let paused = paused?;
if self.limits.cancel.is_some_and(|c| c.load(Ordering::Relaxed)) {
return Err(ResolveError::cancelled());
}
if paused.end.gave_up() {
self.gave_up.store(true, Ordering::Relaxed);
self.stop.store(true, Ordering::Relaxed);
if self.pend.lock().is_ok() {
self.cv.notify_all();
}
let (done, total) = match self.limits.progress {
Some(p) => {
let sn = p.snapshot();
(sn.objects_done, sn.objects_total)
}
None => (self.rows.load(Ordering::Relaxed), self.w.entries.len() as u64),
};
return Err(ResolveError::out_of_memory(
paused.end.as_str(),
paused.available,
paused.budget,
done,
total,
));
}
Ok(())
}
fn fail(&self, ri: usize, e: anyhow::Error) {
self.stop.store(true, Ordering::Relaxed);
if let Ok(mut g) = self.err.lock() {
let take = match &*g {
Some((i, _)) => ri < *i,
None => true,
};
if take {
*g = Some((ri, e));
}
}
if self.pend.lock().is_ok() {
self.cv.notify_all();
}
}
fn take_err(&self) -> Option<anyhow::Error> {
self.err.lock().ok().and_then(|mut g| g.take()).map(|(_, e)| e)
}
fn report(&self) -> ResolveReport {
ResolveReport {
rows: self.rows.load(Ordering::Relaxed),
reinflations: self.reinflations.load(Ordering::Relaxed),
evictions: self.evictions.load(Ordering::Relaxed),
watchdog_drops: self.watchdog_drops.load(Ordering::Relaxed),
peak_base_bytes: self.peak_path.load(Ordering::Relaxed),
pauses: self.pauses.load(Ordering::Relaxed),
paused_ms: self.paused_ms.load(Ordering::Relaxed),
batches: self.batches.load(Ordering::Relaxed),
stalls: self.stalls.load(Ordering::Relaxed),
shed: self.shed.load(Ordering::Relaxed),
workers: self.workers as u32,
gave_up: self.gave_up.load(Ordering::Relaxed),
}
}
}
struct Resolver<'p, 'w, 'a> {
pack: &'p [u8],
w: &'w PackWalk,
hash: GitHashKind,
oid_len: usize,
archive_offset: u64,
bases: &'a dyn BaseSource,
sink: &'a dyn PayloadSink,
limits: ResolveLimits<'a>,
forest: DeltaForest,
ref_wanted: std::collections::HashSet<&'w [u8]>,
oid_to_offset: HashMap<Vec<u8>, u64>,
report: ResolveReport,
}
impl<'p, 'w, 'a> Resolver<'p, 'w, 'a> {
fn new(
pack: &'p [u8],
w: &'w PackWalk,
hash: GitHashKind,
archive_offset: u64,
bases: &'a dyn BaseSource,
sink: &'a dyn PayloadSink,
limits: ResolveLimits<'a>,
) -> Result<Self> {
let forest = DeltaForest::build(w)?;
let mut ref_wanted: std::collections::HashSet<&'w [u8]> = std::collections::HashSet::new();
for e in w.entries.iter() {
if let DeltaBase::Ref(oid) = &e.delta_base {
ref_wanted.insert(oid.as_slice());
}
}
Ok(Self {
pack,
w,
hash,
oid_len: hash.oid_len(),
archive_offset,
bases,
sink,
forest,
ref_wanted,
oid_to_offset: HashMap::new(),
report: ResolveReport::default(),
limits,
})
}
fn run(&mut self, rows: &mut (dyn FnMut(&mut Vec<Resolved>) -> Result<()> + Send)) -> Result<()> {
let n = self.w.entries.len();
if let Some(p) = self.limits.progress {
p.set_total(n as u64);
p.set_phase(Phase::Resolve);
}
if n == 0 {
return Ok(());
}
let w = self.w;
let mut done = EntrySet::new(n);
let mut round = 0usize;
while done.len() < n {
let mut roots: Vec<u32> = if round == 0 {
self.forest.whole_roots().to_vec()
} else {
Vec::new()
};
let mut external: HashMap<&'w [u8], (GitObjectKind, Vec<u8>)> = HashMap::new();
for &r in self.forest.ref_roots() {
if done.contains(r as usize) {
continue;
}
let DeltaBase::Ref(oid) = &w.entries[r as usize].delta_base else {
continue;
};
if self.oid_to_offset.contains_key(oid.as_slice())
|| external.contains_key(oid.as_slice())
{
roots.push(r);
continue;
}
if let Some(c) = self.bases.content(oid)? {
external.insert(oid.as_slice(), c);
roots.push(r);
}
}
if roots.is_empty() {
for &r in self.forest.ref_roots() {
if !done.contains(r as usize)
&& let DeltaBase::Ref(oid) = &w.entries[r as usize].delta_base
{
return Err(ResolveError::missing_base(oid));
}
}
bail!(
"{} of {n} entries could not be resolved and none of them deltas against an \
oid — the pack is corrupt",
n - done.len()
);
}
roots.sort_unstable();
let in_round = self.forest.closure_of(&roots);
let (rep, found) = self.run_round(&roots, &in_round, &external, rows)?;
self.report.absorb(rep);
for (oid, off) in found {
self.oid_to_offset.insert(oid, off);
}
done.union_with(&in_round);
round += 1;
}
Ok(())
}
fn run_round(
&self,
roots: &[u32],
in_round: &EntrySet,
external: &HashMap<&'w [u8], (GitObjectKind, Vec<u8>)>,
rows: &mut (dyn FnMut(&mut Vec<Resolved>) -> Result<()> + Send),
) -> Result<(ResolveReport, Vec<(Vec<u8>, u64)>)> {
let workers = self
.limits
.workers
.max(1)
.min(roots.len().max(1))
.min(MAX_WORKERS);
let round = Round {
pack: self.pack,
w: self.w,
forest: &self.forest,
hash: self.hash,
oid_len: self.oid_len,
archive_offset: self.archive_offset,
bases: self.bases,
sink: self.sink,
limits: &self.limits,
ref_wanted: &self.ref_wanted,
oid_to_offset: &self.oid_to_offset,
external,
roots,
in_round,
workers,
lookahead: self.limits.lookahead_bytes.max(1 << 20),
pend: std::sync::Mutex::new(Pending::default()),
cv: std::sync::Condvar::new(),
drain: std::sync::Mutex::new(Drain {
cursor: 0,
batch: Vec::with_capacity(self.limits.row_batch.min(1 << 20)),
rows,
}),
pause_gate: std::sync::Mutex::new(()),
stop: AtomicBool::new(false),
err: std::sync::Mutex::new(None),
since_check: std::sync::atomic::AtomicU64::new(0),
rows: std::sync::atomic::AtomicU64::new(0),
reinflations: std::sync::atomic::AtomicU64::new(0),
evictions: std::sync::atomic::AtomicU64::new(0),
watchdog_drops: std::sync::atomic::AtomicU64::new(0),
live_path: std::sync::atomic::AtomicU64::new(0),
peak_path: std::sync::atomic::AtomicU64::new(0),
pauses: std::sync::atomic::AtomicU64::new(0),
paused_ms: std::sync::atomic::AtomicU64::new(0),
batches: std::sync::atomic::AtomicU64::new(0),
stalls: std::sync::atomic::AtomicU64::new(0),
shed: std::sync::atomic::AtomicU64::new(0),
gave_up: AtomicBool::new(false),
found_refs: std::sync::Mutex::new(Vec::new()),
};
gatling_for_each_unit(roots.len(), workers, |ri| {
if round.stop.load(Ordering::Relaxed) {
return;
}
let r = SCRATCH.with(|cell| -> Result<()> {
let mut s = cell.borrow_mut();
round.enter(ri)?;
let r = (|| {
round.backpressure(ri)?;
if round.stop.load(Ordering::Relaxed) {
return Ok(());
}
round.run_root(ri, &mut s)
})();
if !s.found_refs.is_empty()
&& let Ok(mut g) = round.found_refs.lock()
{
g.append(&mut s.found_refs);
}
round.leave(ri);
r
});
if let Err(e) = r {
round.fail(ri, e);
}
});
if let Some(e) = round.take_err() {
return Err(e);
}
round.finish()?;
let mut found = round
.found_refs
.lock()
.map_err(|_| anyhow!("the resolver's ref table is poisoned"))?
.split_off(0);
found.sort_unstable();
found.dedup();
Ok((round.report(), found))
}
}
fn gatling_for_each_unit<F: Fn(usize) + Sync>(n: usize, workers: usize, f: F) {
znippy_zoomies::gatling_forkjoin::gatling_run(n, workers, f);
}
pub fn materialize_entry(
pack: &[u8],
w: &PackWalk,
entry: usize,
oid_len: usize,
offset_of_oid: &dyn Fn(&[u8]) -> Option<u64>,
bases: &dyn BaseSource,
) -> Result<(GitObjectKind, Vec<u8>)> {
let e = w
.entries
.get(entry)
.ok_or_else(|| anyhow!("entry {entry} is past the walk ({} entries)", w.entries.len()))?;
match &e.delta_base {
DeltaBase::None => {
let mut out = Vec::with_capacity(e.uncompressed_size as usize);
inflate_into(pack, e, oid_len, &mut out)?;
Ok((whole_kind(e.obj_type)?, out))
}
DeltaBase::Offset(o) => {
let base_entry = w
.entries
.binary_search_by_key(o, |x| x.offset)
.map_err(|_| anyhow!("no entry starts at offset {o} — the pack is corrupt"))?;
let (kind, base) = materialize_entry(pack, w, base_entry, oid_len, offset_of_oid, bases)?;
let mut delta = Vec::with_capacity(e.uncompressed_size as usize);
inflate_into(pack, e, oid_len, &mut delta)?;
let mut out = Vec::new();
apply_delta_into(&base, &delta, &mut out)?;
Ok((kind, out))
}
DeltaBase::Ref(oid) => {
let (kind, base) = match offset_of_oid(oid) {
Some(o) => {
let base_entry = w
.entries
.binary_search_by_key(&o, |x| x.offset)
.map_err(|_| anyhow!("no entry starts at offset {o} — the pack is corrupt"))?;
materialize_entry(pack, w, base_entry, oid_len, offset_of_oid, bases)?
}
None => bases.content(oid)?.ok_or_else(|| ResolveError::missing_base(oid))?,
};
let mut delta = Vec::with_capacity(e.uncompressed_size as usize);
inflate_into(pack, e, oid_len, &mut delta)?;
let mut out = Vec::new();
apply_delta_into(&base, &delta, &mut out)?;
Ok((kind, out))
}
}
}
pub trait ArchiveBytes {
fn at(&self, at: u64, len: u64) -> Result<std::borrow::Cow<'_, [u8]>>;
}
pub struct SliceArchive<'a>(pub &'a [u8]);
impl ArchiveBytes for SliceArchive<'_> {
fn at(&self, at: u64, len: u64) -> Result<std::borrow::Cow<'_, [u8]>> {
let start = usize::try_from(at).unwrap_or(usize::MAX).min(self.0.len());
let end = start.saturating_add(usize::try_from(len).unwrap_or(usize::MAX)).min(self.0.len());
Ok(std::borrow::Cow::Borrowed(&self.0[start..end]))
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct ReadCost {
pub depth: u32,
pub inflated: u64,
pub external_base: bool,
}
pub const MAX_DELTA_LINKS: usize = 100;
pub struct CycleGuard {
links: usize,
seen: std::collections::HashSet<u64>,
}
impl Default for CycleGuard {
fn default() -> Self {
Self::new()
}
}
impl CycleGuard {
pub fn new() -> Self {
Self {
links: 0,
seen: std::collections::HashSet::new(),
}
}
pub fn step(&mut self, at: u64) -> Result<()> {
self.links += 1;
if self.links <= MAX_DELTA_LINKS {
return Ok(());
}
if !self.seen.insert(at) {
bail!(
"the delta chain returns to archive offset {at} after {} links — the pack's \
back-references form a cycle, so the entry cannot be reconstructed. This is a \
corrupt pack and not a deep one: depth alone is never refused here, because a \
verbatim archive grows one link per push (see MAX_DELTA_LINKS)",
self.links
);
}
Ok(())
}
pub fn links(&self) -> usize {
self.links
}
}
const HEADER_PROBE: u64 = 64;
const INFLATE_WINDOW: u64 = 256 << 10;
struct Link {
entry_at: u64,
obj_type: ObjType,
size: u64,
data_at: u64,
}
pub fn materialize_at(
archive: &dyn ArchiveBytes,
at: u64,
oid_len: usize,
offset_of_oid: &dyn Fn(&[u8]) -> Option<u64>,
bases: &dyn BaseSource,
) -> Result<(GitObjectKind, Vec<u8>, ReadCost)> {
materialize_at_with(archive, at, oid_len, offset_of_oid, bases, None)
}
pub struct ChainCache {
map: std::collections::HashMap<u64, (GitObjectKind, Vec<u8>)>,
order: std::collections::VecDeque<u64>,
bytes: usize,
cap: usize,
hits: u64,
misses: u64,
}
impl ChainCache {
pub fn new(cap_bytes: usize) -> Self {
Self {
map: std::collections::HashMap::new(),
order: std::collections::VecDeque::new(),
bytes: 0,
cap: cap_bytes,
hits: 0,
misses: 0,
}
}
fn get(&mut self, at: u64) -> Option<(GitObjectKind, &[u8])> {
match self.map.get(&at) {
Some((k, p)) => {
self.hits += 1;
Some((*k, p.as_slice()))
}
None => {
self.misses += 1;
None
}
}
}
fn put(&mut self, at: u64, kind: GitObjectKind, payload: &[u8]) {
if payload.len() > self.cap || self.map.contains_key(&at) {
return;
}
while self.bytes + payload.len() > self.cap {
let Some(old) = self.order.pop_front() else { break };
if let Some((_, p)) = self.map.remove(&old) {
self.bytes -= p.len();
}
}
self.bytes += payload.len();
self.order.push_back(at);
self.map.insert(at, (kind, payload.to_vec()));
}
pub fn stats(&self) -> (u64, u64) {
(self.hits, self.misses)
}
pub fn bytes(&self) -> usize {
self.bytes
}
}
pub fn materialize_at_with(
archive: &dyn ArchiveBytes,
at: u64,
oid_len: usize,
offset_of_oid: &dyn Fn(&[u8]) -> Option<u64>,
bases: &dyn BaseSource,
mut cache: Option<&mut ChainCache>,
) -> Result<(GitObjectKind, Vec<u8>, ReadCost)> {
let mut chain: Vec<Link> = Vec::new();
let mut external: Option<Vec<u8>> = None;
let mut memo_hit: Option<(GitObjectKind, Vec<u8>)> = None;
let mut cursor = at;
let mut guard = CycleGuard::new();
loop {
guard.step(cursor)?;
if let Some(c) = cache.as_deref_mut() {
if let Some((k, p)) = c.get(cursor) {
memo_hit = Some((k, p.to_vec()));
break;
}
}
let head = archive.at(cursor, HEADER_PROBE)?;
if head.is_empty() {
bail!("the archive holds no entry at offset {cursor}");
}
let (t, size, n) = crate::pack_walk::type_and_size_of(&head)?;
match t {
ObjType::Commit | ObjType::Tree | ObjType::Blob | ObjType::Tag => {
chain.push(Link { entry_at: cursor, obj_type: t, size, data_at: cursor + n as u64 });
break;
}
ObjType::OfsDelta => {
let (distance, d) = crate::pack_walk::ofs_distance_of(head.get(n..).unwrap_or(&[]))?;
let base = cursor.checked_sub(distance).ok_or_else(|| {
anyhow!(
"the ofs-delta at archive offset {cursor} names a base {distance} bytes \
back, which is before the start of the archive"
)
})?;
chain.push(Link { entry_at: cursor, obj_type: t, size, data_at: cursor + (n + d) as u64 });
cursor = base;
}
ObjType::RefDelta => {
let base_oid = head
.get(n..n + oid_len)
.ok_or_else(|| {
anyhow!(
"the ref-delta entry at archive offset {cursor} has no base oid after \
its header"
)
})?
.to_vec();
chain.push(Link { entry_at: cursor, obj_type: t, size, data_at: cursor + (n + oid_len) as u64 });
match offset_of_oid(&base_oid) {
Some(o) => cursor = o,
None => {
external = Some(base_oid);
break;
}
}
}
}
}
let external_base = external.is_some();
let (kind, mut base) = match (memo_hit, external) {
(Some(hit), _) => hit,
(None, Some(oid)) => bases
.content(&oid)?
.ok_or_else(|| ResolveError::missing_base(&oid))?,
(None, None) => {
let last = chain.pop().expect("a chain that ended has a last link");
let mut out = Vec::new();
inflate_from(archive, last.data_at, last.size, &mut out)?;
let kind = whole_kind(last.obj_type)?;
if let Some(c) = cache.as_deref_mut() {
c.put(last.entry_at, kind, &out);
}
(kind, out)
}
};
let depth = chain.len() as u32;
let mut delta = Vec::new();
let mut out = Vec::new();
while let Some(link) = chain.pop() {
inflate_from(archive, link.data_at, link.size, &mut delta)?;
out.clear();
apply_delta_into(&base, &delta, &mut out)?;
std::mem::swap(&mut base, &mut out);
if let Some(c) = cache.as_deref_mut() {
c.put(link.entry_at, kind, &base);
}
}
let cost = ReadCost {
depth,
inflated: base.len() as u64,
external_base,
};
Ok((kind, base, cost))
}
fn inflate_from(
archive: &dyn ArchiveBytes,
data_at: u64,
expected: u64,
out: &mut Vec<u8>,
) -> Result<()> {
let want = usize::try_from(expected).map_err(|_| anyhow!("object size overflows usize"))?;
out.clear();
out.reserve(want);
let mut d = flate2::Decompress::new(true);
let mut pos = data_at;
while out.len() < want {
let chunk = archive.at(pos, INFLATE_WINDOW)?;
if chunk.is_empty() {
bail!("the zlib stream at {data_at} runs past the end of the archive");
}
let before = d.total_in();
let status = d
.decompress_vec(&chunk, out, flate2::FlushDecompress::None)
.map_err(|e| anyhow!("inflating the entry at {data_at}: {e}"))?;
let consumed = d.total_in() - before;
pos += consumed;
if status == flate2::Status::StreamEnd {
break;
}
if consumed == 0 {
bail!(
"the zlib stream at {data_at} stalled after {} of {expected} bytes",
out.len()
);
}
}
if out.len() as u64 != expected {
bail!(
"the entry at {data_at} inflated to {} bytes, its header declared {expected}",
out.len()
);
}
Ok(())
}
fn whole_kind(t: ObjType) -> Result<GitObjectKind> {
Ok(match t {
ObjType::Commit => GitObjectKind::Commit,
ObjType::Tree => GitObjectKind::Tree,
ObjType::Blob => GitObjectKind::Blob,
ObjType::Tag => GitObjectKind::Tag,
ObjType::OfsDelta | ObjType::RefDelta => {
bail!("a delta entry has no type of its own — its base's type is the answer")
}
})
}
fn inflate_into(pack: &[u8], e: &PackEntry, oid_len: usize, out: &mut Vec<u8>) -> Result<()> {
let start = e.offset as usize;
let end = start + e.len as usize;
if end > pack.len() {
bail!("entry at {start} runs past the end of the pack");
}
let header = header_len(&pack[start..end], e, oid_len)?;
out.clear();
out.reserve(e.uncompressed_size as usize);
let mut d = flate2::Decompress::new(true);
d.decompress_vec(&pack[start + header..end], out, flate2::FlushDecompress::Finish)
.map_err(|err| anyhow!("inflating the entry at {start}: {err}"))?;
if out.len() as u64 != e.uncompressed_size {
bail!(
"the entry at {start} inflated to {} bytes, the walk measured {}",
out.len(),
e.uncompressed_size
);
}
Ok(())
}
#[cfg(test)]
fn inflate(pack: &[u8], e: &PackEntry, oid_len: usize) -> Result<Vec<u8>> {
let mut out = Vec::new();
inflate_into(pack, e, oid_len, &mut out)?;
Ok(out)
}
fn header_len(entry: &[u8], e: &PackEntry, oid_len: usize) -> Result<usize> {
let mut i = 0usize;
let mut cont = true;
while cont {
let b = *entry
.get(i)
.ok_or_else(|| anyhow!("the type/size header runs off the entry"))?;
cont = b & 0x80 != 0;
i += 1;
}
match e.obj_type {
ObjType::OfsDelta => {
let mut cont = true;
while cont {
let b = *entry
.get(i)
.ok_or_else(|| anyhow!("the ofs-delta distance runs off the entry"))?;
cont = b & 0x80 != 0;
i += 1;
}
}
ObjType::RefDelta => i += oid_len,
_ => {}
}
if i >= entry.len() {
bail!("the entry's header consumes all of it, leaving no stream");
}
Ok(i)
}
#[cfg(test)]
fn apply_delta(base: &[u8], delta: &[u8]) -> Result<Vec<u8>> {
let mut out = Vec::new();
apply_delta_into(base, delta, &mut out)?;
Ok(out)
}
fn apply_delta_into(base: &[u8], delta: &[u8], out: &mut Vec<u8>) -> Result<()> {
let mut i = 0usize;
let base_size = delta_varint(delta, &mut i)?;
if base_size != base.len() as u64 {
bail!(
"the delta expects a base of {base_size} bytes, its base is {}",
base.len()
);
}
let target_size = delta_varint(delta, &mut i)?;
out.clear();
out.reserve(target_size as usize);
while i < delta.len() {
let op = delta[i];
i += 1;
if op & 0x80 != 0 {
let mut off = 0u64;
for bit in 0..4 {
if op & (1 << bit) != 0 {
off |= u64::from(*delta.get(i).ok_or_else(|| anyhow!("delta ends mid-copy"))?)
<< (bit * 8);
i += 1;
}
}
let mut size = 0u64;
for bit in 0..3 {
if op & (0x10 << bit) != 0 {
size |= u64::from(*delta.get(i).ok_or_else(|| anyhow!("delta ends mid-copy"))?)
<< (bit * 8);
i += 1;
}
}
if size == 0 {
size = 0x1_0000;
}
let from = off as usize;
let to = from
.checked_add(size as usize)
.ok_or_else(|| anyhow!("a delta copy range overflows"))?;
if to > base.len() {
bail!(
"a delta copies base[{from}..{to}] out of a {}-byte base",
base.len()
);
}
out.extend_from_slice(&base[from..to]);
} else {
let n = op as usize;
if n == 0 {
bail!("a delta carries a zero-length insert instruction");
}
let end = i
.checked_add(n)
.ok_or_else(|| anyhow!("a delta insert overflows"))?;
if end > delta.len() {
bail!("a delta insert of {n} bytes runs off the end");
}
out.extend_from_slice(&delta[i..end]);
i = end;
}
}
if out.len() as u64 != target_size {
bail!(
"the delta declares a {target_size}-byte result and produced {}",
out.len()
);
}
Ok(())
}
fn delta_varint(b: &[u8], i: &mut usize) -> Result<u64> {
let mut v = 0u64;
let mut shift = 0u32;
loop {
let byte = *b
.get(*i)
.ok_or_else(|| anyhow!("a delta size varint runs off the end"))?;
*i += 1;
if shift >= 64 {
bail!("a delta size varint is longer than a u64");
}
v |= u64::from(byte & 0x7f) << shift;
shift += 7;
if byte & 0x80 == 0 {
return Ok(v);
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::object::canonical;
use std::path::{Path, PathBuf};
struct Named<'a>(&'a str);
impl Named<'_> {
fn display(&self) -> &str {
self.0
}
}
fn read_idx(bytes: &[u8]) -> Vec<(Vec<u8>, u64)> {
assert_eq!(&bytes[0..4], b"\xfftOc", "not an idx v2");
assert_eq!(u32::from_be_bytes(bytes[4..8].try_into().unwrap()), 2);
let fanout_end = 8 + 256 * 4;
let n = u32::from_be_bytes(bytes[fanout_end - 4..fanout_end].try_into().unwrap()) as usize;
let oids = fanout_end;
let crcs = oids + n * 20;
let offs = crcs + n * 4;
let big = offs + n * 4;
let mut out = Vec::with_capacity(n);
for i in 0..n {
let oid = bytes[oids + i * 20..oids + i * 20 + 20].to_vec();
let raw = u32::from_be_bytes(bytes[offs + i * 4..offs + i * 4 + 4].try_into().unwrap());
let offset = if raw & 0x8000_0000 != 0 {
let j = (raw & 0x7fff_ffff) as usize;
u64::from_be_bytes(bytes[big + j * 8..big + j * 8 + 8].try_into().unwrap())
} else {
u64::from(raw)
};
out.push((oid, offset));
}
out
}
fn real_pairs(cap: usize) -> Vec<(PathBuf, PathBuf)> {
let mut out = Vec::new();
let Ok(repos) = std::fs::read_dir(std::path::PathBuf::from(std::env::var_os("HOME").unwrap_or_default()).join("git")) else {
return out;
};
for repo in repos.flatten() {
let dir = repo.path().join(".git/objects/pack");
let Ok(files) = std::fs::read_dir(&dir) else {
continue;
};
for f in files.flatten() {
let p = f.path();
if p.extension().is_some_and(|e| e == "pack")
&& f.metadata().map(|m| m.len() < 32 << 20).unwrap_or(false)
{
let idx = p.with_extension("idx");
if idx.exists() {
out.push((p, idx));
if out.len() >= cap {
return out;
}
}
}
}
}
out
}
fn sources() -> Vec<(String, Vec<u8>, Vec<u8>)> {
let mut out = Vec::new();
if std::process::Command::new("git")
.arg("--version")
.output()
.is_ok()
{
let pack = crate::pack_fixture::corpus().pack.clone();
let scratch = crate::store::tests::tmpdir("resolve-idx-oracle");
let idx = crate::git_oracle::git_idx_for(&scratch, "corpus.git", &pack)
.expect("stock git indexes the generated corpus");
out.push(("the generated corpus".to_string(), pack, idx));
}
for (pack_path, idx_path) in real_pairs(6) {
let (Ok(pack), Ok(idx)) = (std::fs::read(&pack_path), std::fs::read(&idx_path)) else {
continue;
};
out.push((pack_path.display().to_string(), pack, idx));
}
out
}
#[test]
fn every_oid_we_compute_is_the_oid_git_wrote_in_its_idx() {
let pairs = sources();
assert!(
!pairs.is_empty(),
"neither stock git nor a real pack is available, so this guard has nothing to \
compare and must not pass silently"
);
let mut objects = 0usize;
let mut full = 0usize;
let mut thin = 0usize;
for (pack_path, pack, idx_bytes) in pairs {
let pack_path = Named(&pack_path);
let idx = read_idx(&idx_bytes);
let ours = match resolve(&pack, GitHashKind::Sha1, 0, &NoBases) {
Ok(r) => r,
Err(e) => {
assert!(
e.to_string().contains("thin"),
"{}: {e}",
pack_path.display()
);
eprintln!("{}: THIN, external base — skipped", pack_path.display());
thin += 1;
continue;
}
};
full += 1;
assert_eq!(
ours.len(),
idx.len(),
"{}: we resolved {} objects, git indexed {}",
pack_path.display(),
ours.len(),
idx.len()
);
let theirs: HashMap<Vec<u8>, u64> = idx.into_iter().collect();
for r in &ours {
match theirs.get(&r.oid) {
Some(&off) => assert_eq!(
off,
r.offset,
"{}: {} is at {} in git's idx and we put it at {}",
pack_path.display(),
hex::encode(&r.oid),
off,
r.offset
),
None => panic!(
"{}: we computed {} at offset {}, which git's idx does not contain — the \
resolution is wrong",
pack_path.display(),
hex::encode(&r.oid),
r.offset
),
}
}
eprintln!("{}: {} objects agree with git", pack_path.display(), ours.len());
objects += ours.len();
}
eprintln!("{objects} objects over {full} packs agree with git's own .idx; {thin} thin");
assert!(full > 0, "every pack compared was thin — nothing was compared");
assert!(objects > 1000, "only {objects} objects compared");
}
#[test]
fn a_delta_carries_its_stored_type_and_its_resolved_type_and_size() {
let pack = crate::pack_fixture::corpus().pack.clone();
let pack_path = Named("the generated corpus");
let ours = resolve(&pack, GitHashKind::Sha1, 0, &NoBases).unwrap();
let stream_size: HashMap<u64, u64> = walk(&pack, 20)
.unwrap()
.entries
.iter()
.map(|e| (e.offset, e.uncompressed_size))
.collect();
let deltas: Vec<&Resolved> = ours
.iter()
.filter(|r| matches!(r.stored_type, ObjType::OfsDelta | ObjType::RefDelta))
.collect();
assert!(
!deltas.is_empty(),
"{} carries no deltas, so it proves nothing about resolution",
pack_path.display()
);
let by_offset: HashMap<u64, &Resolved> = ours.iter().map(|r| (r.offset, r)).collect();
let mut non_blob = 0usize;
for d in &deltas {
assert!(
matches!(
d.kind,
GitObjectKind::Blob
| GitObjectKind::Tree
| GitObjectKind::Commit
| GitObjectKind::Tag
),
"a resolved type must be a real git type"
);
if d.kind != GitObjectKind::Blob {
non_blob += 1;
}
if d.delta_base != 0 {
let base = by_offset[&d.delta_base];
assert_eq!(
d.kind,
base.kind,
"a delta resolves to its base's type: {} vs base {}",
hex::encode(&d.oid),
hex::encode(&base.oid)
);
assert!(
base.offset < d.offset,
"an ofs-delta base is always earlier in the pack"
);
}
}
assert!(
non_blob > 0,
"{} deltas and not one resolved to a tree/commit/tag — the resolved type is not being \
taken from the base",
deltas.len()
);
let differing = deltas
.iter()
.filter(|d| stream_size[&d.offset] != d.uncompressed_size)
.count();
assert!(
differing > 0,
"not one delta's resolved size differs from its delta-stream size, which cannot be \
true of a real pack — the resolved size is being copied from the header"
);
eprintln!(
"{}: {} deltas, {non_blob} non-blob, {differing} with a resolved size unlike the \
stream size",
pack_path.display(),
deltas.len()
);
}
#[test]
fn a_thin_pack_is_refused_by_name_and_says_what_is_missing() {
let base_oid = vec![0x5a; 20];
let mut pack = b"PACK".to_vec();
pack.extend_from_slice(&2u32.to_be_bytes());
pack.extend_from_slice(&1u32.to_be_bytes());
let delta = vec![0x04, 0x04, 0x90, 0x04];
pack.push(0x70 | (delta.len() as u8 & 0x0f)); pack.extend_from_slice(&base_oid);
pack.extend_from_slice(&{
use std::io::Write;
let mut e = flate2::write::ZlibEncoder::new(Vec::new(), flate2::Compression::default());
e.write_all(&delta).unwrap();
e.finish().unwrap()
});
pack.extend_from_slice(&[0u8; 20]);
let err = resolve(&pack, GitHashKind::Sha1, 0, &NoBases)
.expect_err("a thin pack with no base source cannot resolve");
let msg = err.to_string();
assert!(msg.contains(&hex::encode(&base_oid)), "names the base: {msg}");
assert!(msg.contains("§14"), "says which decision is missing: {msg}");
struct One(Vec<u8>);
impl BaseSource for One {
fn content(&self, oid: &[u8]) -> Result<Option<(GitObjectKind, Vec<u8>)>> {
Ok((oid == self.0.as_slice()).then(|| (GitObjectKind::Blob, b"abcd".to_vec())))
}
}
let ok = resolve(&pack, GitHashKind::Sha1, 0, &One(base_oid)).expect("with a base it works");
assert_eq!(ok.len(), 1);
assert_eq!(ok[0].kind, GitObjectKind::Blob);
assert_eq!(ok[0].uncompressed_size, 4);
assert_eq!(
ok[0].oid,
GitHashKind::Sha1.oid_of(&canonical(GitObjectKind::Blob, b"abcd")),
"the resolved object is the base copied whole, so it hashes as `abcd`"
);
}
#[test]
fn the_delta_base_column_survives_all_three_arms_over_the_corpus() {
use crate::index_layout::{
FourTables, IndexEntry, ObjType, ObjectIndex, OneTableFourColumns, PackedPayload,
};
use crate::read_stack::{ObjectReadStack, RebuildTriggers};
let pack = crate::pack_fixture::corpus().pack.clone();
const AT: u64 = 4096;
let rows = resolve(&pack, GitHashKind::Sha1, AT, &NoBases).unwrap();
let entries: Vec<IndexEntry> = rows.iter().map(|r| r.index_entry()).collect();
let a = FourTables::build(&entries).unwrap();
let b = OneTableFourColumns::build(&entries).unwrap();
let c = PackedPayload::build(&entries).unwrap();
let stack = ObjectReadStack::<OneTableFourColumns>::in_memory(RebuildTriggers::manual())
.unwrap();
stack.append(&entries).unwrap();
stack.rebuild().unwrap();
let extents: HashMap<u64, u64> = entries
.iter()
.map(|e| {
let r = a.lookup(&e.oid).expect("a stored oid resolves");
(r.offset, r.len)
})
.collect();
assert_eq!(extents.len(), entries.len(), "two objects share an offset");
let named: [(&str, &dyn ObjectIndex); 4] =
[("FourTables", &a), ("OneTableFourColumns", &b), ("PackedPayload", &c), ("ObjectReadStack", &stack)];
let mut ofs_deltas = 0usize;
for (name, idx) in named {
let mut with_base = 0usize;
for e in &entries {
let row = idx.lookup(&e.oid).expect("a stored oid resolves");
assert_eq!(
row,
a.lookup(&e.oid).unwrap(),
"{name} disagrees with FourTables on {}",
hex::encode(&e.oid)
);
if row.obj_type != ObjType::OfsDelta {
continue;
}
with_base += 1;
assert_ne!(
row.delta_base, 0,
"{name}: {} is an ofs-delta with no recorded base",
hex::encode(&e.oid)
);
assert_ne!(
row.delta_base,
row.offset,
"{name}: {} records base {} which is its own offset, not its base's",
hex::encode(&e.oid),
row.delta_base
);
assert!(
row.delta_base >= AT,
"{name}: base {} is below the archive offset {AT} — it was never rebased",
row.delta_base
);
let base_len = extents.get(&row.delta_base).unwrap_or_else(|| {
panic!(
"{name}: {} records base offset {}, which starts no object in this index",
hex::encode(&e.oid),
row.delta_base
)
});
assert!(
row.delta_base + base_len <= row.offset,
"{name}: the base at {} (+{base_len}) overlaps the delta at {}",
row.delta_base,
row.offset
);
}
assert!(
with_base > 0,
"{name}: {} ofs-deltas resolved and 0 of them carry a base offset — the column \
is not being written",
entries
.iter()
.filter(|e| e.obj_type == ObjType::OfsDelta)
.count()
);
ofs_deltas = with_base;
}
assert!(
ofs_deltas > 100,
"the corpus carries only {ofs_deltas} ofs-deltas — too few to prove anything"
);
eprintln!(
"{ofs_deltas} ofs-delta base offsets located their base entry in all four layouts"
);
}
#[test]
fn a_base_many_entries_share_outlives_all_of_them() {
let mut checked = 0usize;
let mut widest = 0usize;
let mut corpora: Vec<(String, Vec<u8>)> =
vec![("the generated corpus".to_string(), crate::pack_fixture::corpus().pack.clone())];
for (pack_path, _) in real_pairs(4) {
if let Ok(bytes) = std::fs::read(&pack_path) {
corpora.push((pack_path.display().to_string(), bytes));
}
}
for (pack_path, pack) in corpora {
let pack_path = Named(&pack_path);
let w = walk(&pack, 20).unwrap();
let mut uses: HashMap<u64, usize> = HashMap::new();
for e in &w.entries {
if let DeltaBase::Offset(o) = e.delta_base {
*uses.entry(o).or_default() += 1;
}
}
let shared = uses.values().filter(|n| **n > 1).count();
let fan_out = uses.values().copied().max().unwrap_or(0);
if shared == 0 {
continue;
}
widest = widest.max(fan_out);
let rows = resolve(&pack, GitHashKind::Sha1, 0, &NoBases)
.unwrap_or_else(|e| panic!("{}: {e}", pack_path.display()));
assert_eq!(
rows.len(),
w.entries.len(),
"{}: {} of {} entries resolved",
pack_path.display(),
rows.len(),
w.entries.len()
);
eprintln!(
"{}: {shared} bases shared by more than one entry, widest fan-out {fan_out}, all \
{} entries resolved",
pack_path.display(),
rows.len()
);
checked += 1;
}
assert!(
checked > 0 && widest > 2,
"no pack compared had a base shared by more than two entries (checked {checked}, \
widest {widest}) — this test cannot see an early free and must not report a pass"
);
}
fn corpora(cap: usize) -> Vec<(String, Vec<u8>)> {
let mut out: Vec<(String, Vec<u8>)> =
vec![("the generated corpus".to_string(), crate::pack_fixture::corpus().pack.clone())];
for (pack_path, _) in real_pairs(cap) {
if let Ok(bytes) = std::fs::read(&pack_path) {
out.push((pack_path.display().to_string(), bytes));
}
}
out
}
fn rows_of(rows: &[Resolved]) -> Vec<(Vec<u8>, u64, u64, u64)> {
let mut v: Vec<_> = rows
.iter()
.map(|r| (r.oid.clone(), r.offset, r.uncompressed_size, r.delta_base))
.collect();
v.sort();
v
}
fn stream(pack: &[u8], limits: ResolveLimits<'_>) -> (Vec<Resolved>, ResolveReport, u64) {
let w = walk(pack, 20).unwrap();
let mut out = Vec::new();
let mut calls = 0u64;
let report = resolve_streaming(
pack,
&w,
GitHashKind::Sha1,
0,
&NoBases,
&NoSink,
limits,
&mut |rows| {
calls += 1;
out.extend(rows.drain(..));
Ok(())
},
)
.unwrap();
(out, report, calls)
}
#[test]
fn a_bounded_base_set_yields_the_same_rows_under_a_hard_ceiling_and_counts_reinflations() {
let mut checked = 0;
for (name, pack) in corpora(4) {
let w = walk(&pack, 20).unwrap();
let mut uses: HashMap<u64, usize> = HashMap::new();
for e in &w.entries {
if let DeltaBase::Offset(o) = e.delta_base {
*uses.entry(o).or_default() += 1;
}
}
let forest = crate::forest::DeltaForest::build(&w).unwrap();
let two_deep = (0..w.entries.len() as u32).any(|i| {
let kids = forest.children(i);
kids.len() > 1 && kids[..kids.len() - 1].iter().any(|c| !forest.children(*c).is_empty())
});
if !uses.values().any(|n| *n > 1) || !two_deep {
continue;
}
let (free, free_report, _) = stream(&pack, ResolveLimits::none());
assert_eq!(free_report.reinflations, 0, "{name}: unbounded must never re-inflate");
assert_eq!(free_report.evictions, 0);
let largest_base = free
.iter()
.filter(|r| uses.contains_key(&r.offset))
.map(|r| r.uncompressed_size)
.max()
.unwrap();
let tight = ResolveLimits { base_budget: 1, ..ResolveLimits::none() };
let (bounded, report, _) = stream(&pack, tight);
assert_eq!(rows_of(&free), rows_of(&bounded), "{name}: the ceiling changed the output");
assert!(report.reinflations > 0, "{name}: no re-inflation — the ceiling never bit");
assert!(report.evictions > 0, "{name}: nothing evicted");
assert!(
report.peak_base_bytes <= largest_base,
"{name}: peak held {} > the largest single base {largest_base} — the ceiling leaked",
report.peak_base_bytes
);
assert!(
free_report.peak_base_bytes >= report.peak_base_bytes,
"{name}: the unbounded pass held less ({}) than the bounded one ({})",
free_report.peak_base_bytes,
report.peak_base_bytes
);
let mid = ResolveLimits { base_budget: largest_base * 3, ..ResolveLimits::none() };
let (mid_rows, mid_report, _) = stream(&pack, mid);
assert_eq!(rows_of(&free), rows_of(&mid_rows), "{name}: mid budget changed the output");
assert!(
mid_report.reinflations <= report.reinflations,
"{name}: more room, more re-inflations ({} > {})",
mid_report.reinflations,
report.reinflations
);
assert!(mid_report.peak_base_bytes <= largest_base * 3);
eprintln!(
"{name}: {} objects, unbounded peak {} B; budget 1 B → peak {} B, {} evictions, {} re-inflations; budget {} B → {} re-inflations",
free.len(),
free_report.peak_base_bytes,
report.peak_base_bytes,
report.evictions,
report.reinflations,
largest_base * 3,
mid_report.reinflations
);
checked += 1;
}
assert!(
checked > 0,
"no pack in the corpus has a path the ceiling can bite on — nothing was proven"
);
}
#[test]
fn rows_are_handed_over_in_batches_and_the_union_is_the_whole_pack() {
let pack = crate::pack_fixture::corpus().pack.clone();
let n = walk(&pack, 20).unwrap().entries.len();
assert!(n > 7);
let (all, report, calls) = stream(&pack, ResolveLimits { row_batch: 7, ..ResolveLimits::none() });
assert_eq!(all.len(), n);
assert_eq!(calls, n.div_ceil(7) as u64, "batches of 7 over {n} rows");
assert_eq!(report.batches, calls);
assert_eq!(report.rows, n as u64);
let (one, _, calls1) = stream(&pack, ResolveLimits { row_batch: usize::MAX, ..ResolveLimits::none() });
assert_eq!(calls1, 1);
assert_eq!(rows_of(&all), rows_of(&one));
let mut oids: Vec<_> = all.iter().map(|r| r.oid.clone()).collect();
oids.sort();
oids.dedup();
assert_eq!(oids.len(), n, "a row was doubled across a batch boundary");
}
#[test]
fn the_watchdog_pauses_a_resolve_mid_pack_drops_the_bases_and_the_rows_are_unchanged() {
use std::sync::Arc;
use std::time::Duration;
let dir = crate::store::tests::tmpdir("resolve-watchdog");
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(MemWatchdog::with(&meminfo, 3_200_000 * 1024, Duration::from_millis(5)));
let pack = crate::pack_fixture::corpus().pack.clone();
let (free, _, _) = stream(&pack, ResolveLimits::none());
let m2 = meminfo.clone();
let d2 = dog.clone();
let raiser = std::thread::spawn(move || {
let deadline = std::time::Instant::now() + Duration::from_secs(30);
while !d2.is_paused() && std::time::Instant::now() < deadline {
std::thread::sleep(Duration::from_millis(2));
}
std::thread::sleep(Duration::from_millis(150));
std::fs::write(&m2, "MemTotal: 16000000 kB\nMemAvailable: 9000000 kB\n").unwrap();
});
let limits = ResolveLimits {
check_every: 3,
watchdog: Some(dog.clone()),
..ResolveLimits::none()
};
let (rows, report, _) = stream(&pack, limits);
raiser.join().unwrap();
assert_eq!(rows_of(&free), rows_of(&rows), "the pause changed the output");
assert_eq!(report.pauses, 1, "one pause: the floor was breached once and the raise ended it");
assert!(report.watchdog_drops >= 1, "the pause must drop the held bases");
assert!(report.paused_ms >= 100, "waited {} ms; must have waited for the raise", report.paused_ms);
assert_eq!(dog.pauses(), 1);
}
#[test]
fn a_short_box_sheds_its_workers_before_it_refuses_the_pack() {
use std::sync::Arc;
use std::time::Duration;
let dir = crate::store::tests::tmpdir("resolve-shed");
let meminfo = dir.join("meminfo");
std::fs::write(&meminfo, "MemTotal: 16000000 kB\nMemAvailable: 1000000 kB\n").unwrap();
let pack = crate::pack_fixture::corpus().pack.clone();
let one = tape_of(&pack, 1, false).0;
let dog = Arc::new(
MemWatchdog::with(&meminfo, 3_200_000 * 1024, Duration::from_millis(2))
.bounded(1 << 20, Duration::from_secs(30)),
);
let progress = crate::progress::IndexProgress::detached();
let (m2, d2, p2) = (meminfo.clone(), dog.clone(), progress.clone());
let raiser = std::thread::spawn(move || {
let deadline = std::time::Instant::now() + Duration::from_secs(30);
while !(p2.sheds() > 0 && d2.is_paused()) && std::time::Instant::now() < deadline {
std::thread::sleep(Duration::from_millis(2));
}
std::fs::write(&m2, "MemTotal: 16000000 kB\nMemAvailable: 9000000 kB\n").unwrap();
});
let w = walk(&pack, 20).unwrap();
let tape = Tape::default();
let sink = TapeSink { tape: &tape, only_commits: false };
let limits = ResolveLimits {
workers: 4,
row_batch: 64,
lookahead_bytes: 1 << 16,
check_every: 2,
watchdog: Some(dog.clone()),
progress: Some(&progress),
..ResolveLimits::none()
};
let report = resolve_streaming(
&pack,
&w,
GitHashKind::Sha1,
0,
&NoBases,
&sink,
limits,
&mut |rows| {
tape.put(b'B', &[&(rows.len() as u64).to_le_bytes()]);
for r in rows.iter() {
tape.put(
b'R',
&[
&r.oid,
&r.offset.to_le_bytes(),
&r.len.to_le_bytes(),
&[r.stored_type as u8, r.kind as u8],
&r.uncompressed_size.to_le_bytes(),
&r.delta_base.to_le_bytes(),
],
);
}
rows.clear();
Ok(())
},
);
raiser.join().unwrap();
let report = report.expect("a shed must not refuse the pack");
assert!(
report.shed > 0,
"the box was short from the first checkpoint and NOTHING was shed — the guard \
proved nothing (workers {}, pauses {})",
report.workers,
report.pauses
);
assert_eq!(report.rows, w.entries.len() as u64, "every entry is still a row");
assert!(tape.take() == one, "the shed changed the stream");
}
#[test]
fn a_cancel_aborts_the_pack_at_the_next_checkpoint() {
let cancel = AtomicBool::new(true);
let pack = crate::pack_fixture::corpus().pack.clone();
let w = walk(&pack, 20).unwrap();
let limits = ResolveLimits { check_every: 2, cancel: Some(&cancel), ..ResolveLimits::none() };
let mut got = 0usize;
let err = resolve_streaming(&pack, &w, GitHashKind::Sha1, 0, &NoBases, &NoSink, limits, &mut |rows| {
got += rows.len();
rows.clear();
Ok(())
})
.unwrap_err();
assert!(err.to_string().contains("cancelled"), "{err:#}");
assert!(got < w.entries.len(), "the cancel did not stop the pack ({got} rows)");
}
#[derive(Default)]
struct Tape {
bytes: std::sync::Mutex<Vec<u8>>,
}
impl Tape {
fn put(&self, tag: u8, parts: &[&[u8]]) {
let mut b = self.bytes.lock().unwrap_or_else(|e| e.into_inner());
b.push(tag);
for p in parts {
b.extend_from_slice(&(p.len() as u64).to_le_bytes());
b.extend_from_slice(p);
}
}
fn take(&self) -> Vec<u8> {
std::mem::take(&mut self.bytes.lock().unwrap_or_else(|e| e.into_inner()))
}
}
struct TapeSink<'t> {
tape: &'t Tape,
only_commits: bool,
}
impl crate::exploded::PayloadSink for TapeSink<'_> {
fn explode(&self, oid: &[u8], kind: GitObjectKind, payload: &[u8]) -> Result<()> {
self.tape.put(b'E', &[oid, &[kind as u8], payload]);
Ok(())
}
fn wants(&self, kind: GitObjectKind) -> bool {
!self.only_commits || matches!(kind, GitObjectKind::Commit | GitObjectKind::Tag)
}
}
fn tape_of(pack: &[u8], workers: usize, only_commits: bool) -> (Vec<u8>, ResolveReport) {
let w = walk(pack, 20).unwrap();
let tape = Tape::default();
let sink = TapeSink { tape: &tape, only_commits };
let limits = ResolveLimits {
workers,
row_batch: 64,
lookahead_bytes: 1 << 16,
..ResolveLimits::none()
};
let report = resolve_streaming(
pack,
&w,
GitHashKind::Sha1,
0,
&NoBases,
&sink,
limits,
&mut |rows| {
tape.put(b'B', &[&(rows.len() as u64).to_le_bytes()]);
for r in rows.iter() {
tape.put(
b'R',
&[
&r.oid,
&r.offset.to_le_bytes(),
&r.len.to_le_bytes(),
&[r.stored_type as u8, r.kind as u8],
&r.uncompressed_size.to_le_bytes(),
&r.delta_base.to_le_bytes(),
],
);
}
rows.clear();
Ok(())
},
)
.expect("resolve");
(tape.take(), report)
}
#[test]
fn the_same_pack_resolves_to_the_same_bytes_at_1_2_and_4_workers() {
let mut proved = 0usize;
for (name, pack) in corpora(4) {
for only_commits in [false, true] {
let (one, r1) = tape_of(&pack, 1, only_commits);
let (two, r2) = tape_of(&pack, 2, only_commits);
let (four, r4) = tape_of(&pack, 4, only_commits);
assert!(!one.is_empty(), "{name}: the tape is empty, so it proves nothing");
assert_eq!(one.len(), two.len(), "{name}: 1 vs 2 workers, tape length");
assert!(one == two, "{name}: 1 vs 2 workers, byte for byte");
assert!(one == four, "{name}: 1 vs 4 workers, byte for byte");
assert_eq!(r1.rows, r2.rows, "{name}");
assert_eq!(r1.rows, r4.rows, "{name}");
assert_eq!(r1.batches, r4.batches, "{name}");
assert_eq!(r1.workers, 1, "{name}");
proved += 1;
}
}
assert!(proved >= 2, "no pack was compared — this guard proved nothing");
}
#[test]
fn the_determinism_tape_rejects_a_stream_whose_order_moved() {
let pack = corpora(2).into_iter().next().expect("a pack").1;
let (good, _) = tape_of(&pack, 4, false);
let a = good.iter().position(|b| *b == b'R').expect("a row record");
let rec = 1 + 6 * 8 + 20 + 8 + 8 + 2 + 8 + 8; assert!(good.len() > a + 2 * rec, "the tape is too short to reorder");
let mut bad = good.clone();
for i in 0..rec {
bad.swap(a + i, a + rec + i);
}
assert_ne!(good, bad, "the two rows were identical — swapping them proved nothing");
assert!(good != bad, "an order-blind comparison would have called these equal");
}
#[test]
fn a_sink_that_wants_nothing_is_never_handed_a_payload() {
for (name, pack) in corpora(3) {
let w = walk(&pack, 20).unwrap();
let tape = Tape::default();
let sink = TapeSink { tape: &tape, only_commits: true };
let mut rows = 0u64;
let report = resolve_streaming(
&pack,
&w,
GitHashKind::Sha1,
0,
&NoBases,
&sink,
ResolveLimits { workers: 4, ..ResolveLimits::none() },
&mut |r| {
rows += r.len() as u64;
r.clear();
Ok(())
},
)
.expect("resolve");
assert_eq!(rows, w.entries.len() as u64, "{name}: every entry is a row");
assert_eq!(report.rows, rows, "{name}");
let calls = tape
.take()
.chunks(1)
.filter(|c| c[0] == b'E')
.count();
let non_commit = w
.entries
.iter()
.filter(|e| matches!(e.obj_type, ObjType::Blob | ObjType::Tree))
.count();
if non_commit > 0 {
assert!(
calls < w.entries.len(),
"{name}: the sink was handed all {} objects although it wanted only commits",
w.entries.len()
);
}
}
}
fn base_ceiling_pack(roots: usize, depth: u32, body_bytes: usize) -> Vec<u8> {
use crate::pack_fixture::{build_pack, Item};
let mut state = 0x5eed_0000_0000_0001u64;
let mut body = |n: usize| -> Vec<u8> {
let mut v = Vec::with_capacity(n + 8);
while v.len() < n {
state = state.wrapping_add(0x9e37_79b9_7f4a_7c15);
let mut z = state;
z = (z ^ (z >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9);
z = (z ^ (z >> 27)).wrapping_mul(0x94d0_49bb_1331_11eb);
v.extend_from_slice(&(z ^ (z >> 31)).to_le_bytes());
}
v.truncate(n);
v
};
let mut items: Vec<Item> = Vec::new();
for _ in 0..roots {
let mut level = vec![items.len()];
items.push(Item::Whole(GitObjectKind::Blob, body(body_bytes)));
for _ in 1..depth {
let mut next = Vec::with_capacity(level.len() * 2);
for base in level {
for _ in 0..2 {
next.push(items.len());
items.push(Item::OfsDelta { base, body: body(body_bytes) });
}
}
level = next;
}
}
build_pack(&items)
}
#[test]
fn four_workers_hold_no_more_bases_than_the_one_budget_allows() {
const ROOTS: usize = 4;
const DEPTH: u32 = 8;
const OBJECT: usize = 4 * 1024;
const BUDGET: u64 = 20 * 1024;
let pack = base_ceiling_pack(ROOTS, DEPTH, OBJECT);
let w = walk(&pack, 20).unwrap();
let mut is_base: std::collections::HashSet<u64> = std::collections::HashSet::new();
for e in &w.entries {
if let DeltaBase::Offset(o) = e.delta_base {
is_base.insert(o);
}
}
let mut four_worker_peak = 0u64;
for workers in [1usize, 4] {
let free_limits = ResolveLimits { workers, ..ResolveLimits::none() };
let (free, free_report, _) = stream(&pack, free_limits);
assert_eq!(free_report.evictions, 0, "{workers} workers: the unbounded pass evicted");
let largest = free
.iter()
.filter(|r| is_base.contains(&r.offset))
.map(|r| r.uncompressed_size)
.max()
.expect("the pack has bases");
assert_eq!(largest, OBJECT as u64, "every object in this pack is one size");
let limits = ResolveLimits { workers, base_budget: BUDGET, ..ResolveLimits::none() };
let mut rows = Vec::new();
let report = resolve_streaming(
&pack,
&w,
GitHashKind::Sha1,
0,
&NoBases,
&NoSink,
limits,
&mut |r| {
rows.extend(r.drain(..));
Ok(())
},
)
.expect("resolve");
assert_eq!(report.workers as usize, workers, "the fan-out took the worker count");
assert_eq!(
rows_of(&free),
rows_of(&rows),
"{workers} workers: the ceiling changed the output"
);
assert!(
report.evictions > 0,
"{workers} workers: nothing was evicted, so the ceiling never bit and this \
guard proves nothing"
);
assert!(
free_report.peak_base_bytes > report.peak_base_bytes,
"{workers} workers: unbounded held {} and bounded {} — the ceiling did nothing",
free_report.peak_base_bytes,
report.peak_base_bytes
);
let allowed = BUDGET + workers as u64 * largest;
assert!(
report.peak_base_bytes <= allowed,
"{workers} workers: held {} bytes against a {BUDGET}-byte budget plus {workers} × \
the {largest}-byte largest base = {allowed}",
report.peak_base_bytes
);
if workers == 4 {
four_worker_peak = report.peak_base_bytes;
let per_thread = workers as u64 * (BUDGET + largest);
assert!(
four_worker_peak < per_thread,
"four workers held {four_worker_peak} bytes, which is four ceilings of \
{BUDGET} + {largest} = {per_thread}: the budget is per thread, not per pack"
);
}
eprintln!(
"{workers} worker(s): peak {} B (unbounded {} B), budget {BUDGET} B, largest base \
{largest} B, {} evictions, {} re-inflations",
report.peak_base_bytes,
free_report.peak_base_bytes,
report.evictions,
report.reinflations
);
}
}
#[test]
fn a_chain_memo_answers_the_chain_and_a_hit_walks_nothing() {
use crate::pack_fixture::{Item, build_pack};
let base = b"100644 a\0".iter().copied().chain([7u8; 20]).collect::<Vec<u8>>();
let mut mid = base.clone();
mid.extend_from_slice(b"100644 b\0");
mid.extend_from_slice(&[9u8; 20]);
let mut top = mid.clone();
top.extend_from_slice(b"100644 c\0");
top.extend_from_slice(&[11u8; 20]);
let pack = build_pack(&[
Item::Whole(GitObjectKind::Tree, base.clone()),
Item::OfsDelta { base: 0, body: mid.clone() },
Item::OfsDelta { base: 1, body: top.clone() },
]);
let w = crate::pack_walk::walk(&pack, 20).unwrap();
let archive = SliceArchive(&pack);
let none = |_: &[u8]| None;
let top_at = w.entries[2].offset;
let mid_at = w.entries[1].offset;
let (k0, plain, cost0) = materialize_at(&archive, top_at, 20, &none, &NoBases).unwrap();
assert_eq!((k0, plain.as_slice(), cost0.depth), (GitObjectKind::Tree, top.as_slice(), 2));
let mut memo = ChainCache::new(1 << 20);
let (k1, first, cost1) =
materialize_at_with(&archive, top_at, 20, &none, &NoBases, Some(&mut memo)).unwrap();
assert_eq!((k1, first.as_slice(), cost1.depth), (GitObjectKind::Tree, top.as_slice(), 2), "a cold memo is the chain");
let (k2, again, cost2) =
materialize_at_with(&archive, top_at, 20, &none, &NoBases, Some(&mut memo)).unwrap();
assert_eq!((k2, again.as_slice()), (GitObjectKind::Tree, top.as_slice()));
assert_eq!(cost2.depth, 0, "a hit at the asked offset walks no link");
let (k3, middle, cost3) =
materialize_at_with(&archive, mid_at, 20, &none, &NoBases, Some(&mut memo)).unwrap();
assert_eq!((k3, middle.as_slice(), cost3.depth), (GitObjectKind::Tree, mid.as_slice(), 0), "every link on the way down was memoised");
let (hits, misses) = memo.stats();
assert_eq!((hits, misses), (2, 3), "first read: 3 misses (top, mid, base); then 2 hits");
assert!(memo.bytes() == base.len() + mid.len() + top.len(), "three payloads held");
let mut tiny = ChainCache::new(mid.len());
let (_, b1, _) = materialize_at_with(&archive, top_at, 20, &none, &NoBases, Some(&mut tiny)).unwrap();
let (_, b2, _) = materialize_at_with(&archive, top_at, 20, &none, &NoBases, Some(&mut tiny)).unwrap();
assert_eq!((b1.as_slice(), b2.as_slice()), (top.as_slice(), top.as_slice()));
assert!(tiny.bytes() <= mid.len(), "never over its cap");
}
}