znippy-plugin-git 0.1.9

//! `__gunnar_refs__` — the ref namespace as an append-only Arrow push log.
//!
//! One `RecordBatch` per push, written through [`crate::pushlog`]. A push that
//! updates three branches is three rows in one batch, and those three either all
//! land or none do, because the frame either completes or it does not. That is
//! the whole transaction mechanism — there is no lock file, no journal and (D18)
//! no database.
//!
//! ## Columns
//!
//! | column | meaning |
//! |---|---|
//! | `name` | full ref name, e.g. `refs/heads/main` |
//! | `target` | the oid it now points at; **null means the ref was deleted** |
//! | `peeled` | for an annotated tag, the commit it peels to |
//! | `symref_target` | for a symbolic ref (`HEAD` → `refs/heads/main`), its target |
//! | `push_seq` | monotonic push counter — the ordering authority |
//! | `updated_ms` | wall clock, unix ms; for humans, never for ordering |
//!
//! `updated_ms` is deliberately not the ordering key: two pushes inside the same
//! millisecond, or a clock that steps backwards, would silently reorder the ref
//! namespace. `push_seq` is assigned by the writer and is the only thing
//! [`fold`] compares.
//!
//! ## Current state
//!
//! The log is the history; [`fold`] replays it into the current namespace, last
//! writer wins by `push_seq`, and a null `target` removes the ref. Reading the
//! current refs is therefore a scan of a structure sized by *pushes*, not by
//! repository size.

use std::collections::BTreeMap;
use std::path::Path;
use std::sync::{Arc, Mutex, MutexGuard};

use anyhow::{Result, anyhow};
use znippy_common::GUNNAR_REFS_MODULE;
use znippy_common::arrow::array::{Array, StringArray, StringBuilder, UInt64Array, UInt64Builder};
use znippy_common::arrow::datatypes::{DataType, Field, Schema};
use znippy_common::arrow::record_batch::RecordBatch;

use crate::pushlog::{PushLog, PushLogScan, read_sealed};

/// One ref update inside a push. Owned by the `git-storage-trait` contract;
/// re-exported here so `crate::refs::RefUpdate` stays a valid path.
pub use git_storage_trait::RefUpdate;

/// The state of one ref after replaying the log.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RefState {
    pub target: Option<String>,
    pub peeled: Option<String>,
    pub symref_target: Option<String>,
    pub push_seq: u64,
    pub updated_ms: u64,
}

pub fn refs_schema() -> Arc<Schema> {
    Arc::new(Schema::new(vec![
        Field::new("name", DataType::Utf8, false),
        Field::new("target", DataType::Utf8, true),
        Field::new("peeled", DataType::Utf8, true),
        Field::new("symref_target", DataType::Utf8, true),
        Field::new("push_seq", DataType::UInt64, false),
        Field::new("updated_ms", DataType::UInt64, false),
    ]))
}

/// Build the single `RecordBatch` that *is* one push.
pub fn build_push_batch(updates: &[RefUpdate], push_seq: u64, updated_ms: u64) -> Result<RecordBatch> {
    let n = updates.len();
    let mut name = StringBuilder::with_capacity(n, n * 32);
    let mut target = StringBuilder::with_capacity(n, n * 64);
    let mut peeled = StringBuilder::with_capacity(n, n * 64);
    let mut symref = StringBuilder::with_capacity(n, n * 32);
    let mut seq = UInt64Builder::with_capacity(n);
    let mut ms = UInt64Builder::with_capacity(n);

    for u in updates {
        name.append_value(&u.name);
        match &u.target {
            Some(t) => target.append_value(t),
            None => target.append_null(),
        }
        match &u.peeled {
            Some(t) => peeled.append_value(t),
            None => peeled.append_null(),
        }
        match &u.symref_target {
            Some(t) => symref.append_value(t),
            None => symref.append_null(),
        }
        seq.append_value(push_seq);
        ms.append_value(updated_ms);
    }

    RecordBatch::try_new(
        refs_schema(),
        vec![
            Arc::new(name.finish()),
            Arc::new(target.finish()),
            Arc::new(peeled.finish()),
            Arc::new(symref.finish()),
            Arc::new(seq.finish()),
            Arc::new(ms.finish()),
        ],
    )
    .map_err(|e| anyhow!("refs push batch: {e}"))
}

