gunnar-core 3.0.0

//! The reference history, **read**.
//!
//! [`crate::arrow_log`] has always written this. Every compaction moves the
//! reflog it takes out of the live log into `refs-archive/` as one complete,
//! self-contained Arrow IPC stream, named so that a lexical sort is
//! chronological. What it never had was a reader: `REFS_ARCHIVE_DIR`'s own doc
//! comment said *"Never read at open"*, and until this module the only code in
//! the whole workspace that opened a segment was `arrow_log`'s own
//! `#[cfg(test)]` block. gunnar wrote a complete history of every reference
//! transaction it had ever performed and nothing but its own tests could see
//! it.
//!
//! That is what this module closes, and it closes it with **no new
//! dependency**: the bytes are already Arrow, written by the arrow crates
//! `gunnar-store` already links.
//!
//! # Two things, not one
//!
//! **Reading.** [`History::read`] returns every transaction, oldest first,
//! across every archive segment and the live log. It takes **no lock**: an
//! archive segment is immutable once renamed into place, and the live log is
//! read through the same [`valid_prefix_len`] scan an open uses, so a torn tail
//! from a writer that is still running is truncated rather than replayed. A
//! history read is therefore safe against a *running* server, which is the only
//! way it is ever useful.
//!
//! **Sealing.** [`History::seal`] writes the whole history as **one** Arrow IPC
//! stream holding **one** batch, with the push boundary carried as a
//! [`COL_PUSH_SEQ`] column instead of as a frame boundary.
//!
//! # Why the seal is a column and not a frame
//!
//! The live log's invariant is one `RecordBatch` per transaction, and on a
//! live, fsynced, crash-recoverable log that is exactly right: repeating the
//! schema per frame is what buys independent parseability when a writer can
//! still be interrupted mid-frame.
//!
//! A sealed artifact is by definition past that point — it is fsynced, renamed
//! into place, and never appended to — so the per-frame schema is paid for a
//! property that can no longer be exercised. Carrying the push boundary as a
//! column instead recovers it. This is what Iceberg does and gunnar did not: a
//! snapshot boundary is a sequence-number column inside a manifest, never a
//! file boundary.
//!
//! The saving is measured on gunnar's own bytes rather than modelled; see
//! `sealing_costs_far_less_per_push_than_the_live_log_it_seals`: **99.7 B/push
//! against the live log's 666.2, 6.68x**, 201 transactions, 2026-08-04.
//!
//! # The artifact is readable by things that are not gunnar
//!
//! The ref-log design claims the log is "queryable — DuckDB over the stream", and a claim like
//! that cannot be settled by the code that wrote the bytes. Checked with a
//! foreign implementation, **pyarrow 24.0.0**, against a seal taken from a live
//! server after three real `git push` operations (2026-08-04):
//!
//! ```text
//! >>> with pa.ipc.open_stream("hist.arrow") as r: t = r.read_all()
//! schema: ['push_seq','refname','old_oid','new_oid','old_symref','new_symref','actor','at_unix_ms']
//! batches: 1   rows: 3
//!   push_seq=0 refs/heads/main  -       -> c415176  actor=gunnar
//!   push_seq=1 refs/heads/main  c415176 -> b5d8140  actor=gunnar
//!   push_seq=2 refs/heads/topic -       -> 9933705  actor=gunnar
//! ```
//!
//! One batch, and the oids are the commits `git log` reports. That is the
//! column-not-a-frame property and the round trip, checked by something with no
//! stake in either.
//!
//! # What a snapshot row is, and why it is kept
//!
//! A compaction writes one snapshot batch at the head of the rewritten live
//! log: every reference at its current value, actor [`COMPACTION_ACTOR`], with
//! a null `old_oid` *and* a null `old_symref` for references that certainly had
//! a previous value. That is derived state, not a transaction, and reading it
//! as one would invent history that never happened.
//!
//! It is **kept** and **marked** ([`RefTransaction::is_snapshot`]) rather than
//! dropped, for the reason `COMPACTION_ACTOR`'s own documentation gives: a
//! reflog reader must be able to tell the two apart. Dropping it here would
//! silently empty the history of a repository whose `refs-archive/` was lost or
//! never copied, which is precisely the case where the snapshot is the only
//! surviving evidence.

use std::path::{Path, PathBuf};
use std::sync::Arc;

