crates/gunnar-core/src/history.rs
//! 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));
}
}