gunnar-store 1.5.0

//! 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).map_err(gix_error::Error::from)?;
                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())
                .map_err(gix_error::Error::from)
                .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, &current) {
                // `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(&current) {
                    // Already placed, and so is everything beneath it.
                    break;
                }
                let object = state.objects.get(&current).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().map_err(gix_error::Error::from)?.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,
            }))
        }
    }
}