use arrow_array::builder::{FixedSizeBinaryBuilder, Int64Builder, StringBuilder, UInt64Builder};
use arrow_array::{Array, FixedSizeBinaryArray, Int64Array, RecordBatch, StringArray, UInt64Array};
use arrow_ipc::reader::StreamReader;
use arrow_schema::{DataType, Field, Schema, SchemaRef};

use crate::arrow_log::{
    arrow_err, check_schema, column, encode_stream, ipc_guard, ref_schema, sync_parent_dir,
    valid_prefix_len, write_options, COL_ACTOR, COL_AT, COL_NEW_OID, COL_NEW_SYMREF, COL_OLD_OID,
    COL_OLD_SYMREF, COL_REFNAME, COMPACTION_ACTOR, REFS_ARCHIVE_DIR, REFS_LOG,
};
use gunnar_store::{Error, Result};

/// The column a sealed history carries the push boundary in.
///
/// Dense and monotonic across the whole artifact, starting at 0 for the oldest
/// transaction. A compaction's snapshot batch consumes one value like any other
/// frame did, so `push_seq` is *the frame index of the log this was sealed
/// from* and not a count of pushes a client made.
pub const COL_PUSH_SEQ: &str = "push_seq";

/// The conventional file name for a sealed history, beside the log it seals.
pub const SEALED_HISTORY: &str = "refs-history.arrow";

/// One row of the reference history: a single reference moving, once.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RefTransaction {
    /// Which frame of the source log this row came from. See [`COL_PUSH_SEQ`].
    pub push_seq: u64,
    /// The full reference name, `refs/heads/main` and not `main`.
    pub refname: String,
    /// The value before, absent for a creation **and** on a snapshot row.
    pub old_oid: Option<gix_hash::ObjectId>,
    /// The value after, absent for a deletion and for a symbolic target.
    pub new_oid: Option<gix_hash::ObjectId>,
    /// The symbolic value before, for a `HEAD`-shaped reference.
    pub old_symref: Option<String>,
    /// The symbolic value after.
    pub new_symref: Option<String>,
    /// Who did it. [`COMPACTION_ACTOR`] means this is a snapshot row.
    pub actor: String,
    /// When, in milliseconds since the epoch.
    pub at_unix_ms: i64,
}

impl RefTransaction {
    /// Whether this row is a compaction's snapshot of state rather than a
    /// reference actually moving.
    ///
    /// The test is the one `COMPACTION_ACTOR` documents: the actor **and** a
    /// null previous value, both, since a genuine creation by a client also has
    /// a null previous value and a client could in principle be named
    /// `gunnar:compaction`.
    pub fn is_snapshot(&self) -> bool {
        self.actor == COMPACTION_ACTOR && self.old_oid.is_none() && self.old_symref.is_none()
    }

    /// Whether this row deletes the reference: no value after, either kind.
    pub fn is_deletion(&self) -> bool {
        self.new_oid.is_none() && self.new_symref.is_none()
    }
}

/// What one [`History::seal`] produced.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct SealReport {
    /// Frames in the source log, i.e. the number of distinct `push_seq` values.
    pub pushes: usize,
    /// Rows written. One reference moving once is one row.
    pub rows: usize,
    /// Bytes of the sealed artifact.
    pub bytes: u64,
    /// Bytes of live log and archive segments the seal read.
    pub bytes_in: u64,
    /// Archive segments read.
    pub segments_read: usize,
}

impl SealReport {
    /// Sealed bytes per source frame — the number the column-not-a-frame
    /// argument is about. `0.0` when there was nothing to seal.
    pub fn bytes_per_push(&self) -> f64 {
        if self.pushes == 0 {
            return 0.0;
        }
        self.bytes as f64 / self.pushes as f64
    }

    /// How much smaller the seal is than the bytes it read. `0.0` when there
    /// was nothing to seal, so a caller cannot mistake an empty run for a win.
    pub fn shrink_x(&self) -> f64 {
        if self.bytes == 0 {
            return 0.0;
        }
        self.bytes_in as f64 / self.bytes as f64
    }
}

/// Every reference transaction a repository's log still holds, oldest first.
#[derive(Clone, Debug)]
pub struct History {
    hash_kind: gix_hash::Kind,
    /// One entry per frame of the source log, in order.
    batches: Vec<RecordBatch>,
    segments_read: usize,
    live_batches_read: usize,
    bytes_in: u64,
}