/// The folded namespace and the sequence counter, held in memory.
///
/// See [`RefLog`]'s "one pass" note. Built by exactly one scan — the first that
/// needs it after this handle was constructed — and then moved forward by the
/// writer instead of re-derived.
struct RefCache {
    state: BTreeMap<String, RefState>,
    next_seq: u64,
}

impl RefCache {
    /// Move the cache over one appended batch. Only ever called with a batch
    /// whose `push_seq` is the highest in the log, which is what makes this
    /// identical to re-running [`fold`]: last writer wins by `push_seq`, and
    /// this writer is the last one.
    fn apply(&mut self, updates: &[RefUpdate], seq: u64, ms: u64) {
        for u in updates {
            if u.target.is_none() && u.symref_target.is_none() {
                self.state.remove(&u.name);
            } else {
                self.state.insert(
                    u.name.clone(),
                    RefState {
                        target: u.target.clone(),
                        peeled: u.peeled.clone(),
                        symref_target: u.symref_target.clone(),
                        push_seq: seq,
                        updated_ms: ms,
                    },
                );
            }
        }
        self.next_seq = self.next_seq.max(seq + 1);
    }
}

/// The ref log of one repository.
///
/// # One pass per push, not two per push and both O(the log)
///
/// Until 2026-08-30 a single ref update read the whole log **twice**: once
/// through [`current`](Self::current) to compare the caller's expectation
/// against the namespace, and once more inside [`push`](Self::push) to re-derive
/// `push_seq`. Both are `O(pushes)`, so the cost of landing a ref grew with
/// every ref that had ever landed — the busier the repository, the slower its
/// next push, which is the wrong direction for the one number a client waits on.
///
/// The log is still the authority and the frame is still the transaction. What
/// changed is that the fold and the counter are **carried forward** in memory by
/// the writer that already holds the gate, instead of being recomputed from
/// disk by every reader and every writer. One scan happens, lazily, the first
/// time this handle is asked anything; after that a push is one `build`, one
/// `append`, one `fsync` and one in-memory apply.
///
/// **The crash contract is unchanged**, because the cache is not a durability
/// mechanism: it is only ever advanced *after* [`PushLog::append`] has returned,
/// i.e. after the frame is on the platter, so a crash loses exactly the frames
/// it lost before and a restarted process rebuilds the cache with the same
/// single scan recovery has always done. A failed append leaves the cache where
/// it was.
///
/// **It assumes one writer per store**, which is what gunnar already guarantees
/// and enforces one level up: `GitStore` serialises ref work behind a process-
/// local `ref_gate`, and two processes over one store were already unsound
/// before this (that mutex could never have seen the other one). A caller that
/// mutates the file behind this handle's back — the recovery tests do — must
/// re-open the log or call [`invalidate`](Self::invalidate), exactly as a
/// restarted process would.
pub struct RefLog {
    log: PushLog,
    cache: Mutex<Option<RefCache>>,
}

impl RefLog {
    pub fn new(path: impl Into<std::path::PathBuf>) -> Self {
        Self { log: PushLog::new(path, refs_schema()), cache: Mutex::new(None) }
    }

