crates/gunnar-store/src/mem.rs
//! In-memory stores: the secret one that production uses in tests, and the
//! [`GitServe`](git_storage_trait::GitServe) one the conformance suite is
//! **self-tested** against.
//!
//! Cheap to clone: every type here is an `Arc` around its state, so a clone is
//! the *same* store. That is what lets the conformance suite simulate a restart
//! against it without a process boundary.
use std::collections::BTreeMap;
use std::sync::{Arc, RwLock};
use crate::error::{Error, Result};
use crate::secret::SecretStore;
fn poisoned<T>(what: &'static str) -> impl FnOnce(T) -> Error {
move |_| {
Error::backend(format!(
"the {what} lock was poisoned by a panicking writer"
))
}
}
/// A [`SecretStore`] over a `BTreeMap`.
#[derive(Debug, Clone, Default)]
pub struct MemSecretStore {
secrets: Arc<RwLock<BTreeMap<String, Vec<u8>>>>,
}
impl MemSecretStore {
/// A new, empty secret store.
pub fn new() -> Self {
Self::default()
}
}
impl SecretStore for MemSecretStore {
fn get(&self, key: &str) -> Result<Option<Vec<u8>>> {
let secrets = self.secrets.read().map_err(poisoned("secret"))?;
Ok(secrets.get(key).cloned())
}
fn put(&self, key: &str, value: &[u8]) -> Result<()> {
let mut secrets = self.secrets.write().map_err(poisoned("secret"))?;
secrets.insert(key.to_owned(), value.to_vec());
Ok(())
}
fn delete(&self, key: &str) -> Result<()> {
let mut secrets = self.secrets.write().map_err(poisoned("secret"))?;
secrets.remove(key);
Ok(())
}
fn get_or_create(&self, key: &str, mk: &mut dyn FnMut() -> Result<Vec<u8>>) -> Result<Vec<u8>> {
// The write lock is taken *before* the existence check, so a second
// caller waits here rather than minting a second host key.
let mut secrets = self.secrets.write().map_err(poisoned("secret"))?;
if let Some(existing) = secrets.get(key) {
return Ok(existing.clone());
}
let value = mk()?;
secrets.insert(key.to_owned(), value.clone());
Ok(value)
}
}
#[cfg(feature = "conformance")]
pub use git_ops::{MemGitOps, MemGitServe};
#[cfg(feature = "conformance")]
mod git_ops {
//! **The conformance suite's control**, and nothing more than that.
//!
//! # What this is for, and what it is not evidence of
//!
//! `conformance::red` breaks a backend in seven specific ways and asserts
//! the suite catches each one. To do that it needs a backend that is *not*
//! broken to wrap, and neither real implementation lives in this repository
//! — the Arrow arm is in znippy and the gix arm is in edda. So this is that
//! backend: the smallest thing that can honestly answer the contract.
//!
//! Two types, because the contract has two layers: [`MemGitOps`] is the
//! storage half and [`MemGitServe`] is the reading half over it. They are
//! one store — `MemGitServe` holds a `MemGitOps` and every clone of either
//! shares the same state — and the split exists so that the layer a check is
//! about is the layer it names.
//!
//! It is **not** a third storage engine and must never be treated as one. It
//! keeps every pack it is given in a `Vec<u8>`, resolves deltas eagerly at
//! `put_pack`, and has no durability of any kind. `locked_on_target.md`'s
//! two implementations are the two implementations.
//!
//! # One implementation, said out loud: this reader agrees with the suite's writer
//!
//! [`crate::conformance::pack`] writes the fixture packs and this parses
//! them, and both were written together, so their agreeing proves nothing
//! about git's format. The outside check is `gunnar_pack::index_pack` over
//! the same bytes, driven by a dev-dependency test. What this control *does*
//! prove is the thing it is for: that each check in the suite is capable of
//! going red.
use std::collections::{BTreeMap, BTreeSet};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, RwLock};
use anyhow::{anyhow, bail, ensure, Context};
use git_storage_trait::{
Caps, Extent, GcReport, GitOps, GitServe, ObjType, Observed, Oid, PackStats, ReachSet,
RefCas, RefRejection, RefRow, RefTarget, RefUpdate, Stored, TxId,
};
use gix_hash::ObjectId;
use gix_object::{CommitRefIter, Kind, TagRefIter, TreeRefIter};
use crate::conformance::pack;
use crate::refname::{self, HEAD};
/// Every held name under `prefix`, for the directory/file conflict check.
///
/// A range scan, not a full walk: the prefix is `<name>/`, so this touches
/// only the handful of names that could conflict.
fn names_under(refs: &BTreeMap<String, RefRow>, prefix: &str) -> Vec<String> {
refs.range(prefix.to_owned()..)
.take_while(|(name, _)| name.starts_with(prefix))
.map(|(name, _)| name.clone())
.collect()
}
/// What is **actually** at `name`, for a compare-and-swap rejection.
///
/// The three-state is the point, and the reason is recorded on
/// [`Observed`]: a row that exists and cannot be read is not a row that was
/// absent, and reporting it as absent tells a pushing client its create is
/// free when it is not. A row here is already decoded, so the only
/// undecodable shape left is one naming neither an object nor another ref —
/// which the apply path cannot produce and which is therefore reported
/// rather than assumed away.
fn observed(state: &State, name: &str) -> Observed {
match state.refs.get(name) {
None => Observed::Nothing,
Some(row) => match (&row.symref_target, &row.oid) {
(Some(points_to), _) => Observed::Value(RefTarget::Symbolic(points_to.clone())),
(None, Some(raw)) => Observed::Value(RefTarget::Object(raw.clone())),
(None, None) => Observed::Unreadable(
"a ref row naming neither an object nor another ref".to_owned(),
),
},
}
}
#[derive(Debug, Clone)]
struct Object {
obj_type: ObjType,
kind: Kind,
/// The object's content, delta chain applied.
content: Vec<u8>,
/// The pack entry, header and all, byte for byte as it arrived.
stored: Vec<u8>,
/// How many leading bytes of [`Object::stored`] are the entry header —
/// type and size, plus an `OFS_DELTA`'s base distance. The rest is the
/// deflated payload, which is what a pack-copying emitter moves
/// unchanged.
header_len: usize,
/// The size the entry header declares: the delta stream's length for a
/// delta, the content's length for a base entry. Re-emitting the entry
/// under a new header needs the number that was in the old one.
entry_size: u64,
/// For a delta entry, the object its payload is expressed against.
///
/// **A format fact, not a graph fact.** It is the one reason
/// [`MemGitServe::emit_pack`] may put an object in a pack that was not
/// asked for: a pack carrying a delta must carry its base.
base: Option<ObjectId>,
/// `(offset, len)` in this store's own byte space: packs concatenated in
/// arrival order. Stable for the life of the object, which is all the
/// contract asks of an extent.
extent: Extent,
}
#[derive(Debug, Default)]
struct State {
objects: BTreeMap<ObjectId, Object>,
refs: BTreeMap<String, RefRow>,
/// Every pack, kept so `extent` addresses something real and `gc` has
/// bytes to measure rather than to invent.
packs: Vec<Vec<u8>>,
bytes: u64,
}
#[derive(Debug)]
struct Inner {
hash_kind: gix_hash::Kind,
state: RwLock<State>,
next_pack_id: AtomicU64,
next_push_seq: AtomicU64,
}
/// An in-memory [`GitOps`]. Clones share one store.
#[derive(Debug, Clone)]
pub struct MemGitOps {
inner: Arc<Inner>,
}
impl MemGitOps {
/// A new, empty store for `hash_kind`.
pub fn new(hash_kind: gix_hash::Kind) -> Self {
MemGitOps {
inner: Arc::new(Inner {
hash_kind,
state: RwLock::new(State::default()),
next_pack_id: AtomicU64::new(0),
next_push_seq: AtomicU64::new(0),
}),
}
}
/// The hash kind this store was opened at.
pub fn hash_kind(&self) -> gix_hash::Kind {
self.inner.hash_kind
}
fn state(&self) -> anyhow::Result<std::sync::RwLockReadGuard<'_, State>> {
self.inner
.state
.read()
.map_err(|_| anyhow!("the store lock was poisoned by a panicking writer"))
}
fn state_mut(&self) -> anyhow::Result<std::sync::RwLockWriteGuard<'_, State>> {
self.inner
.state
.write()
.map_err(|_| anyhow!("the store lock was poisoned by a panicking writer"))
}
fn id(&self, oid: Oid<'_>) -> anyhow::Result<ObjectId> {
let id = ObjectId::try_from(oid)
.map_err(|_| anyhow!("{} bytes is not an object id", oid.len()))?;
if id.kind() != self.inner.hash_kind {
bail!(
"a {:?} object id in a {:?} repository",
id.kind(),
self.inner.hash_kind
);
}
Ok(id)
}
/// Absorb a pack: resolve every delta, compute every id, index it all.
fn absorb(&self, bytes: &[u8]) -> anyhow::Result<TxId> {
let raw = pack::parse(bytes, self.inner.hash_kind)?;
let mut state = self.state_mut()?;
let base = state.bytes;
// Resolved payloads by pack offset, so an `OFS_DELTA` can reach its
// base without a second parse.
let mut by_offset: BTreeMap<u64, (Kind, Vec<u8>)> = BTreeMap::new();
// The id at each pack offset, so a delta can name its base as an
// object rather than as a position. `emit_pack` needs the object.
let mut id_at: BTreeMap<u64, ObjectId> = BTreeMap::new();
let mut fresh = Vec::with_capacity(raw.len());
for entry in &raw {
let (kind, content) = match entry.base_offset {
None => {
let kind = pack::kind_of(entry.obj_type)
.context("a base entry with a delta type")?;
(kind, entry.payload.clone())
}
Some(offset) => {
let (kind, base) = by_offset.get(&offset).with_context(|| {
format!("an OFS_DELTA whose base at {offset} is not in this pack")
})?;
(*kind, pack::apply_delta(base, &entry.payload)?)
}
};
by_offset.insert(entry.extent.0, (kind, content.clone()));
let id = gix_object::compute_hash(self.inner.hash_kind, kind, &content)?;
id_at.insert(entry.extent.0, id);
let base_id = match entry.base_offset {
None => None,
Some(offset) => Some(*id_at.get(&offset).with_context(|| {
format!("an OFS_DELTA whose base at {offset} is not in this pack")
})?),
};
let (offset, len) = entry.extent;
fresh.push((
id,
Object {
obj_type: entry.obj_type,
kind,
content,
stored: bytes[offset as usize..(offset + len) as usize].to_vec(),
header_len: entry.header_len as usize,
entry_size: entry.payload.len() as u64,
base: base_id,
extent: (base + offset, len),
},
));
}
for (id, object) in fresh {
// Re-pushing an object already held is a no-op, not a rewrite:
// the bytes are the same bytes, by definition of the id.
state.objects.entry(id).or_insert(object);
}
state.bytes = base + bytes.len() as u64;
state.packs.push(bytes.to_vec());
Ok(TxId {
pack_id: Some(self.inner.next_pack_id.fetch_add(1, Ordering::SeqCst)),
extent: Some((base, bytes.len() as u64)),
push_seq: None,
})
}
/// Apply ref updates with no old-value check — the force semantics
/// [`GitOps::put_refs`] has, as against [`GitOps::update_ref`]'s
/// compare-and-swap.
fn apply_refs(&self, state: &mut State, updates: &[RefUpdate]) -> anyhow::Result<()> {
let writing: Vec<&str> = updates
.iter()
.filter(|u| u.target.is_some() || u.symref_target.is_some())
.map(|u| u.name.as_str())
.collect();
let deleting: Vec<&str> = updates
.iter()
.filter(|u| u.target.is_none() && u.symref_target.is_none())
.map(|u| u.name.as_str())
.collect();
let held = &state.refs;
refname::check_no_name_conflict(
&writing,
&deleting,
|name| Ok(held.contains_key(name)),
|prefix| Ok(names_under(held, prefix)),
)?;
// Decide the whole batch before touching the namespace. A name that
// does not validate or an oid that is not hex refuses the *batch*;
// deciding inside the apply loop would leave the edits before the
// bad one applied, which is the partial write `put_refs_cas`
// promises never to happen and which `put_refs` has no reason to do
// either.
enum Decided {
Symbolic(String),
Direct {
oid: Vec<u8>,
peeled: Option<Vec<u8>>,
},
Delete,
}
let mut decided = Vec::with_capacity(updates.len());
for update in updates {
refname::validate(&update.name)?;
decided.push(match (&update.target, &update.symref_target) {
(_, Some(points_to)) => {
refname::validate(points_to)?;
Decided::Symbolic(points_to.clone())
}
(Some(hex), None) => Decided::Direct {
oid: self.parse_hex(hex)?.as_slice().to_vec(),
peeled: match &update.peeled {
Some(hex) => Some(self.parse_hex(hex)?.as_slice().to_vec()),
None => None,
},
},
(None, None) => Decided::Delete,
});
}
for (update, decided) in updates.iter().zip(decided) {
match decided {
Decided::Symbolic(points_to) => {
state.refs.insert(
update.name.clone(),
RefRow {
name: update.name.clone(),
oid: None,
peeled: None,
symref_target: Some(points_to),
},
);
}
Decided::Direct { oid, peeled } => {
state.refs.insert(
update.name.clone(),
RefRow {
name: update.name.clone(),
oid: Some(oid),
peeled,
symref_target: None,
},
);
}
Decided::Delete => {
state.refs.remove(&update.name);
}
}
}
Ok(())
}
fn parse_hex(&self, hex: &str) -> anyhow::Result<ObjectId> {
let id = ObjectId::from_hex(hex.as_bytes())
.with_context(|| format!("not a hex object id: {hex:?}"))?;
if id.kind() != self.inner.hash_kind {
bail!(
"object id {hex} is {:?} but this repository is {:?}",
id.kind(),
self.inner.hash_kind
);
}
Ok(id)
}
/// Everything reachable from `tips`, adding nothing already in `seen`.
///
/// **A gitlink tree entry is not followed.** It names a commit in
/// *another* repository, which this one is not expected to hold;
/// following it would report every repository with a submodule as
/// corrupt.
///
/// An id this store does not have contributes only itself and stops. A
/// `have` from a client whose history diverged is exactly that, and the
/// safe direction is to send more, never less.
fn walk(&self, state: &State, tips: &[ObjectId], seen: &mut BTreeSet<ObjectId>) {
let mut stack: Vec<ObjectId> = tips.to_vec();
while let Some(id) = stack.pop() {
if !seen.insert(id) {
continue;
}
let Some(object) = state.objects.get(&id) else {
continue;
};
match object.kind {
Kind::Commit => {
let mut iter =
CommitRefIter::from_bytes(&object.content, self.inner.hash_kind);
if let Ok(tree) = iter.tree_id() {
stack.push(tree);
}
stack.extend(
CommitRefIter::from_bytes(&object.content, self.inner.hash_kind)
.parent_ids(),
);
}
Kind::Tree => {
for entry in
TreeRefIter::from_bytes(&object.content, self.inner.hash_kind).flatten()
{
if entry.mode.is_commit() {
continue;
}
stack.push(entry.oid.to_owned());
}
}
Kind::Tag => {
if let Ok(target) =
TagRefIter::from_bytes(&object.content, self.inner.hash_kind)
.target_id()
{
stack.push(target);
}
}
Kind::Blob => {}
}
}
}
}
impl GitOps for MemGitOps {
fn put(&self, pack: &[u8], refs: &[RefUpdate]) -> anyhow::Result<TxId> {
// Pack first, then the refs that point into it. The order is the
// contract's, and it is observable here only as the order of these
// two lines — a store with no crash has no way to demonstrate it.
let mut tx = self.absorb(pack)?;
tx.push_seq = self.put_refs(refs)?.push_seq;
Ok(tx)
}
fn put_pack(&self, bytes: &[u8]) -> anyhow::Result<TxId> {
self.absorb(bytes)
}
fn put_refs(&self, updates: &[RefUpdate]) -> anyhow::Result<TxId> {
if updates.is_empty() {
return Ok(TxId::default());
}
let mut state = self.state_mut()?;
self.apply_refs(&mut state, updates)?;
Ok(TxId {
pack_id: None,
extent: None,
push_seq: Some(self.inner.next_push_seq.fetch_add(1, Ordering::SeqCst)),
})
}
fn get(&self, oid: Oid<'_>) -> anyhow::Result<Option<Stored>> {
let id = self.id(oid)?;
let state = self.state()?;
Ok(state.objects.get(&id).map(|object| Stored {
obj_type: object.obj_type,
// Post-resolution: what the object inflates to with its chain
// applied, which is not `bytes.len()` for a delta.
uncompressed_size: object.content.len() as u64,
extent: object.extent,
bytes: object.stored.clone(),
}))
}
fn has(&self, oid: Oid<'_>) -> anyhow::Result<bool> {
let id = self.id(oid)?;
Ok(self.state()?.objects.contains_key(&id))
}
fn size(&self, oid: Oid<'_>) -> anyhow::Result<Option<u64>> {
let id = self.id(oid)?;
Ok(self
.state()?
.objects
.get(&id)
.map(|o| o.content.len() as u64))
}
fn extents(&self, oids: &[Oid<'_>]) -> anyhow::Result<Vec<Option<Extent>>> {
let state = self.state()?;
Ok(oids
.iter()
.map(|raw| {
ObjectId::try_from(*raw)
.ok()
.and_then(|id| state.objects.get(&id))
.map(|o| o.extent)
})
.collect())
}
/// The ref namespace, **without `HEAD`**.
///
/// The filter is the contract and it is one `!=` (the `GitServe` HEAD
/// ruling). Every backend walks `refs/` and *"deliberately excludes the
/// pseudo-refs such as `HEAD`"*, and the reason is a type constraint
/// rather than taste: a
/// name type that admits `HEAD` also admits `MERGE_HEAD`, so a row stream
/// carrying pseudo-refs means widening the name type or filtering at
/// every consumer. `HEAD` is still *stored* here, in the same map, and it
/// is read back through [`MemGitServe::head`] — one place, not two, so
/// the accessor and the stream cannot disagree.
fn refs(&self) -> anyhow::Result<Vec<RefRow>> {
// `BTreeMap` iteration is already sorted by name, which is the order
// the gix arm sorts into as well.
Ok(self
.state()?
.refs
.iter()
.filter(|(name, _)| name.as_str() != HEAD)
.map(|(_, row)| row.clone())
.collect())
}
fn update_ref(
&self,
name: &str,
old: Option<Oid<'_>>,
new: Option<Oid<'_>>,
) -> anyhow::Result<TxId> {
refname::validate(name)?;
let mut state = self.state_mut()?;
let current = state.refs.get(name).and_then(|r| r.oid.clone());
match (old, ¤t) {
// `old: None` means "must not exist" — this is a create.
(None, Some(_)) => bail!("ref {name} must not exist, and does"),
(None, None) => {}
(Some(_), None) => bail!("ref {name} must exist, and does not"),
(Some(want), Some(have)) if want == have.as_slice() => {}
(Some(_), Some(_)) => {
bail!("ref {name} compare-and-swap failed: it does not hold the expected value")
}
}
match new {
Some(raw) => {
let id = self.id(raw)?;
let held = &state.refs;
refname::check_no_name_conflict(
&[name],
&[],
|n| Ok(n != name && held.contains_key(n)),
|prefix| Ok(names_under(held, prefix)),
)?;
state.refs.insert(
name.to_owned(),
RefRow {
name: name.to_owned(),
oid: Some(id.as_slice().to_vec()),
peeled: None,
symref_target: None,
},
);
}
None => {
state.refs.remove(name);
}
}
Ok(TxId {
pack_id: None,
extent: None,
push_seq: Some(self.inner.next_push_seq.fetch_add(1, Ordering::SeqCst)),
})
}
/// **Every edit or none**, with the expectation checked before anything
/// is applied.
///
/// # The same-value case applies to a memory backend too
///
/// The defect it exists for is gix's — a ref store that short-circuits an
/// edit whose new value equals the current one never evaluates the
/// expectation the caller wrote, so an `old: None` (*must not exist*)
/// becomes a silent success and *"exactly one creator wins"* stops being
/// true. This store has no such short-circuit, but that is a property of
/// today's [`Self::apply_refs`] rather than a guarantee, so the check is a
/// whole pass of its own **before** the apply pass and deliberately does
/// not fuse with it. A later optimisation that added a short-circuit
/// could not break the create guard from there.
///
/// # One snapshot
///
/// The write lock is taken once and held across both passes, so every
/// expectation is compared against one instant of the namespace and the
/// batch cannot interleave with another writer.
///
/// # What this arm cannot raise, said rather than hidden
///
/// **[`RefRejection::Locked`] never comes out of this backend.** A second
/// writer *blocks* on the `RwLock` rather than failing against it, so
/// contention here is a wait and never a rejection. The variant is in the
/// contract because the gix arm takes real per-ref lock files and raises
/// it; a poisoned lock stays an ordinary error, because it is a fault and
/// not the transient thing a caller retries.
fn put_refs_cas(&self, edits: &[RefCas<'_>]) -> anyhow::Result<TxId> {
// Not an error: a deletions-free push with nothing to apply calls
// this, and refusing it would make every call site special-case the
// empty batch.
if edits.is_empty() {
return Ok(TxId::default());
}
let mut state = self.state_mut()?;
// Pass one: the whole expectation, against one snapshot, applying
// nothing.
for edit in edits {
let expected = match edit.old {
None => Observed::Nothing,
Some(raw) => Observed::oid(raw),
};
let actual = observed(&state, &edit.name);
if actual != expected {
return Err(anyhow::Error::new(RefRejection::Cas {
name: edit.name.clone(),
expected,
actual,
}));
}
}
// Pass two: the apply, through the same code a push uses, so the
// name-conflict rule and the validation are not a second definition
// (one implementation).
let updates: Vec<RefUpdate> = edits
.iter()
.map(|edit| match edit.new {
// The peeled target the `RefCas` carries rides through into
// the ref row, so a tag applied atomically advertises with
// zero re-peels — the same column `put_refs` fills for a
// non-atomic push (`RefCas::peeled`).
Some(raw) => {
let update =
RefUpdate::set(edit.name.clone(), self.id(raw)?.to_hex().to_string());
Ok(match &edit.peeled {
Some(p) => update.with_peeled(self.id(p)?.to_hex().to_string()),
None => update,
})
}
None => Ok(RefUpdate::delete(edit.name.clone())),
})
.collect::<anyhow::Result<_>>()?;
self.apply_refs(&mut state, &updates)?;
Ok(TxId {
pack_id: None,
extent: None,
push_seq: Some(self.inner.next_push_seq.fetch_add(1, Ordering::SeqCst)),
})
}
fn reachable(&self, want: &[Oid<'_>], have: &[Oid<'_>]) -> anyhow::Result<Vec<Vec<u8>>> {
let tips: Vec<ObjectId> = want
.iter()
.map(|raw| self.id(raw))
.collect::<anyhow::Result<_>>()?;
let excludes: Vec<ObjectId> = have
.iter()
.map(|raw| self.id(raw))
.collect::<anyhow::Result<_>>()?;
let state = self.state()?;
let mut client_has = BTreeSet::new();
self.walk(&state, &excludes, &mut client_has);
let mut all = client_has.clone();
self.walk(&state, &tips, &mut all);
Ok(all
.difference(&client_has)
.map(|id| id.as_slice().to_vec())
.collect())
}
fn gc(&self) -> anyhow::Result<GcReport> {
let mut state = self.state_mut()?;
let bytes_before = state.bytes;
let tips: Vec<ObjectId> = state
.refs
.values()
.filter_map(|row| row.oid.as_deref())
.filter_map(|raw| ObjectId::try_from(raw).ok())
.collect();
let mut live = BTreeSet::new();
self.walk(&state, &tips, &mut live);
state.objects.retain(|id, _| live.contains(id));
let rows = state.objects.len() as u64;
let delta_rows = state
.objects
.values()
.filter(|o| matches!(o.obj_type, ObjType::OfsDelta | ObjType::RefDelta))
.count() as u64;
// Read back rather than assert: walk the refs again over what
// survived and check every id the walk reaches is still held. `walk`
// records an id before it looks for it, so an object that was
// wrongly dropped appears in `recheck` and fails this — which is the
// whole point of measuring the result instead of echoing the input.
let mut recheck = std::collections::BTreeSet::new();
self.walk(&state, &tips, &mut recheck);
let verified = recheck.iter().all(|id| state.objects.contains_key(id));
Ok(GcReport {
strategy: "mem-forget-unreachable",
archive: std::path::PathBuf::from("<memory>"),
retired: None,
bytes_before,
// Nothing is moved, only forgotten, so the byte space and every
// extent in it stay exactly as they were. A generation that
// rewrote them would have to answer for every extent it
// invalidated, which is why this control does not have one.
bytes_after: state.bytes,
rows,
delta_rows,
verified,
// This arm forgets rows; it never retires a whole pack, because
// it never rewrites the byte space a pack lives in.
retired_packs: 0,
})
}
}
// ── the reading half ─────────────────────────────────────────────────────
/// An in-memory [`GitServe`] over a [`MemGitOps`]. Clones share one store.
///
/// # What it genuinely implements, and what it cannot
///
/// A control that quietly always succeeds makes the whole suite decorative,
/// so the split is stated here rather than left to be discovered:
///
/// | method | what this does |
/// |---|---|
/// | `read` / `header` / `sizes` | **real**: the delta chain was resolved at `put_pack`, so these answer the inflated object and the post-resolution size, and the type they report is the resolved kind — never `OfsDelta` |
/// | `head` / `set_head` | **real**: one row, in the same map `refs()` reads, filtered out of the row stream in exactly one place |
/// | `emit_pack` | **real, and a pack copy**: entry payloads go out byte for byte, so `recompressed` is 0 because nothing was re-deflated, not because nothing counted it. An `OFS_DELTA`'s header is rewritten (its base moved), which is what any copying emitter does |
/// | `select` | **real, and it can decline** — see [`MemGitServe::select`] |
///
/// What it is **not** is a third storage engine. It has no durability, it
/// keeps every pack in a `Vec<u8>`, and it deltifies nothing: `emit_pack`
/// never *produces* a delta, it only re-emits one it was given. A backend
/// measured against it would be measured against a memcpy.
#[derive(Debug, Clone)]
pub struct MemGitServe {
ops: MemGitOps,
}
impl MemGitServe {
/// A new, empty store for `hash_kind`.
pub fn new(hash_kind: gix_hash::Kind) -> Self {
MemGitServe {
ops: MemGitOps::new(hash_kind),
}
}
/// The storage half, for a caller that wants the eleven by themselves.
///
/// The **same** store, not a copy: `GitServe: GitOps` means this type
/// already answers everything `MemGitOps` does, and this exists only for
/// a caller holding a concrete type rather than a `dyn`.
pub fn ops(&self) -> &MemGitOps {
&self.ops
}
/// The hash kind this store was opened at.
pub fn hash_kind(&self) -> gix_hash::Kind {
self.ops.hash_kind()
}
/// The resolved kind and content of one object.
fn resolved(&self, oid: Oid<'_>) -> anyhow::Result<Option<(ObjType, Vec<u8>)>> {
let id = self.ops.id(oid)?;
Ok(self
.ops
.state()?
.objects
.get(&id)
.map(|object| (pack::obj_type_of(object.kind), object.content.clone())))
}
/// `id` and every delta base beneath it **that the request already
/// carries**, bases first, appended to `order` unless already `seen`.
///
/// # 🔴 It used to follow the chain past the request, and that shipped
/// broken clones
///
/// This walked the whole stored chain and added every base, *"a format
/// requirement rather than a graph one: an `OFS_DELTA` names its base by
/// in-pack distance, so a pack carrying the delta must carry the base"*.
/// That is true of the pack format and false of the clone. A base pulled
/// in is an object the client never asked for — so a
/// `--filter=blob:none` fetch gets back what it asked to be left out —
/// and if the base is a **tree** it arrives owing children the pack does
/// not contain, so `git index-pack --check-self-contained-and-connected`
/// (what `git clone` runs) refuses the transfer with
/// `did not receive expected object <oid>` while the server logs a
/// success.
///
/// The control does not get to be more permissive than the engines it
/// judges: both real arms follow stock `pack-objects`' rule — reuse a
/// stored delta only when its base is also being packed — and the
/// conformance suite now requires it, so this implements it too. The
/// base decides how an entry is *encoded*; it never decides what the
/// pack *contains*. An entry whose base stops here is rebuilt whole by
/// the caller and counted in `recompressed`.
///
/// `requested` is the deduplicated request, and it is what turns the
/// walk back: the chain is followed only while the next base is in it.
/// Nothing here reads a commit, a tree or a tag.
fn with_bases(
state: &State,
requested: &BTreeSet<ObjectId>,
id: ObjectId,
seen: &mut BTreeSet<ObjectId>,
order: &mut Vec<ObjectId>,
) -> anyhow::Result<()> {
let mut chain = Vec::new();
let mut at = Some(id);
while let Some(current) = at {
if seen.contains(¤t) {
// Already placed, and so is everything beneath it.
break;
}
let object = state.objects.get(¤t).with_context(|| {
format!(
"emit_pack was asked for {}, which this store does not hold; a pack \
emitter that skipped it would send a pack no client can close",
current.to_hex()
)
})?;
chain.push(current);
// **The one line that stops the pack growing.** A base outside
// the request is not followed and not added.
at = object.base.filter(|base| requested.contains(base));
// A chain longer than the store is a cycle, which the id of a
// delta base cannot actually form — asserted rather than trusted
// because the alternative is a hang.
ensure!(
chain.len() <= state.objects.len(),
"the delta chain under {} does not terminate",
id.to_hex()
);
}
for current in chain.into_iter().rev() {
if seen.insert(current) {
order.push(current);
}
}
Ok(())
}
}
impl GitOps for MemGitServe {
fn put(&self, pack: &[u8], refs: &[RefUpdate]) -> anyhow::Result<TxId> {
self.ops.put(pack, refs)
}
fn put_pack(&self, bytes: &[u8]) -> anyhow::Result<TxId> {
self.ops.put_pack(bytes)
}
fn put_refs(&self, updates: &[RefUpdate]) -> anyhow::Result<TxId> {
self.ops.put_refs(updates)
}
fn get(&self, oid: Oid<'_>) -> anyhow::Result<Option<Stored>> {
self.ops.get(oid)
}
fn has(&self, oid: Oid<'_>) -> anyhow::Result<bool> {
self.ops.has(oid)
}
fn size(&self, oid: Oid<'_>) -> anyhow::Result<Option<u64>> {
self.ops.size(oid)
}
fn extents(&self, oids: &[Oid<'_>]) -> anyhow::Result<Vec<Option<Extent>>> {
self.ops.extents(oids)
}
fn refs(&self) -> anyhow::Result<Vec<RefRow>> {
self.ops.refs()
}
fn update_ref(
&self,
name: &str,
old: Option<Oid<'_>>,
new: Option<Oid<'_>>,
) -> anyhow::Result<TxId> {
self.ops.update_ref(name, old, new)
}
fn put_refs_cas(&self, edits: &[RefCas<'_>]) -> anyhow::Result<TxId> {
self.ops.put_refs_cas(edits)
}
fn reachable(&self, want: &[Oid<'_>], have: &[Oid<'_>]) -> anyhow::Result<Vec<Vec<u8>>> {
self.ops.reachable(want, have)
}
fn gc(&self) -> anyhow::Result<GcReport> {
self.ops.gc()
}
}
impl GitServe for MemGitServe {
fn read(&self, oid: Oid<'_>) -> anyhow::Result<Option<(ObjType, Vec<u8>)>> {
self.resolved(oid)
}
fn header(&self, oid: Oid<'_>) -> anyhow::Result<Option<(ObjType, u64)>> {
Ok(self
.resolved(oid)?
.map(|(obj_type, content)| (obj_type, content.len() as u64)))
}
/// Positional, and `None` means **unknown**.
///
/// There is no bulk index here to make this cheaper than a loop, and the
/// contract says a backend without one *"loops over `header` and says so
/// in its own docs"*. This is that sentence.
fn sizes(&self, oids: &[Oid<'_>]) -> anyhow::Result<Vec<Option<u64>>> {
let state = self.ops.state()?;
oids.iter()
.map(|raw| {
let id = self.ops.id(raw)?;
Ok(state.objects.get(&id).map(|o| o.content.len() as u64))
})
.collect()
}
/// `HEAD`, read off the same map [`GitOps::refs`] reads and filters.
///
/// `None` for a repository that has never pointed one — an empty
/// repository has no `HEAD`.
fn head(&self) -> anyhow::Result<Option<RefRow>> {
Ok(self.ops.state()?.refs.get(HEAD).cloned())
}
/// Point `HEAD`, through the same write path a push uses.
///
/// A target starting with `refs/` is symbolic; anything else must be a
/// hex oid of this store's width and is written as a detached target.
/// Anything that is neither is **refused** rather than guessed at: a
/// `HEAD` naming something that resolves to nothing is an unclonable
/// repository, and it is cheaper to say so here. A symbolic target is
/// deliberately not checked for existence — `git init` points `HEAD` at a
/// branch before that branch exists.
fn set_head(&self, target: &str) -> anyhow::Result<TxId> {
let update = if target.starts_with("refs/") {
refname::validate(target)?;
RefUpdate::symbolic(HEAD, target)
} else {
let hex_len = self.ops.hash_kind().len_in_bytes() * 2;
let lowered = target.to_ascii_lowercase();
if lowered.len() != hex_len || ObjectId::from_hex(lowered.as_bytes()).is_err() {
bail!(
"HEAD can be pointed at a ref name under `refs/` or at a {hex_len}-\
character hex oid; `{target}` is neither, and guessing which was meant \
is how a repository ends up unclonable"
);
}
RefUpdate::set(HEAD, lowered)
};
self.ops.put_refs(&[update])
}
/// **Exactly `objects`, deduplicated, and nothing else at all.**
///
/// # It does not walk the graph, and it does not close over delta bases
///
/// The caller holds a selection that is already post-filter,
/// post-shallow and post-`include-tag`, and closing over it here would
/// add back precisely what `--filter=blob:none` or `--depth=N` excluded —
/// an over-send that passes `index-pack --strict` and `fsck` and that no
/// exit code can see. Following delta links instead of graph links was
/// the same over-send wearing a different hat, plus a failure mode the
/// graph version does not have: see [`MemGitServe::with_bases`]. Both
/// are gone, so `objects` out equals `objects` in.
///
/// # It is a copy where it can be, and the receipt says where it was not
///
/// An entry whose stored base is inside the request goes out byte for
/// byte, with only an `OFS_DELTA`'s leading bytes rewritten — the
/// distance is positional, so copying it unchanged would name the wrong
/// base — and counts as `copied`. An entry whose base the request does
/// not carry is **rebuilt whole** and counts as `recompressed`. So the
/// pair is a measurement rather than a literal `0`, and
/// `copied + recompressed == objects` always.
///
/// # `caps`
///
/// `ofs_delta: false` is **refused** when the emission would carry a
/// delta entry, rather than silently emitting one the client will not
/// parse. `thin` is ignored: rebuilding is correct for a thin client
/// too, and costs CPU rather than correctness. `have` is used for
/// neither — this arm deltifies nothing, so there is nothing for a
/// common tip to be expressed against, and it is **never** consulted for
/// membership.
fn emit_pack(
&self,
objects: &[Oid<'_>],
_have: &[Oid<'_>],
caps: &Caps,
out: &mut dyn std::io::Write,
) -> anyhow::Result<PackStats> {
let hash_kind = self.ops.hash_kind();
let state = self.ops.state()?;
// The request, deduplicated. Built before anything is placed,
// because "is this base in the request" has to be answerable for a
// base that appears later in the caller's order than its delta.
let mut requested = BTreeSet::new();
let mut asked = Vec::with_capacity(objects.len());
for raw in objects {
let id = self.ops.id(raw)?;
if requested.insert(id) {
asked.push(id);
}
}
let mut seen = BTreeSet::new();
let mut order = Vec::with_capacity(asked.len());
for id in asked {
MemGitServe::with_bases(&state, &requested, id, &mut seen, &mut order)?;
}
let mut buf = Vec::new();
buf.extend_from_slice(b"PACK");
buf.extend_from_slice(&2u32.to_be_bytes());
buf.extend_from_slice(&(order.len() as u32).to_be_bytes());
let mut at: BTreeMap<ObjectId, u64> = BTreeMap::new();
let mut copied = 0u64;
let mut recompressed = 0u64;
for id in &order {
let object = state
.objects
.get(id)
.expect("with_bases only records objects it found");
let start = buf.len() as u64;
// The base only counts if the request carries it. One outside it
// is not added and not named — the entry is rebuilt instead.
let base_in_pack = object.base.filter(|base| requested.contains(base));
match (object.obj_type, base_in_pack) {
(ObjType::OfsDelta, Some(base)) => {
ensure!(
caps.ofs_delta,
"{} is stored as an OFS_DELTA and the client did not offer \
ofs-delta; emitting it anyway would be a pack this client cannot \
parse, so this refuses instead",
id.to_hex()
);
let base_at = *at.get(&base).context(
"a delta was emitted before its base, which no reader can resolve",
)?;
pack::write_entry_header(&mut buf, ObjType::OfsDelta, object.entry_size);
pack::write_offset(&mut buf, start - base_at);
buf.extend_from_slice(&object.stored[object.header_len..]);
copied += 1;
}
// **The boundary a narrowed request cuts.** There is no base
// to name, so the stored delta cannot be reused at all: the
// resolved content goes out as a base-type entry. This is
// the only thing here that inflates or deflates, and it is
// what `recompressed` exists to report.
(ObjType::OfsDelta, None) => {
pack::write_entry_header(
&mut buf,
pack::obj_type_of(object.kind),
object.content.len() as u64,
);
pack::deflate_into(&mut buf, &object.content)?;
recompressed += 1;
}
(ObjType::RefDelta, _) => bail!(
"{} is a REF_DELTA, which this store never ingests",
id.to_hex()
),
_ => {
buf.extend_from_slice(&object.stored);
copied += 1;
}
}
at.insert(*id, start);
}
let mut hasher = gix_hash::hasher(hash_kind);
hasher.update(&buf);
buf.extend_from_slice(hasher.try_finalize()?.as_slice());
out.write_all(&buf)?;
Ok(PackStats {
bytes: buf.len() as u64,
objects: order.len() as u64,
copied,
recompressed,
})
}
/// Negotiation, **and it declines**.
///
/// `Ok(None)` is the escape hatch and it is load-bearing: the Arrow arm
/// answers it when its reachability projection is empty, which a clean
/// restart produces, and a control that always answered could not
/// exercise the path a caller takes then. Two conditions produce it here,
/// and both are real rather than a knob:
///
/// * **the store holds no objects** — there is nothing to project, which
/// is the clean-restart shape;
/// * **a `want` this store does not hold** — the closure beneath a tip it
/// has never seen is not something it can answer cheaply or at all, and
/// *"walk it yourself"* is the honest reply. Note that
/// [`GitOps::reachable`] answers the same question differently and on
/// purpose: there an unknown id contributes only itself, because the
/// safe direction for a `have` is to send more.
fn select(&self, want: &[Oid<'_>], have: &[Oid<'_>]) -> anyhow::Result<Option<ReachSet>> {
let state = self.ops.state()?;
if state.objects.is_empty() {
return Ok(None);
}
let tips: Vec<ObjectId> = want
.iter()
.map(|raw| self.ops.id(raw))
.collect::<anyhow::Result<_>>()?;
if tips.iter().any(|id| !state.objects.contains_key(id)) {
return Ok(None);
}
let excludes: Vec<ObjectId> = have
.iter()
.map(|raw| self.ops.id(raw))
.collect::<anyhow::Result<_>>()?;
let mut client_has = BTreeSet::new();
self.ops.walk(&state, &excludes, &mut client_has);
let mut all = client_has.clone();
self.ops.walk(&state, &tips, &mut all);
let objects: Vec<ObjectId> = all.difference(&client_has).copied().collect();
let commits = objects
.iter()
.filter(|id| {
state
.objects
.get(*id)
.is_some_and(|o| o.kind == Kind::Commit)
})
.map(|id| id.as_slice().to_vec())
.collect();
Ok(Some(ReachSet {
// The voucher a thin delta may name a base from, so an id the
// client claims and this store does not hold is not in it: it
// could not be a base for anything this store can express.
client_has: client_has
.iter()
.filter(|id| state.objects.contains_key(*id))
.map(|id| id.as_slice())
.collect(),
objects: objects.iter().map(|id| id.as_slice().to_vec()).collect(),
commits,
}))
}
}
}