impl History {
    /// Read `store_dir`'s whole reference history: every archive segment,
    /// oldest first, then the live log.
    ///
    /// `store_dir` is the `gunnar-store/` directory inside a repository, the
    /// one holding `refs.arrow`. **No lock is taken**; see the module
    /// documentation for why that is safe and why it is the point.
    ///
    /// Fails with [`Error::HashKindMismatch`] if the log belongs to a
    /// repository of the other hash kind, through the same schema check an open
    /// uses — reading a sha256 history as sha1 is a mistake, not a truncation.
    pub fn read(store_dir: impl AsRef<Path>, hash_kind: gix_hash::Kind) -> Result<Self> {
        Self::read_upto(store_dir, hash_kind, u64::MAX)
    }

    /// [`History::read`], but stopping the **live** log at `live_len` bytes.
    ///
    /// The archive segments are immutable and are always read whole; only the
    /// live log is cut, and only at a byte offset a caller obtained from
    /// [`live_mark`], which is always a complete-message boundary. So this
    /// reconstructs the history exactly as it stood when that mark was taken,
    /// which is what makes a watermark verifiable rather than merely recorded:
    /// the state this replays to can be compared against what a seal wrote.
    ///
    /// `u64::MAX` is the whole log, and is how [`History::read`] is expressed
    /// so that there is one reader and not two (one implementation).
    pub fn read_upto(
        store_dir: impl AsRef<Path>,
        hash_kind: gix_hash::Kind,
        live_len: u64,
    ) -> Result<Self> {
        let store_dir = store_dir.as_ref();
        let expected = ref_schema(hash_kind);
        let mut batches = Vec::new();
        let mut bytes_in = 0u64;

        let segments = archive_segments(&store_dir.join(REFS_ARCHIVE_DIR))?;
        let segments_read = segments.len();
        for segment in &segments {
            bytes_in += read_stream(segment, Some(&expected), &mut batches)?;
        }

        let before_live = batches.len();
        let live = store_dir.join(REFS_LOG);
        if live.exists() {
            let bytes = std::fs::read(&live)?;
            // The bytes this read cost are the bytes it read off the platter,
            // which is the whole file however far into it the replay goes.
            bytes_in += bytes.len() as u64;
            let valid = valid_prefix_len(&bytes).min(clamp_len(live_len));
            decode_stream(&bytes[..valid], Some(&expected), &mut batches)?;
        }

        Ok(History {
            hash_kind,
            live_batches_read: batches.len() - before_live,
            batches,
            segments_read,
            bytes_in,
        })
    }

    /// The hash kind this history's oids are.
    pub fn hash_kind(&self) -> gix_hash::Kind {
        self.hash_kind
    }

    /// Frames read, across archive and live log. One push is one frame.
    pub fn pushes(&self) -> usize {
        self.batches.len()
    }

    /// Archive segments read. `0` means the archive is absent or empty, and
    /// then [`History::pushes`] only covers the bounded live window.
    pub fn segments_read(&self) -> usize {
        self.segments_read
    }

    /// Frames that came from the live log rather than the archive.
    pub fn live_pushes(&self) -> usize {
        self.live_batches_read
    }

    /// Frames that came from the archive. This is the number that was
    /// unreachable before this module existed.
    pub fn archived_pushes(&self) -> usize {
        self.batches.len() - self.live_batches_read
    }

    /// Bytes read to produce this history.
    pub fn bytes_in(&self) -> u64 {
        self.bytes_in
    }

    /// Every row, oldest first, with its `push_seq`.
    pub fn transactions(&self) -> Result<Vec<RefTransaction>> {
        let mut out = Vec::new();
        for (push_seq, batch) in self.batches.iter().enumerate() {
            decode_batch(batch, push_seq as u64, &mut out)?;
        }
        Ok(out)
    }

    /// The sealed artifact's bytes: one Arrow IPC stream, one batch,
    /// `push_seq` as a column.
    pub fn seal_to_vec(&self) -> Result<Vec<u8>> {
        let schema = sealed_schema(self.hash_kind);
        let batch = self.sealed_batch(&schema)?;
        encode_stream(&schema, std::slice::from_ref(&batch), &write_options())
    }