    /// The cache, loaded from the log if this handle has not needed it yet.
    fn cached(&self) -> Result<MutexGuard<'_, Option<RefCache>>> {
        let mut guard = self
            .cache
            .lock()
            .map_err(|_| anyhow!("the ref log cache is poisoned"))?;
        if guard.is_none() {
            let scan = self.log.scan()?;
            *guard = Some(RefCache {
                state: fold(&scan.pushes)?,
                next_seq: max_push_seq(&scan.pushes).map_or(0, |m| m + 1),
            });
        }
        Ok(guard)
    }

    /// Forget the folded namespace and the counter, so the next question is
    /// answered by a fresh scan of the file.
    ///
    /// For a caller that changed the log by some route other than this handle.
    /// Nothing on a serving path calls it; the recovery tests do, and so does
    /// [`append_batch`](Self::append_batch), which hands the sequence out of the
    /// writer's hands.
    pub fn invalidate(&self) {
        if let Ok(mut g) = self.cache.lock() {
            *g = None;
        }
    }

    /// The next `push_seq` this log should use: one past the highest already
    /// recorded. Derived from the log the first time it is asked, so a crashed
    /// writer that lost its counter cannot reuse a sequence number and silently
    /// reorder history; carried forward from there.
    pub fn next_push_seq(&self) -> Result<u64> {
        Ok(self.cached()?.as_ref().expect("loaded above").next_seq)
    }

    /// Append one push atomically, durable before it returns. Returns the
    /// `push_seq` it was given.
    pub fn push(&self, updates: &[RefUpdate]) -> Result<u64> {
        self.push_with(updates, true)
    }

    /// Append one push **without its own `fsync`**.
    ///
    /// For a frame that carries *rebuildable* state rather than a durable fact —
    /// gunnar's `HEAD`, which is projected from the catalog row on every open of
    /// the store and re-projected if it is missing. The frame is still whole or
    /// absent (the framing gives that, not the barrier), and the very next
    /// durable append to this log fsyncs the same file and takes it with it.
    ///
    /// Never for a ref a client was told landed. [`push`](Self::push) is that.
    pub fn push_rebuildable(&self, updates: &[RefUpdate]) -> Result<u64> {
        self.push_with(updates, false)
    }

    fn push_with(&self, updates: &[RefUpdate], durable: bool) -> Result<u64> {
        let mut guard = self.cached()?;
        let cache = guard.as_mut().expect("loaded above");
        let seq = cache.next_seq;
        let ms = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_millis() as u64)
            .unwrap_or(0);
        let batch = build_push_batch(updates, seq, ms)?;
        // The cache moves ONLY after the frame is on the platter (or, for a
        // rebuildable frame, in the file): a failed append must leave the
        // namespace and the counter exactly where they were.
        self.log.append_with(&batch, durable)?;
        cache.apply(updates, seq, ms);
        Ok(seq)
    }

    /// Append an already-built push batch, whose `push_seq` the caller assigned.
    ///
    /// The counter is out of this handle's hands here, so the cache is dropped
    /// rather than advanced — the next reader pays one scan. [`push`](Self::push)
    /// is the path a server takes; this one is for a bench that drives the
    /// append directly.
    pub fn append_batch(&self, batch: &RecordBatch) -> Result<u64> {
        let offset = self.log.append(batch)?;
        self.invalidate();
        Ok(offset)
    }

    pub fn scan(&self) -> Result<PushLogScan> {
        self.log.scan()
    }

    /// Full passes over the log file this handle has paid. One, for a handle
    /// that has only ever pushed — see [`PushLog::scans`].
    pub fn scans(&self) -> u64 {
        self.log.scans()
    }

    /// Fold every frame into one. See [`PushLog::compact`] — rows and their
    /// order are preserved, so [`fold`] answers identically before and after.
    pub fn compact(&self) -> Result<crate::pushlog::CompactionReport> {
        self.log.compact()
    }

    /// Compact if the log has grown past `policy`.
    pub fn maybe_compact(
        &self,
        policy: crate::pushlog::CompactionPolicy,
    ) -> Result<Option<crate::pushlog::CompactionReport>> {
        self.log.maybe_compact(policy)
    }

    /// The current ref namespace. The **whole-snapshot** accessor, for the
    /// batch callers — `refs`, `put_refs_cas`, the live set — that read every
    /// ref at once and must see them at one instant. `head` and a single-ref
    /// CAS want one row, not a copy of the namespace; they take [`get`](Self::get).
    pub fn current(&self) -> Result<BTreeMap<String, RefState>> {
        Ok(self.cached()?.as_ref().expect("loaded above").state.clone())
    }

    /// **One ref by name, without folding a copy of the whole namespace.**
    /// O(log r) against [`current`](Self::current)'s O(r): `head` and a
    /// single-ref compare-and-swap each need exactly one row, and cloning the
    /// entire `BTreeMap<String, RefState>` to read one key is the cost this
    /// removes — a mirror with thousands of refs paid it on every `head`, which
    /// a client asks on every connection. The one `RefState` it returns is
    /// cloned out so no lock is held past the lookup, exactly as `current` does
    /// for the whole map.
    pub fn get(&self, name: &str) -> Result<Option<RefState>> {
        Ok(self
            .cached()?
            .as_ref()
            .expect("loaded above")
            .state
            .get(name)
            .cloned())
    }

    /// The reserved section to seal into the archive.
    pub fn seal_section(&self) -> Result<znippy_common::ReservedSection> {
        self.log.seal_section(GUNNAR_REFS_MODULE)
    }
}