    /// Seal to `path`, atomically: write a `.writing` scratch file beside it,
    /// fsync, rename, fsync the directory.
    ///
    /// A crash anywhere before the rename leaves a `.writing` file and no
    /// artifact, never a truncated one that a reader would silently stop at —
    /// the same discipline [`crate::arrow_log`] uses for an archive segment.
    pub fn seal(&self, path: impl AsRef<Path>) -> Result<SealReport> {
        use std::io::Write as _;

        let path = path.as_ref();
        let bytes = self.seal_to_vec()?;
        if let Some(parent) = path.parent() {
            if !parent.as_os_str().is_empty() {
                std::fs::create_dir_all(parent)?;
            }
        }
        let name = path
            .file_name()
            .map(|n| n.to_string_lossy().into_owned())
            .unwrap_or_else(|| SEALED_HISTORY.to_owned());
        let temporary = path.with_file_name(format!("{name}.writing"));
        {
            let mut file = std::fs::OpenOptions::new()
                .write(true)
                .create(true)
                .truncate(true)
                .open(&temporary)?;
            file.write_all(&bytes)?;
            file.flush()?;
            file.sync_all()?;
        }
        std::fs::rename(&temporary, path)?;
        sync_parent_dir(path)?;

        let rows = self.batches.iter().map(|b| b.num_rows()).sum();
        Ok(SealReport {
            pushes: self.batches.len(),
            rows,
            bytes: bytes.len() as u64,
            bytes_in: self.bytes_in,
            segments_read: self.segments_read,
        })
    }

    fn sealed_batch(&self, schema: &SchemaRef) -> Result<RecordBatch> {
        let width = self.hash_kind.len_in_bytes() as i32;
        let mut push_seqs = UInt64Builder::new();
        let mut names = StringBuilder::new();
        let mut old_oids = FixedSizeBinaryBuilder::new(width);
        let mut new_oids = FixedSizeBinaryBuilder::new(width);
        let mut old_symrefs = StringBuilder::new();
        let mut new_symrefs = StringBuilder::new();
        let mut actors = StringBuilder::new();
        let mut ats = Int64Builder::new();

        for (push_seq, batch) in self.batches.iter().enumerate() {
            let refnames = column::<StringArray>(batch, COL_REFNAME)?;
            let olds = column::<FixedSizeBinaryArray>(batch, COL_OLD_OID)?;
            let news = column::<FixedSizeBinaryArray>(batch, COL_NEW_OID)?;
            let old_syms = column::<StringArray>(batch, COL_OLD_SYMREF)?;
            let new_syms = column::<StringArray>(batch, COL_NEW_SYMREF)?;
            let batch_actors = column::<StringArray>(batch, COL_ACTOR)?;
            let batch_ats = column::<Int64Array>(batch, COL_AT)?;

            for row in 0..batch.num_rows() {
                push_seqs.append_value(push_seq as u64);
                names.append_value(refnames.value(row));
                append_optional_binary(&mut old_oids, olds, row)?;
                append_optional_binary(&mut new_oids, news, row)?;
                append_optional_str(&mut old_symrefs, old_syms, row);
                append_optional_str(&mut new_symrefs, new_syms, row);
                actors.append_value(batch_actors.value(row));
                ats.append_value(batch_ats.value(row));
            }
        }

        RecordBatch::try_new(
            schema.clone(),
            vec![
                Arc::new(push_seqs.finish()),
                Arc::new(names.finish()),
                Arc::new(old_oids.finish()),
                Arc::new(new_oids.finish()),
                Arc::new(old_symrefs.finish()),
                Arc::new(new_symrefs.finish()),
                Arc::new(actors.finish()),
                Arc::new(ats.finish()),
            ],
        )
        .map_err(arrow_err)
    }
}

/// Read a sealed history back: the artifact's own reader.
///
/// Present so that a seal is verifiable rather than merely written — a sealed
/// file nothing can open is the bug this module exists to fix, one directory
/// along (measured, not asserted).
pub fn read_sealed(path: impl AsRef<Path>) -> Result<Vec<RefTransaction>> {
    read_sealed_bytes(&std::fs::read(path.as_ref())?)
}

/// [`read_sealed`] over bytes already in hand.
pub fn read_sealed_bytes(bytes: &[u8]) -> Result<Vec<RefTransaction>> {
    let valid = valid_prefix_len(bytes);
    if valid == 0 {
        return Ok(Vec::new());
    }
    ipc_guard(&bytes[..valid])?;
    let reader =
        StreamReader::try_new(std::io::Cursor::new(&bytes[..valid]), None).map_err(arrow_err)?;
    let batches: Vec<RecordBatch> = reader
        .collect::<std::result::Result<Vec<_>, _>>()
        .map_err(arrow_err)?;
    let mut out = Vec::new();
    for batch in &batches {
        let push_seqs = column::<UInt64Array>(batch, COL_PUSH_SEQ)?;
        let mut rows = Vec::new();
        // The push_seq passed here is a placeholder: the sealed batch carries
        // the real one per row, and it is read back below. Passing the
        // placeholder through `decode_batch` rather than duplicating the row
        // decode is deliberate (one implementation).
        decode_batch(batch, 0, &mut rows)?;
        for (row, mut transaction) in rows.into_iter().enumerate() {
            transaction.push_seq = push_seqs.value(row);
            out.push(transaction);
        }
    }
    Ok(out)
}

/// The sealed artifact's schema: the live log's, plus [`COL_PUSH_SEQ`] in
/// front.
///
/// `push_seq` leads because it is the sort key, and a reader that wants only
/// the boundary column reads the narrowest possible projection.
pub fn sealed_schema(hash_kind: gix_hash::Kind) -> SchemaRef {
    let width = hash_kind.len_in_bytes() as i32;
    Arc::new(Schema::new(vec![
        Field::new(COL_PUSH_SEQ, DataType::UInt64, false),
        Field::new(COL_REFNAME, DataType::Utf8, false),
        Field::new(COL_OLD_OID, DataType::FixedSizeBinary(width), true),
        Field::new(COL_NEW_OID, DataType::FixedSizeBinary(width), true),
        Field::new(COL_OLD_SYMREF, DataType::Utf8, true),
        Field::new(COL_NEW_SYMREF, DataType::Utf8, true),
        Field::new(COL_ACTOR, DataType::Utf8, false),
        Field::new(COL_AT, DataType::Int64, false),
    ]))
}

/// Archive segments in `dir`, oldest first.
///
/// `.writing` files are **not** globbed: they are what a crash mid-archive
/// leaves, and a reader that picked one up would be reading a stream whose
/// writer never finished. The name format puts the millisecond first and the
/// compaction sequence second, so a plain lexical sort is chronological.
///
/// Reversing this sort breaks multi-segment archive reads; a single compaction
/// and a one-segment archive cannot catch that ordering seam.
pub(crate) fn archive_segments(dir: &Path) -> Result<Vec<PathBuf>> {
    let entries = match std::fs::read_dir(dir) {
        Ok(entries) => entries,
        // A repository that has never compacted has no archive directory, and
        // that is not an error: its whole history is still live.
        Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
        Err(err) => return Err(err.into()),
    };
    let mut segments: Vec<PathBuf> = entries
        .filter_map(|entry| entry.ok())
        .map(|entry| entry.path())
        .filter(|path| path.extension().map(|e| e == "arrow").unwrap_or(false))
        .collect();
    segments.sort();
    Ok(segments)
}

/// Read one complete Arrow IPC stream, appending its batches.
///
/// The [`valid_prefix_len`] scan is the same one an open uses: a torn tail from
/// a writer that is still running is truncated, not replayed, so this is safe
/// against a live log. Returns the bytes read.
pub(crate) fn read_stream(
    path: &Path,
    expected: Option<&SchemaRef>,
    into: &mut Vec<RecordBatch>,
) -> Result<u64> {
    let bytes = std::fs::read(path)?;
    let valid = valid_prefix_len(&bytes);
    decode_stream(&bytes[..valid], expected, into)?;
    Ok(bytes.len() as u64)
}

/// Decode an already-validated stream prefix. The one place a stream's batches
/// are pulled out, so [`read_stream`] and [`History::read_upto`] cannot come to
/// disagree about what a stream holds.
fn decode_stream(
    valid: &[u8],
    expected: Option<&SchemaRef>,
    into: &mut Vec<RecordBatch>,
) -> Result<()> {
    if valid.is_empty() {
        return Ok(());
    }
    ipc_guard(valid)?;
    let reader = StreamReader::try_new(std::io::Cursor::new(valid), None).map_err(arrow_err)?;
    if let Some(expected) = expected {
        check_schema(reader.schema().as_ref(), expected.as_ref())?;
    }
    for batch in reader {
        into.push(batch.map_err(arrow_err)?);
    }
    Ok(())
}

fn clamp_len(len: u64) -> usize {
    usize::try_from(len).unwrap_or(usize::MAX)
}

// ─────────────────────────── the live watermark ──────────────────────────────