fn max_push_seq(batches: &[RecordBatch]) -> Option<u64> {
    let mut max = None;
    for b in batches {
        let seq = b.column_by_name("push_seq")?.as_any().downcast_ref::<UInt64Array>()?;
        for i in 0..seq.len() {
            max = Some(max.map_or(seq.value(i), |m: u64| m.max(seq.value(i))));
        }
    }
    max
}

/// Replay pushes into the current namespace. Last writer wins by `push_seq`; a
/// null `target` and no `symref_target` deletes the ref.
///
/// Ordering is by `push_seq` and never by position in the vector, so a caller
/// that hands the batches over out of order still gets the right answer.
pub fn fold(batches: &[RecordBatch]) -> Result<BTreeMap<String, RefState>> {
    let mut rows: Vec<(u64, usize, RefState, String)> = Vec::new();

    for (bi, b) in batches.iter().enumerate() {
        let name = col::<StringArray>(b, "name")?;
        let target = col::<StringArray>(b, "target")?;
        let peeled = col::<StringArray>(b, "peeled")?;
        let symref = col::<StringArray>(b, "symref_target")?;
        let seq = col::<UInt64Array>(b, "push_seq")?;
        let ms = col::<UInt64Array>(b, "updated_ms")?;

        for i in 0..b.num_rows() {
            rows.push((
                seq.value(i),
                bi,
                RefState {
                    target: (!target.is_null(i)).then(|| target.value(i).to_string()),
                    peeled: (!peeled.is_null(i)).then(|| peeled.value(i).to_string()),
                    symref_target: (!symref.is_null(i)).then(|| symref.value(i).to_string()),
                    push_seq: seq.value(i),
                    updated_ms: ms.value(i),
                },
                name.value(i).to_string(),
            ));
        }
    }

    rows.sort_by_key(|(seq, bi, _, _)| (*seq, *bi));

    let mut out: BTreeMap<String, RefState> = BTreeMap::new();
    for (_, _, state, name) in rows {
        if state.target.is_none() && state.symref_target.is_none() {
            out.remove(&name);
        } else {
            out.insert(name, state);
        }
    }
    Ok(out)
}

/// Read the sealed `__gunnar_refs__` section out of an archive. `Ok(None)` when
/// the archive carries none — distinct from an archive whose refs are empty.
pub fn read_refs(archive: &Path) -> Result<Option<BTreeMap<String, RefState>>> {
    match read_sealed(archive, GUNNAR_REFS_MODULE)? {
        Some(batches) => Ok(Some(fold(&batches)?)),
        None => Ok(None),
    }
}