/// A position in the **live** reference log, plus enough identity to decide
/// later whether that position still means the same bytes.
///
/// # Why a position needs identity at all
///
/// The live log is append-only *within one inode*, and that is what makes a
/// byte offset a stable name for a prefix: everything below [`LiveMark::len`]
/// is written, fsynced and never touched again, and [`valid_prefix_len`] only
/// ever moves forward. A compaction breaks that — it stages a replacement and
/// **renames it over** the live log, so the file at the same path is a
/// different inode holding a snapshot plus a window, and byte 4096 of it is a
/// different message entirely.
///
/// So the mark carries `(dev, ino)`. A compaction changes the inode and the
/// mark stops applying, which is detected with one `stat` rather than by
/// hashing the prefix on every read. `len` and `tail` close the one gap that
/// leaves: an inode number freed by the compaction's unlink and immediately
/// reissued to the replacement. Such a log would have to have grown back past
/// `len` **and** carry the identical [`TAIL_WINDOW`] bytes ending there, which
/// is not something a differently-framed stream does.
///
/// It deliberately does not hash the whole prefix. That would be O(history) on
/// every open, which is the cost the archive exists to avoid; the bounded
/// window is O(1). What it therefore does **not** prove is that a prefix was
/// not rewritten in place below `len` by something that is not gunnar — nothing
/// short of a full digest does, and nothing writes this file but the process
/// holding its `O_EXCL` lock.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct LiveMark {
    dev: u64,
    ino: u64,
    /// Length of the stream's leading schema message, which is byte-identical
    /// for the life of one inode and is what lets a reader parse a *tail*.
    schema_len: u64,
    /// Complete-message prefix length when the mark was taken.
    len: u64,
    /// Digest of the last [`TAIL_WINDOW`] bytes at or before `len`.
    tail: u64,
}

/// How much of the marked prefix the identity digest covers.
const TAIL_WINDOW: usize = 8192;

impl LiveMark {
    /// The marked prefix length in bytes. Everything at or after this is what a
    /// reader treats as "since the mark".
    pub fn len(&self) -> u64 {
        self.len
    }

    /// Whether the mark names an empty log — a repository whose references have
    /// never been journalled.
    pub fn is_empty(&self) -> bool {
        self.len == 0
    }

    /// Bytes of one encoded mark.
    pub const ENCODED_LEN: usize = 40;

    /// The mark as bytes, so a caller can persist it without knowing what is in
    /// it.
    ///
    /// The encoding lives here rather than at the call site because the fields
    /// are this module's and the rule that they are all load-bearing is this
    /// module's; a caller that spelled the layout itself would be a second
    /// definition of the mark, free to drift by one field.
    pub fn encode(&self) -> [u8; Self::ENCODED_LEN] {
        let mut out = [0u8; Self::ENCODED_LEN];
        out[0..8].copy_from_slice(&self.dev.to_le_bytes());
        out[8..16].copy_from_slice(&self.ino.to_le_bytes());
        out[16..24].copy_from_slice(&self.schema_len.to_le_bytes());
        out[24..32].copy_from_slice(&self.len.to_le_bytes());
        out[32..40].copy_from_slice(&self.tail.to_le_bytes());
        out
    }

    /// The inverse of [`LiveMark::encode`]. `None` on any length but exactly
    /// [`LiveMark::ENCODED_LEN`] — a short read is a truncated file, and a
    /// truncated watermark that decoded to a *shorter prefix* would silently
    /// widen the delta.
    pub fn decode(bytes: &[u8]) -> Option<Self> {
        if bytes.len() != Self::ENCODED_LEN {
            return None;
        }
        let u = |at: usize| u64::from_le_bytes(bytes[at..at + 8].try_into().expect("8 bytes"));
        Some(Self {
            dev: u(0),
            ino: u(8),
            schema_len: u(16),
            len: u(24),
            tail: u(32),
        })
    }
}

/// Take a [`LiveMark`] on `store_dir`'s live reference log.
///
/// **No lock is taken**, the same as [`History::read`] and for the same reason:
/// the prefix this names is written and fsynced, and a torn tail from a writer
/// that is still running is excluded by [`valid_prefix_len`] rather than
/// replayed.
///
/// A missing log marks as empty rather than failing: a repository that has
/// never had a reference transaction is a normal state, and a mark on it is the
/// honest "nothing has happened yet".
pub fn live_mark(store_dir: impl AsRef<Path>) -> Result<LiveMark> {
    let path = store_dir.as_ref().join(REFS_LOG);
    let bytes = match std::fs::read(&path) {
        Ok(bytes) => bytes,
        Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
            // The digest of an empty prefix, not a zero: a log that appears
            // later must still verify against this mark by the same rule every
            // other mark is verified by, and `0` is a value `window_digest`
            // does not produce.
            return Ok(LiveMark {
                dev: 0,
                ino: 0,
                schema_len: 0,
                len: 0,
                tail: window_digest(&[]),
            });
        }
        Err(err) => return Err(err.into()),
    };
    // The `stat` comes AFTER the read on purpose. An append between the two
    // would only extend the file, and the mark names a prefix, so a later
    // identity is still an identity for these bytes; a `stat` taken *first*
    // could name an inode a compaction then replaced under the read.
    let (dev, ino) = ids(&path)?;
    let len = valid_prefix_len(&bytes);
    let schema_len = crate::arrow_log::message_len(&bytes).unwrap_or(0);
    Ok(LiveMark {
        dev,
        ino,
        schema_len: schema_len as u64,
        len: len as u64,
        tail: window_digest(&bytes[..len]),
    })
}

/// Every reference transaction the live log has recorded **since** `mark`.
///
/// `Ok(None)` — deliberately not an error and deliberately not an empty vector
/// — when the mark no longer describes the log: the inode changed (a
/// compaction), the log is shorter than the mark, or the marked prefix's tail
/// digest disagrees. A caller that overlaid an empty delta in that case would
/// serve the sealed namespace as though nothing had happened since, which is
/// exactly the staleness this exists to remove. `None` means *fall back*, and
/// there is no third answer that is a wrong one.
///
/// The read is O(the tail), and that is the *I/O* as well as the parse. Three
/// positional reads — the leading schema message, the digest window ending at
/// the mark, and the bytes after it — never the whole file. Reading the file and
/// then slicing would have been simpler and would have made every advertisement
/// pay for the whole history in page cache and memcpy, which is precisely the
/// cost the archive exists to avoid: at 400 pushes `refs.arrow` is already
/// 273 kB (`fs.rs`, measured 2026-08-04) against a tail of a few hundred bytes.
///
/// A tail alone is not a stream — every batch message is decoded against the
/// schema message that opened the file — so the schema is spliced in front of
/// it. `mark.len` is a complete-message boundary by construction, which is what
/// makes both the splice and the tail's own [`valid_prefix_len`] scan legal from
/// a standing start.
pub fn read_live_since(
    store_dir: impl AsRef<Path>,
    hash_kind: gix_hash::Kind,
    mark: &LiveMark,
) -> Result<Option<Vec<RefTransaction>>> {
    use std::os::unix::fs::FileExt as _;

    let path = store_dir.as_ref().join(REFS_LOG);
    let file = match std::fs::File::open(&path) {
        Ok(file) => file,
        Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
            // A mark on a log that has since vanished applies to nothing. An
            // empty mark on an absent log is the normal never-journalled state.
            return Ok(mark.is_empty().then(Vec::new));
        }
        Err(err) => return Err(err.into()),
    };
    let (dev, ino, file_len) = {
        use std::os::unix::fs::MetadataExt as _;
        let meta = file.metadata()?;
        (meta.dev(), meta.ino(), meta.size())
    };

    // Identity first, and with one `stat` rather than a digest: the log is
    // append-only within an inode, so an unchanged inode means an unchanged
    // prefix. A compaction renames a replacement over it and lands a new one.
    if !mark.is_empty() && ((dev, ino) != (mark.dev, mark.ino) || file_len < mark.len) {
        return Ok(None);
    }
    let from = clamp_len(mark.len);

    // The bounded digest, which is what catches an inode number reissued to a
    // replacement and a prefix rewritten in place. See [`LiveMark`] for what it
    // does and does not prove.
    let window_at = mark.len.saturating_sub(TAIL_WINDOW as u64);
    let mut window = vec![0u8; (mark.len - window_at) as usize];
    file.read_exact_at(&mut window, window_at)?;
    if digest_of(&window, from) != mark.tail {
        return Ok(None);
    }

    let mut tail = vec![0u8; (file_len - mark.len) as usize];
    file.read_exact_at(&mut tail, mark.len)?;
    // A torn tail from a writer that is still running is excluded rather than
    // replayed, exactly as an open would exclude it.
    let tail_valid = valid_prefix_len(&tail);
    if tail_valid == 0 {
        return Ok(Some(Vec::new()));
    }

    let schema_len = clamp_len(mark.schema_len);
    let mut stream = Vec::with_capacity(schema_len + tail_valid);
    if schema_len > 0 {
        let mut schema = vec![0u8; schema_len];
        file.read_exact_at(&mut schema, 0)?;
        stream.extend_from_slice(&schema);
    }
    stream.extend_from_slice(&tail[..tail_valid]);

    let mut batches = Vec::new();
    decode_stream(&stream, Some(&ref_schema(hash_kind)), &mut batches)?;
    let mut out = Vec::new();
    for (push_seq, batch) in batches.iter().enumerate() {
        decode_batch(batch, push_seq as u64, &mut out)?;
    }
    Ok(Some(out))
}