fn col<'a, T: Array + 'static>(b: &'a RecordBatch, name: &str) -> Result<&'a T> {
    b.column_by_name(name)
        .ok_or_else(|| anyhow!("refs: no `{name}` column"))?
        .as_any()
        .downcast_ref::<T>()
        .ok_or_else(|| anyhow!("refs: `{name}` has an unexpected type"))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::pushlog::truncate_for_test;

    fn tmpdir(tag: &str) -> std::path::PathBuf {
        let ns = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos();
        let d = std::env::temp_dir().join(format!("znippy_refs_{tag}_{ns}"));
        std::fs::create_dir_all(&d).unwrap();
        d
    }

    fn oid(c: char) -> String {
        std::iter::repeat_n(c, 40).collect()
    }

    /// **The keyed [`RefLog::get`] answers exactly what a whole-namespace
    /// [`RefLog::current`] holds for that key — for every key, present or
    /// absent.** `head` and a single-ref CAS moved onto `get` to skip cloning
    /// the namespace; this is the property that lets them, and it must hold
    /// after deletes and overwrites, not just on a fresh log.
    ///
    /// Seen RED by keying `get` with `.range(name..).next()` (a prefix match
    /// rather than an exact one): `refs/heads/m` then returned `refs/heads/main`
    /// where `current` had nothing, and the absent-key assertion named both.
    #[test]
    fn get_answers_exactly_what_current_holds() {
        let dir = tmpdir("keyed-get");
        let log = RefLog::new(dir.join("refs.log"));
        log.push(&[
            RefUpdate::set("refs/heads/main", oid('a')),
            RefUpdate::set("refs/heads/topic", oid('b')),
            RefUpdate::set("refs/tags/v1", oid('c')),
        ])
        .unwrap();
        log.push(&[RefUpdate::set("refs/heads/main", oid('d'))]).unwrap();
        log.push(&[RefUpdate::delete("refs/heads/topic")]).unwrap();

        let whole = log.current().unwrap();
        // Every key the namespace holds is returned identically by the keyed
        // read.
        for (name, state) in &whole {
            assert_eq!(
                log.get(name).unwrap().as_ref(),
                Some(state),
                "get({name}) != current[{name}]"
            );
        }
        // A deleted ref is absent both ways — the keyed read must not resurrect
        // it from an earlier frame.
        assert_eq!(log.get("refs/heads/topic").unwrap(), None);
        assert!(!whole.contains_key("refs/heads/topic"));
        // A never-written ref, and a prefix of a real one, are both absent.
        assert_eq!(log.get("refs/heads/nope").unwrap(), None);
        assert_eq!(log.get("refs/heads/mai").unwrap(), None);
        // The overwrite is the current value, not the first.
        assert_eq!(log.get("refs/heads/main").unwrap().unwrap().target, Some(oid('d')));

        std::fs::remove_dir_all(&dir).ok();
    }

    #[test]
    fn last_writer_wins_and_a_null_target_deletes() {
        let dir = tmpdir("fold");
        let log = RefLog::new(dir.join("refs.log"));
        log.push(&[
            RefUpdate::set("refs/heads/main", oid('a')),
            RefUpdate::set("refs/heads/topic", oid('b')),
        ])
        .unwrap();
        log.push(&[RefUpdate::set("refs/heads/main", oid('c'))]).unwrap();
        log.push(&[RefUpdate::delete("refs/heads/topic")]).unwrap();

        let refs = log.current().unwrap();
        assert_eq!(refs["refs/heads/main"].target, Some(oid('c')), "second push must win");
        assert!(!refs.contains_key("refs/heads/topic"), "a null target deletes the ref");
        assert_eq!(refs.len(), 1);

        std::fs::remove_dir_all(&dir).ok();
    }

    /// A multi-ref push is one frame, so a crash during it must leave the ref
    /// namespace exactly as it was before — never with some of its refs updated.
    /// This is the property that makes the format usable without a lock file,
    /// and the one a bolt-on ref file could not offer.
    #[test]
    fn a_crash_mid_push_leaves_no_partial_ref_update() {
        let dir = tmpdir("atomic");
        let path = dir.join("refs.log");
        let log = RefLog::new(&path);
        log.push(&[RefUpdate::set("refs/heads/main", oid('a'))]).unwrap();
        let before = std::fs::metadata(&path).unwrap().len();

        // A push touching three refs at once.
        log.push(&[
            RefUpdate::set("refs/heads/main", oid('9')),
            RefUpdate::set("refs/heads/a", oid('1')),
            RefUpdate::set("refs/heads/b", oid('2')),
        ])
        .unwrap();
        let after = std::fs::metadata(&path).unwrap().len();
        let intact = std::fs::read(&path).unwrap();

        for cut in (before + 1)..after {
            std::fs::write(&path, &intact).unwrap();
            truncate_for_test(&path, cut).unwrap();
            // A fresh handle, because that is what a crash leaves: the file as
            // the truncation left it and a process with no memory of the log.
            // `log`'s in-memory fold is deliberately not consulted here — it
            // would be answering about a push that this test just un-wrote
            // behind its back, which no crash can do.
            let refs = RefLog::new(&path).current().unwrap();
            assert_eq!(
                refs.len(),
                1,
                "cut at {cut}: a torn push must not publish ANY of its refs (got {refs:?})"
            );
            assert_eq!(
                refs["refs/heads/main"].target,
                Some(oid('a')),
                "cut at {cut}: main must still be the pre-push value"
            );
            assert!(!refs.contains_key("refs/heads/a"), "cut at {cut}: leaked a partial ref");
            assert!(!refs.contains_key("refs/heads/b"), "cut at {cut}: leaked a partial ref");
        }

        // Intact again: all three land together.
        std::fs::write(&path, &intact).unwrap();
        let refs = RefLog::new(&path).current().unwrap();
        assert_eq!(refs.len(), 3, "the complete push publishes all three refs");
        assert_eq!(refs["refs/heads/main"].target, Some(oid('9')));

        std::fs::remove_dir_all(&dir).ok();
    }

    /// `push_seq` is re-derived from the log, so a writer that crashed and lost
    /// its in-memory counter cannot reuse a sequence number.
    #[test]
    fn push_seq_is_recovered_from_the_log_not_from_memory() {
        let dir = tmpdir("seq");
        let path = dir.join("refs.log");
        let a = RefLog::new(&path);
        assert_eq!(a.push(&[RefUpdate::set("refs/heads/main", oid('a'))]).unwrap(), 0);
        assert_eq!(a.push(&[RefUpdate::set("refs/heads/main", oid('b'))]).unwrap(), 1);

        // A brand new writer over the same file — the "process restarted" case.
        let b = RefLog::new(&path);
        assert_eq!(
            b.next_push_seq().unwrap(),
            2,
            "a restarted writer must continue the sequence, not restart it"
        );
        assert_eq!(b.push(&[RefUpdate::set("refs/heads/main", oid('c'))]).unwrap(), 2);
        assert_eq!(b.current().unwrap()["refs/heads/main"].target, Some(oid('c')));

        std::fs::remove_dir_all(&dir).ok();
    }

    /// Ordering is by `push_seq`, never by arrival order — a wall clock that
    /// steps backwards, or two pushes in the same millisecond, must not reorder
    /// the namespace.
    #[test]
    fn ordering_is_by_push_seq_not_by_timestamp_or_position() {
        // Same `updated_ms` for both, and the LATER push handed over FIRST.
        let newer = build_push_batch(&[RefUpdate::set("refs/heads/main", oid('c'))], 7, 1000).unwrap();
        let older = build_push_batch(&[RefUpdate::set("refs/heads/main", oid('a'))], 3, 9999).unwrap();
        let refs = fold(&[newer, older]).unwrap();
        assert_eq!(
            refs["refs/heads/main"].target,
            Some(oid('c')),
            "push_seq 7 must beat push_seq 3 regardless of order or clock"
        );
        assert_eq!(refs["refs/heads/main"].push_seq, 7);
    }

    /// **The cost of landing a ref does not grow with the log.**
    ///
    /// This is the whole point of the 2026-08-30 change and the reason it is a
    /// fix rather than a tuning: before it, one `update_ref` read the entire log
    /// twice — once to fold the namespace for the compare, once more to
    /// re-derive `push_seq` — so a repository's 1000th push paid a thousand
    /// times the scan its first one did.
    ///
    /// The assertion is on **passes over the file**, not on a clock: a
    /// wall-clock ratio on a shared build machine is a coin toss, while "how
    /// many times was the whole log read" is the property itself and is exact.
    /// One handle, N pushes, one scan — for every N.
    #[test]
    fn the_cost_of_a_push_does_not_grow_with_the_log() {
        for prior in [1usize, 100, 1000] {
            let dir = tmpdir(&format!("flat{prior}"));
            let path = dir.join("refs.log");
            let log = RefLog::new(&path);
            for i in 0..prior {
                log.push(&[RefUpdate::set(
                    format!("refs/heads/b{i:05}"),
                    format!("{i:040x}"),
                )])
                .unwrap();
            }
            let after_writes = log.scans();
            // The read half a CAS does, then the write half.
            let _ = log.current().unwrap();
            log.push(&[RefUpdate::set("refs/heads/main", oid('a'))]).unwrap();

            assert_eq!(
                after_writes, 1,
                "{prior} pushes through one handle must cost ONE pass over the log, not one per                  push (saw {after_writes})"
            );
            assert_eq!(
                log.scans(),
                1,
                "the compare-and-swap pair (fold, then append) must add no pass at all"
            );
            assert_eq!(log.next_push_seq().unwrap(), prior as u64 + 1);
            assert_eq!(
                log.current().unwrap()["refs/heads/main"].target,
                Some(oid('a'))
            );
            // And the answer is still the log's, not the cache's opinion of it:
            // a cold handle over the same file folds to exactly the same thing.
            assert_eq!(RefLog::new(&path).current().unwrap(), log.current().unwrap());
            std::fs::remove_dir_all(&dir).ok();
        }
    }

    /// A rebuildable frame skips the barrier and **nothing else**: it is in the
    /// log, it folds like any other, and the next durable push carries it to
    /// the platter.
    #[test]
    fn a_rebuildable_push_is_a_whole_frame_that_simply_was_not_fsynced() {
        let dir = tmpdir("rebuildable");
        let path = dir.join("refs.log");
        let log = RefLog::new(&path);
        log.push_rebuildable(&[RefUpdate::symbolic("HEAD", "refs/heads/main")])
            .unwrap();
        assert_eq!(
            RefLog::new(&path).current().unwrap()["HEAD"].symref_target.as_deref(),
            Some("refs/heads/main"),
            "the frame is complete on disk; only the fsync was skipped"
        );
        assert_eq!(log.push(&[RefUpdate::set("refs/heads/main", oid('a'))]).unwrap(), 1);
        let cold = RefLog::new(&path).current().unwrap();
        assert_eq!(cold.len(), 2, "both frames fold, in one log");
        assert_eq!(cold["refs/heads/main"].target, Some(oid('a')));
        std::fs::remove_dir_all(&dir).ok();
    }

    #[test]
    fn symbolic_and_peeled_refs_round_trip() {
        let dir = tmpdir("sym");
        let log = RefLog::new(dir.join("refs.log"));
        log.push(&[
            RefUpdate::symbolic("HEAD", "refs/heads/main"),
            RefUpdate::set("refs/tags/v1", oid('t')).with_peeled(oid('e')),
        ])
        .unwrap();
        let refs = log.current().unwrap();
        assert_eq!(refs["HEAD"].symref_target.as_deref(), Some("refs/heads/main"));
        assert!(refs["HEAD"].target.is_none(), "a symref has no direct target");
        assert_eq!(refs["refs/tags/v1"].peeled, Some(oid('e')));
        std::fs::remove_dir_all(&dir).ok();
    }
}