/// `(device, inode)` of `path`.
fn ids(path: &Path) -> Result<(u64, u64)> {
    use std::os::unix::fs::MetadataExt as _;
    let meta = std::fs::metadata(path)?;
    Ok((meta.dev(), meta.ino()))
}

/// A 64-bit digest of the last [`TAIL_WINDOW`] bytes of `prefix`, mixed with
/// its length so that two different prefixes sharing a tail do not share a
/// digest.
///
/// FNV-1a with a `splitmix64` finalizer, written out rather than pulled in: a
/// hashing crate would be a third-party package on a graph the dependency guard
/// counts in packages, for eleven lines of arithmetic.
fn window_digest(prefix: &[u8]) -> u64 {
    digest_of(
        &prefix[prefix.len().saturating_sub(TAIL_WINDOW)..],
        prefix.len(),
    )
}

/// [`window_digest`] over a window a caller already read positionally, so the
/// reader never has to hold the whole prefix to check it. One definition of the
/// digest, two ways in (one implementation).
fn digest_of(window: &[u8], prefix_len: usize) -> u64 {
    let mut h: u64 = 0xcbf2_9ce4_8422_2325;
    for &b in window {
        h ^= b as u64;
        h = h.wrapping_mul(0x0000_0100_0000_01b3);
    }
    h ^= prefix_len as u64;
    h ^= h >> 30;
    h = h.wrapping_mul(0xbf58_476d_1ce4_e5b9);
    h ^= h >> 27;
    h = h.wrapping_mul(0x94d0_49bb_1331_11eb);
    h ^ (h >> 31)
}

pub(crate) fn decode_batch(
    batch: &RecordBatch,
    push_seq: u64,
    into: &mut Vec<RefTransaction>,
) -> Result<()> {
    let refnames = column::<StringArray>(batch, COL_REFNAME)?;
    let olds = column::<FixedSizeBinaryArray>(batch, COL_OLD_OID)?;
    let news = column::<FixedSizeBinaryArray>(batch, COL_NEW_OID)?;
    let old_syms = column::<StringArray>(batch, COL_OLD_SYMREF)?;
    let new_syms = column::<StringArray>(batch, COL_NEW_SYMREF)?;
    let actors = column::<StringArray>(batch, COL_ACTOR)?;
    let ats = column::<Int64Array>(batch, COL_AT)?;

    for row in 0..batch.num_rows() {
        into.push(RefTransaction {
            push_seq,
            refname: refnames.value(row).to_owned(),
            old_oid: optional_oid(olds, row)?,
            new_oid: optional_oid(news, row)?,
            old_symref: optional_string(old_syms, row),
            new_symref: optional_string(new_syms, row),
            actor: actors.value(row).to_owned(),
            at_unix_ms: ats.value(row),
        });
    }
    Ok(())
}

fn optional_oid(array: &FixedSizeBinaryArray, row: usize) -> Result<Option<gix_hash::ObjectId>> {
    if array.is_null(row) {
        return Ok(None);
    }
    gix_hash::ObjectId::try_from(array.value(row))
        .map(Some)
        .map_err(|err| {
            Error::backend(format!(
                "the reference history holds a malformed oid: {err}"
            ))
        })
}

fn optional_string(array: &StringArray, row: usize) -> Option<String> {
    if array.is_null(row) {
        None
    } else {
        Some(array.value(row).to_owned())
    }
}

fn append_optional_binary(
    builder: &mut FixedSizeBinaryBuilder,
    array: &FixedSizeBinaryArray,
    row: usize,
) -> Result<()> {
    if array.is_null(row) {
        builder.append_null();
    } else {
        builder.append_value(array.value(row)).map_err(arrow_err)?;
    }
    Ok(())
}

fn append_optional_str(builder: &mut StringBuilder, array: &StringArray, row: usize) {
    if array.is_null(row) {
        builder.append_null();
    } else {
        builder.append_value(array.value(row));
    }
}