crates/gunnar-core/src/arrow_log.rs
//! Secrets as an **Arrow IPC log**: there is no database.
//!
//! `ArrowLogSecretStore` is the live half of this module: it stores SSH host keys
//! and bootstrap material in `secrets.arrow`, with append durability and immediate
//! compaction of superseded values so rotated secrets do not remain on disk.
//!
//! The old Arrow reference store has been retired. Reference storage now lives in
//! the znippy ref log; this module keeps only the shared Arrow framing/durability
//! machinery and the secret store. The reference schema constants remain because
//! `crate::history` can still read/seal legacy Arrow reflog history.
use std::collections::BTreeMap;
use std::fs::OpenOptions;
use std::io::{Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex, RwLock};
use std::time::{SystemTime, UNIX_EPOCH};
use arrow_array::builder::{BinaryBuilder, Int64Builder, StringBuilder};
use arrow_array::{Array, BinaryArray, RecordBatch, StringArray};
use arrow_ipc::reader::StreamReader;
use arrow_ipc::writer::{
write_message, DictionaryTracker, IpcDataGenerator, IpcWriteContext, IpcWriteOptions,
};
use arrow_schema::{DataType, Field, Schema, SchemaRef};
use gunnar_store::{Error, Result, SecretStore};
/// The Arrow IPC stream holding this repository's reference log.
pub const REFS_LOG: &str = "refs.arrow";
/// The Arrow IPC stream holding this repository's secrets.
pub const SECRETS_LOG: &str = "secrets.arrow";
/// The `O_EXCL` lock file that refuses a second writer.
pub const LOCK_FILE: &str = "gunnar-store.lock";
/// Where compaction puts the reflog history it takes out of the live log.
///
/// One complete, self-contained Arrow IPC stream per compaction, named so that
/// a lexical sort is chronological. Never read at open.
pub const REFS_ARCHIVE_DIR: &str = "refs-archive";
/// The actor recorded on the snapshot row a compaction writes.
///
/// It is not a ref move, and a reflog reader must be able to tell the two apart:
/// a snapshot row has a null `old_oid` *and* a null `old_symref` for a ref that
/// certainly had a previous value.
pub const COMPACTION_ACTOR: &str = "gunnar:compaction";
/// The actor recorded on a log row when none was supplied.
const DEFAULT_ACTOR: &str = "gunnar";
/// Every Arrow IPC message begins with this, followed by a 4-byte length.
///
/// Named here rather than imported because `arrow-ipc` keeps it private. It is a
/// constant of the *format*, not of the crate, and the tests below read real
/// bytes produced by `arrow-ipc` to prove the two agree.
const CONTINUATION_MARKER: [u8; 4] = [0xff; 4];
pub(crate) const COL_REFNAME: &str = "refname";
pub(crate) const COL_OLD_OID: &str = "old_oid";
pub(crate) const COL_NEW_OID: &str = "new_oid";
pub(crate) const COL_OLD_SYMREF: &str = "old_symref";
pub(crate) const COL_NEW_SYMREF: &str = "new_symref";
pub(crate) const COL_ACTOR: &str = "actor";
pub(crate) const COL_AT: &str = "at_unix_ms";
const COL_KEY: &str = "key";
const COL_VALUE: &str = "value";
// ───────────────────────────── compaction policy ─────────────────────────────
/// When the secret log folds itself automatically.
///
/// `reflog_window` remains in the public constructor for compatibility with old
/// callers; secrets always compact to the current values only.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct CompactionPolicy {
/// Compact once the live log holds this many framed batches.
///
/// Checked on the secret write path. `0` disables threshold-based automatic
/// compaction; replacements are still compacted immediately.
pub after_batches: usize,
/// Ignored by the secret store; retained for API compatibility with callers
/// that still construct the old reference-log policy shape.
pub reflog_window: usize,
}
impl CompactionPolicy {
/// A policy with the given trigger and window.
pub fn new(after_batches: usize, reflog_window: usize) -> Self {
CompactionPolicy {
after_batches,
reflog_window,
}
}
/// Never compact automatically: the pre-compaction behaviour, kept because
/// it is what the growth guard measures against.
pub fn never() -> Self {
CompactionPolicy {
after_batches: 0,
reflog_window: 0,
}
}
fn should_compact(&self, batches: usize) -> bool {
self.after_batches > 0 && batches >= self.after_batches
}
}
impl Default for CompactionPolicy {
/// 1024 secret writes between threshold compactions.
fn default() -> Self {
CompactionPolicy {
after_batches: 1024,
reflog_window: 256,
}
}
}
// ───────────────────────────── schemas ───────────────────────────────────────
/// The reference log's schema for a repository of `hash_kind`.
///
/// The oid width is what pins the hash kind on disk: reopening a sha256 log as
/// sha1 finds a `FixedSizeBinary(32)` where it wanted `FixedSizeBinary(20)` and
/// fails with [`Error::HashKindMismatch`] rather than mixing two repositories.
pub(crate) fn ref_schema(hash_kind: gix_hash::Kind) -> SchemaRef {
let width = hash_kind.len_in_bytes() as i32;
Arc::new(Schema::new(vec![
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),
]))
}
/// The secret log's schema. A null `value` is a delete.
fn secret_schema() -> SchemaRef {
Arc::new(Schema::new(vec![
Field::new(COL_KEY, DataType::Utf8, false),
Field::new(COL_VALUE, DataType::Binary, true),
Field::new(COL_ACTOR, DataType::Utf8, false),
Field::new(COL_AT, DataType::Int64, false),
]))
}
fn hash_kind_of_width(width: i32) -> Option<gix_hash::Kind> {
match width {
20 => Some(gix_hash::Kind::Sha1),
32 => Some(gix_hash::Kind::Sha256),
_ => None,
}
}
/// Reject a stored schema that is not the one we are about to append to.
///
/// A differing oid width is reported as [`Error::HashKindMismatch`] rather than
/// as a generic schema error, because that is the case an operator can act on.
pub(crate) fn check_schema(stored: &Schema, expected: &Schema) -> Result<()> {
if stored.fields() == expected.fields() {
return Ok(());
}
let width_of =
|schema: &Schema, name: &str| match schema.field_with_name(name).map(|f| f.data_type()) {
Ok(DataType::FixedSizeBinary(width)) => Some(*width),
_ => None,
};
if let (Some(found), Some(want)) = (
width_of(stored, COL_NEW_OID),
width_of(expected, COL_NEW_OID),
) {
if found != want {
if let (Some(stored_kind), Some(want_kind)) =
(hash_kind_of_width(found), hash_kind_of_width(want))
{
return Err(Error::HashKindMismatch {
stored: stored_kind,
requested: want_kind,
});
}
}
}
Err(Error::backend(format!(
"the log on disk has schema {stored:?}, which is not the schema this store writes ({expected:?})"
)))
}
// ───────────────────────────── framing ───────────────────────────────────────
/// How many bytes of `log` are complete, framed Arrow IPC messages.
///
/// This is the torn-write recovery point of the Arrow-IPC ref log and the only thing between
/// a half-written push and a log that reads as empty forever. Everything from
/// the returned offset onwards is a message that was never finished, and is
/// **truncated away** at open, not skipped, because an append after garbage is
/// unreachable by any reader.
///
/// A zero metadata length is the stream's end-of-stream marker. This backend
/// never writes one (it would make every later append unreachable), but a
/// compacted or externally-produced log may carry one, so the scan stops
/// *before* it and a subsequent append overwrites it.
pub(crate) fn valid_prefix_len(log: &[u8]) -> usize {
let mut pos = 0usize;
while let Some(total) = message_len(&log[pos..]) {
pos += total;
}
pos
}
/// The on-disk length of the **one** complete framed message `rest` starts
/// with, or `None` when it does not start with one.
///
/// Split out of [`valid_prefix_len`] rather than copied, because
/// [`crate::history::live_mark`] needs the length of the *first* message — the
/// schema — and a second walk of the same framing would be a second definition
/// of what a complete message is (one implementation). The `while let` above is the whole of
/// `valid_prefix_len` now, so a change here changes both.
pub(crate) fn message_len(rest: &[u8]) -> Option<usize> {
if rest.len() < 8 || rest[..4] != CONTINUATION_MARKER {
return None;
}
let meta_len = i32::from_le_bytes([rest[4], rest[5], rest[6], rest[7]]);
if meta_len <= 0 {
return None;
}
let meta_len = meta_len as usize;
if rest.len() < 8 + meta_len {
return None;
}
// Full flatbuffer verification: random bytes that happen to start with a
// continuation marker do not get past this.
let message = arrow_ipc::root_as_message(&rest[8..8 + meta_len]).ok()?;
let body_len = usize::try_from(message.bodyLength()).ok()?;
// `write_message` refuses to emit a body that is not already a multiple
// of the write alignment, so the recorded body length is exactly what is
// on disk. There is no trailing pad to account for.
let total = 8usize.checked_add(meta_len)?.checked_add(body_len)?;
if rest.len() < total {
return None;
}
Some(total)
}
// ───────────────────────────── backing store ─────────────────────────────────
/// Where a log's bytes actually live.
///
/// The file variant is the normal one. The memory variant is the no-OS shape: no
/// filesystem, no `libc`, the log is just bytes. It runs the **same**
/// encode/frame/replay code, so the conformance run against it is evidence about
/// this backend rather than about a second implementation.
enum Backing {
File { file: std::fs::File, path: PathBuf },
Memory { bytes: Arc<Mutex<Vec<u8>>> },
}
fn poisoned(what: &'static str) -> Error {
Error::backend(format!(
"the {what} lock was poisoned by a panicking writer"
))
}
/// The scratch file a compaction writes before the rename. Named beside the
/// live log so the rename cannot cross a filesystem boundary and stop being
/// atomic.
fn compacting_path(path: &Path) -> PathBuf {
let name = path
.file_name()
.map(|n| n.to_string_lossy().into_owned())
.unwrap_or_else(|| "log".to_owned());
path.with_file_name(format!("{name}.compacting"))
}
/// A rename is only durable once the *directory entry* is on the platter.
pub(crate) fn sync_parent_dir(path: &Path) -> Result<()> {
let parent = match path.parent() {
Some(parent) if !parent.as_os_str().is_empty() => parent.to_path_buf(),
_ => PathBuf::from("."),
};
std::fs::File::open(parent)?.sync_all()?;
Ok(())
}
/// How every message in this log is framed.
///
/// **8-byte buffer alignment, not `arrow-ipc`'s default 64.** This is the single
/// largest lever on the size of the log, and the log's size *is* its open cost:
/// the whole live stream is read and replayed at every open.
///
/// A ref transaction is a **one-row** batch of seven columns. Arrow IPC pads
/// every buffer up to the write alignment, so those rows — 32 bytes of oid and a
/// short name — are surrounded by padding that is an order of magnitude larger
/// than the data. At 64 bytes the padding is paid on each of the batch's
/// buffers plus the metadata; at 8 bytes, which is the Arrow IPC format's own
/// stated minimum, it is paid eight times more cheaply.
///
/// MEASURED 2026-08-04, 4000 single-ref transactions against a
/// never-compacting log on md1 (xfs, RAID0 NVMe), **four interleaved A/B pairs**
/// from two binaries differing only in this function, loadavg 2.6–7.0
/// throughout. Medians:
///
/// | alignment | bytes/push | live log at 4001 pushes | durable pushes/s | µs/push | reopen |
/// |---|---|---|---|---|---|
/// | 64 (arrow's default) | 1 536.1 | 6 145 984 B | 15 615 | 64.05 | 20.50 ms |
/// | **8** | **680.1** | **2 721 128 B** | **17 450** | **57.31** | **16.72 ms** |
/// | | **2.258× smaller** | 2.258× | **1.118×** | 1.118× | **1.226× faster** |
///
/// Every one of the four pairs favoured 8 on every metric; the byte figures are
/// exact and load-independent.
///
/// **The throughput win is a consequence of the size win, not a separate
/// effect.** `fsync`'s share of the wall clock is 0.830–0.838 *in both arms* —
/// the sync did not get relatively cheaper, there were simply 2.26× fewer bytes
/// for it to force through the XFS log. That also explains why the one thing
/// tried before this — `fsync` → `fdatasync` — measured **no change at all**
/// (16 252 → 16 449 pushes/s, inside the run-to-run spread): on XFS an append
/// commits the size update through the log either way, so the timestamp
/// metadata `fdatasync` skips was never what was being paid for. Recorded so it
/// is not tried a third time.
///
/// # Why this is safe to change
///
/// 64 is a *recommendation* in the Arrow IPC specification, for the benefit of
/// readers that want to hand a buffer straight to a SIMD kernel; 8 is the
/// specification's requirement. Every message on disk carries its own
/// `bodyLength`, so a stream written at 64 and appended to at 8 stays readable
/// message by message and **no migration is needed** — which
/// `an_existing_64_byte_aligned_log_is_still_read_after_the_alignment_changed`
/// asserts against real bytes rather than by argument.
///
/// What is given up is the guarantee that a mapped body's buffers land on
/// 64-byte boundaries. Nothing in gunnar maps this log: it is read with
/// `StreamReader` over a `Cursor`, which copies into freshly-allocated,
/// suitably-aligned Arrow buffers regardless. An external reader (the module
/// docs promise DuckDB can read this file) is served by the format guarantee,
/// not by the recommendation.
pub(crate) fn write_options() -> IpcWriteOptions {
// 8 is one of the four values `IpcWriteOptions::try_new` accepts, so this
// cannot fail; `expect` rather than a fallback to the default, because a
// silent fallback to 64 would make this whole function unobservable.
IpcWriteOptions::try_new(8, false, arrow_ipc::MetadataVersion::V5)
.expect("8 is a legal Arrow IPC write alignment")
}
/// Encode `batches` as a complete, self-contained Arrow IPC stream.
///
/// Used both for the rewritten live log and for an archive segment. A fresh
/// generator and dictionary tracker each time, because this is a **new** stream:
/// carrying the old stream's tracker over would emit deltas against dictionaries
/// the new stream never wrote.
pub(crate) fn encode_stream(
schema: &SchemaRef,
batches: &[RecordBatch],
options: &IpcWriteOptions,
) -> Result<Vec<u8>> {
let generator = IpcDataGenerator::default();
let mut tracker = DictionaryTracker::new(false);
let mut out = Vec::new();
let encoded = generator.schema_to_bytes_with_dictionary_tracker(schema, &mut tracker, options);
write_message(&mut out, encoded, options).map_err(arrow_err)?;
for batch in batches {
let (dictionaries, message) = generator
.encode(
batch,
&mut tracker,
options,
&mut IpcWriteContext::default(),
)
.map_err(arrow_err)?;
for dictionary in dictionaries {
write_message(&mut out, dictionary, options).map_err(arrow_err)?;
}
write_message(&mut out, message, options).map_err(arrow_err)?;
}
Ok(out)
}
impl Backing {
fn read_all(&self) -> Result<Vec<u8>> {
match self {
Backing::File { file, .. } => {
let mut handle = file;
handle.seek(SeekFrom::Start(0))?;
let mut out = Vec::new();
handle.read_to_end(&mut out)?;
Ok(out)
}
Backing::Memory { bytes } => {
Ok(bytes.lock().map_err(|_| poisoned("in-memory log"))?.clone())
}
}
}
fn truncate_to(&self, len: u64) -> Result<()> {
match self {
Backing::File { file, .. } => Ok(file.set_len(len)?),
Backing::Memory { bytes } => {
bytes
.lock()
.map_err(|_| poisoned("in-memory log"))?
.truncate(len as usize);
Ok(())
}
}
}
fn append(&self, buf: &[u8]) -> Result<()> {
match self {
Backing::File { file, .. } => {
let mut handle = file;
handle.seek(SeekFrom::End(0))?;
handle.write_all(buf)?;
handle.flush()?;
Ok(())
}
Backing::Memory { bytes } => {
bytes
.lock()
.map_err(|_| poisoned("in-memory log"))?
.extend_from_slice(buf);
Ok(())
}
}
}
/// Make everything written so far durable.
///
/// For the memory backing there is no volatile/durable distinction, so this
/// is a no-op that still counts. See the note on
/// [`ArrowLogStore::in_memory`] about which claims that shape can support.
fn sync(&self) -> Result<()> {
match self {
Backing::File { file, .. } => Ok(file.sync_all()?),
Backing::Memory { .. } => Ok(()),
}
}
/// Replace every byte of this log with `bytes`, atomically.
///
/// **Never mutates in place**: a new file is written beside the
/// live one, fsynced, and *then* renamed over it, and the rename is followed
/// by an fsync of the directory so the new name is durable too. A crash
/// before the rename leaves the old log complete and serving; a crash after
/// it leaves the new one complete. There is no third outcome, which is why
/// an interrupted compaction is simply retried.
fn replace_all(&mut self, bytes: &[u8], fsync: bool) -> Result<()> {
self.stage(bytes, fsync)?;
self.swap(bytes, fsync)
}
/// Write the replacement beside the live log and make it durable, without
/// touching the live log. This is the half a crash can interrupt.
///
/// Split out from [`Backing::swap`] so the crash test can stop here and
/// assert what an interrupted compaction leaves behind, using the
/// production staging code, not a re-implementation of it.
fn stage(&self, bytes: &[u8], fsync: bool) -> Result<()> {
let Backing::File { path, .. } = self else {
return Ok(());
};
let temporary = compacting_path(path);
let mut fresh = OpenOptions::new()
.write(true)
.create(true)
.truncate(true)
.open(&temporary)?;
fresh.write_all(bytes)?;
fresh.flush()?;
if fsync {
fresh.sync_all()?;
}
Ok(())
}
/// Put the staged replacement in place. A rename, which is atomic on a real
/// filesystem, followed by an fsync of the directory entry.
fn swap(&mut self, bytes: &[u8], fsync: bool) -> Result<()> {
match self {
Backing::File { file, path } => {
std::fs::rename(compacting_path(path), &*path)?;
if fsync {
sync_parent_dir(path)?;
}
// The old descriptor now points at an unlinked inode.
*file = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(&*path)?;
Ok(())
}
Backing::Memory { bytes: cell } => {
let mut held = cell.lock().map_err(|_| poisoned("in-memory log"))?;
held.clear();
held.extend_from_slice(bytes);
Ok(())
}
}
}
}
impl std::fmt::Debug for Backing {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Backing::File { path, .. } => f.debug_struct("File").field("path", path).finish(),
Backing::Memory { .. } => f.write_str("Memory"),
}
}
}
// ───────────────────────────── the log ───────────────────────────────────────
/// What the durability tests read.
///
/// `written` counts every byte handed to the backing store; `durable` counts the
/// bytes covered by a *completed* sync. A transaction that returns `Ok` with
/// `durable < written` has acknowledged a push that a power cut would lose.
///
/// `written_seq` and `durable_seq` are the same two facts counted in *commits*
/// rather than bytes, and they exist because [`SyncGate`] needs a watermark that
/// **never goes backwards**. A byte offset does: a compaction rewrites the log
/// shorter, so a waiter holding a pre-compaction byte target would wait for an
/// offset the file will never reach again. A commit count only ever rises.
#[derive(Debug, Default)]
struct Durability {
written: AtomicU64,
durable: AtomicU64,
/// Commits handed to the kernel.
written_seq: AtomicU64,
/// Commits covered by a *completed* sync.
durable_seq: AtomicU64,
syncs: AtomicUsize,
/// Time spent inside `fsync`: the ceiling a ref transaction pays, made visible.
fsync_nanos: AtomicU64,
}
/// The descriptor a durability barrier is issued on, held **apart from the log's
/// own** so that one thread can `fsync` while another appends. This is group
/// commit, and it is the whole reason the two are separated.
///
/// # Why the fleet was paying N syncs for N pushes
///
/// A ref transaction is `lock · validate · encode · write · fsync · unlock`, and
/// the `fsync` is 83% of it (MEASURED 2026-08-04, md1). With the sync
/// inside the writer mutex, thirty-two concurrent pushers queue thirty-two
/// syncs end to end and the log's total throughput does not move:
///
/// | concurrent writers | total durable pushes/s | aggregate lock wait |
/// |---|---|---|
/// | 1 | 13 038 | 0.000 s |
/// | 8 | 14 505 | 2.224 s |
/// | 32 | 14 907 | **9.934 s** against 0.550 s of wall clock |
///
/// Flat throughput and lock wait rising linearly with the writer count is the
/// signature of a serialised barrier, not of a busy disk.
///
/// # What it costs now
///
/// **Five interleaved A/B pairs** from two binaries differing only in this
/// change, md1, loadavg matched within each pair (11–26 across the set).
/// Medians of total durable transactions per second:
///
/// | concurrent writers | before | after | |
/// |---|---|---|---|
/// | 1 | 14 387 | 14 836 | 1.03× — nothing to group, and nothing lost |
/// | 2 | 15 901 | 15 964 | 1.00× |
/// | 4 | 15 324 | 26 355 | **1.72×** |
/// | 8 | 15 950 | 37 213 | **2.33×** |
/// | 16 | 15 700 | 40 648 | **2.59×** |
/// | 32 | 15 070 | 42 117 | **2.79×** |
///
/// Aggregate lock wait at 32 writers went **8.5–10.5 s → 0.27–0.66 s**, about
/// 19×, which is the mechanism showing through: the pushers stopped queueing
/// behind one another's `fsync`.
///
/// The single-writer row is the one that had to not move, and it did not. Group
/// commit is a **concurrency** lever; it makes a lone pusher neither faster nor
/// slower, and a change that had quietly taxed the serial path would show here.
/// The 32-writer column is the widest-spread one (21 236 to 50 873 across the
/// five runs) because the box was carrying loadavg 22–26 while it ran; the
/// median is quoted rather than the best, and the *shape* — flat before, rising
/// after — is what the claim rests on.
///
/// # What this does instead
///
/// The descriptor here is a `dup` of the log's — the same open file
/// description and the same inode, so an `fsync` issued on it flushes bytes
/// another thread's `write` put there through the other descriptor. A committer
/// therefore:
///
/// 1. appends its frame under the writer mutex and takes a commit sequence;
/// 2. **releases the writer mutex**, so the next pusher can already be writing;
/// 3. asks this gate to make that sequence durable.
///
/// Step 3 is leader/follower. Whoever gets here first reads `written_seq`
/// *before* syncing and publishes it as `durable_seq` after — so the one sync it
/// performs is credited with every commit that had already reached the kernel
/// when it started, which is exactly the set that `fsync` is guaranteed to
/// cover. Everyone whose sequence is at or below that watermark returns without
/// syncing at all. N pushers cost the number of *rounds*, not N.
///
/// # The one property that changed, stated plainly
///
/// The acknowledgement barrier is unchanged: `transaction` still does not return
/// `Ok` until this gate has made that transaction's bytes durable, and
/// `a_transaction_is_durable_before_it_returns_ok` still asserts it against the
/// file's own length.
///
/// What did change is **reader visibility**. The in-memory ref map is published
/// under the writer mutex, which is now released before the sync, so for the
/// duration of one barrier a concurrent *reader* can observe a ref move whose
/// bytes are not yet on the platter. Nobody has been promised anything — the
/// pusher has not been acknowledged — but a fetch served inside that window and
/// followed by a power cut would have handed out a ref the log no longer names.
/// This is the ordinary read-uncommitted trade a group-committing log makes, and
/// it is written down here rather than discovered later.
struct SyncGate {
/// `None` for the in-memory backing, where there is no volatile/durable
/// distinction to enforce and the gate is a counter.
file: Mutex<Option<std::fs::File>>,
}
impl SyncGate {
fn detached() -> Self {
SyncGate {
file: Mutex::new(None),
}
}
fn on(backing: &Backing) -> Result<Self> {
Ok(SyncGate {
file: Mutex::new(match backing {
Backing::File { file, .. } => Some(file.try_clone()?),
Backing::Memory { .. } => None,
}),
})
}
/// Point the gate at the log's *current* descriptor.
///
/// A compaction renames a replacement over the live log and reopens it, so
/// the descriptor this gate holds would otherwise keep referring to the old,
/// now-unlinked inode and every later `fsync` would flush a file nothing
/// reads. Called from [`Log::rewrite`], under the writer mutex.
fn rebind(&self, backing: &Backing) -> Result<()> {
let mut held = self.file.lock().map_err(|_| poisoned("log sync gate"))?;
*held = match backing {
Backing::File { file, .. } => Some(file.try_clone()?),
Backing::Memory { .. } => None,
};
Ok(())
}
/// Return once commit `target` is durable, doing as little work as the rest
/// of the fleet leaves to do.
fn barrier(&self, durability: &Durability, target: u64, fsync: bool) -> Result<()> {
if !fsync {
return Ok(());
}
// The common case under load: a leader already covered us and we never
// touch the mutex at all.
if durability.durable_seq.load(Ordering::SeqCst) >= target {
return Ok(());
}
let held = self.file.lock().map_err(|_| poisoned("log sync gate"))?;
// Re-checked after queueing, because the leader we queued behind was
// very likely syncing on our behalf. This is the line that turns N
// syncs into one.
if durability.durable_seq.load(Ordering::SeqCst) >= target {
return Ok(());
}
let Some(file) = held.as_ref() else {
return Ok(());
};
// Read BEFORE the sync, published AFTER it. An `fsync` is only
// guaranteed to cover writes that completed before it started, and
// `written`/`written_seq` are bumped *after* the bytes reach the kernel,
// so this watermark can never over-claim.
let covers_seq = durability.written_seq.load(Ordering::SeqCst);
let covers_bytes = durability.written.load(Ordering::SeqCst);
let started = std::time::Instant::now();
file.sync_all()?;
durability
.fsync_nanos
.fetch_add(started.elapsed().as_nanos() as u64, Ordering::SeqCst);
durability.syncs.fetch_add(1, Ordering::SeqCst);
durability.durable.fetch_max(covers_bytes, Ordering::SeqCst);
durability
.durable_seq
.fetch_max(covers_seq, Ordering::SeqCst);
Ok(())
}
}
/// One Arrow IPC stream, opened for append.
struct Log {
backing: Backing,
schema: SchemaRef,
options: IpcWriteOptions,
generator: IpcDataGenerator,
tracker: DictionaryTracker,
durability: Arc<Durability>,
/// The descriptor durability barriers are issued on. Shared with the store,
/// which reaches it *without* the writer mutex — that separation is the
/// group commit; see [`SyncGate`].
gate: Arc<SyncGate>,
/// Framed record batches currently in the live stream. This is what the
/// compaction trigger reads, and what an open would have to replay.
batches: usize,
/// Set to `false` only by the deliberately-broken store the durability
/// guard is asserted against.
fsync: bool,
}
impl std::fmt::Debug for Log {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Log")
.field("backing", &self.backing)
.field("durability", &self.durability)
.finish()
}
}
impl Log {
/// Open `backing` as a stream of `schema`, recovering any torn tail, and
/// return every batch already in it.
fn open(backing: Backing, schema: SchemaRef, fsync: bool) -> Result<(Self, Vec<RecordBatch>)> {
let bytes = backing.read_all()?;
let valid = valid_prefix_len(&bytes);
if valid < bytes.len() {
tracing::warn!(
discarded = bytes.len() - valid,
kept = valid,
"the log ends in an incomplete Arrow IPC message; truncating to the last \
complete one so the next append is reachable"
);
backing.truncate_to(valid as u64)?;
backing.sync()?;
}
let gate = Arc::new(if fsync {
SyncGate::on(&backing)?
} else {
// The deliberately-broken store the durability guard is asserted
// against. A gate with no descriptor cannot sync, which is the
// point: `the_durability_guard_catches_a_missing_fsync` must still
// be able to produce a log that acknowledges before the platter.
SyncGate::detached()
});
let mut log = Log {
backing,
schema,
options: write_options(),
generator: IpcDataGenerator::default(),
tracker: DictionaryTracker::new(false),
durability: Arc::new(Durability::default()),
gate,
batches: 0,
fsync,
};
log.durability.written.store(valid as u64, Ordering::SeqCst);
log.durability.durable.store(valid as u64, Ordering::SeqCst);
let batches = if valid == 0 {
log.write_schema()?;
Vec::new()
} else {
let expected = log.schema.clone();
ipc_guard(&bytes[..valid])?;
let reader = StreamReader::try_new(std::io::Cursor::new(&bytes[..valid]), None)
.map_err(arrow_err)?;
check_schema(reader.schema().as_ref(), expected.as_ref())?;
reader
.collect::<std::result::Result<Vec<_>, _>>()
.map_err(arrow_err)?
};
log.batches = batches.len();
Ok((log, batches))
}
/// Replace the whole stream with `batches`, atomically.
///
/// The old stream keeps serving until the rename lands; see
/// [`Backing::replace_all`].
fn rewrite(&mut self, batches: &[RecordBatch]) -> Result<()> {
let bytes = encode_stream(&self.schema, batches, &self.options)?;
self.backing.replace_all(&bytes, self.fsync)?;
// The live log is now a DIFFERENT INODE: `replace_all` renamed a
// replacement over it and reopened. The gate still holds a descriptor
// on the old, now-unlinked one, and every later barrier would flush a
// file nothing reads — a silent loss of durability that no test of the
// *contents* could see. Rebind before anything can sync again.
self.gate.rebind(&self.backing)?;
let len = bytes.len() as u64;
self.durability.written.store(len, Ordering::SeqCst);
if self.fsync {
self.durability.syncs.fetch_add(1, Ordering::SeqCst);
self.durability.durable.store(len, Ordering::SeqCst);
// Everything staged before this point is durable by construction:
// `replace_all` fsynced a complete replacement that contains it.
// Publishing the watermark releases any waiter whose sequence the
// shortened file will never reach as a byte offset again — which is
// exactly why the watermark counts commits and not bytes.
let staged = self.durability.written_seq.load(Ordering::SeqCst);
self.durability
.durable_seq
.fetch_max(staged, Ordering::SeqCst);
} else {
self.durability.durable.store(0, Ordering::SeqCst);
}
// A rewritten log is a NEW stream. Its schema message was emitted by a
// fresh generator, so the appending one has to be fresh too or the next
// append encodes against state this stream never wrote.
self.generator = IpcDataGenerator::default();
self.tracker = DictionaryTracker::new(false);
self.batches = batches.len();
Ok(())
}
fn write_schema(&mut self) -> Result<()> {
let encoded = self.generator.schema_to_bytes_with_dictionary_tracker(
&self.schema,
&mut self.tracker,
&self.options,
);
let mut framed = Vec::new();
write_message(&mut framed, encoded, &self.options).map_err(arrow_err)?;
self.commit(&framed)
}
/// Append **one** `RecordBatch` as **one** framed IPC message, and return
/// the **commit sequence** that must be made durable before the caller may
/// acknowledge it.
///
/// This does not sync. The caller releases the writer mutex and then calls
/// [`Log::barrier`] with what this returned — which is what lets the next
/// pusher encode and write while this one is still on the platter. See
/// [`SyncGate`].
fn append(&mut self, batch: &RecordBatch) -> Result<u64> {
let (dictionaries, message) = self
.generator
.encode(
batch,
&mut self.tracker,
&self.options,
&mut IpcWriteContext::default(),
)
.map_err(arrow_err)?;
let mut framed = Vec::new();
// This schema carries no dictionary-encoded column, so this loop is
// empty in practice; it is here so that adding one cannot silently drop
// the dictionary messages the batch would then depend on.
for dictionary in dictionaries {
write_message(&mut framed, dictionary, &self.options).map_err(arrow_err)?;
}
write_message(&mut framed, message, &self.options).map_err(arrow_err)?;
let sequence = self.stage(&framed)?;
self.batches += 1;
Ok(sequence)
}
/// Write `framed` and take a commit sequence for it. **Not durable yet.**
///
/// The two counters are bumped *after* the bytes reach the kernel, which is
/// what makes [`SyncGate::barrier`]'s read-before-sync watermark honest: a
/// sequence that is visible in `written_seq` is a sequence whose bytes an
/// `fsync` starting now is guaranteed to cover.
fn stage(&self, framed: &[u8]) -> Result<u64> {
self.backing.append(framed)?;
self.durability
.written
.fetch_add(framed.len() as u64, Ordering::SeqCst);
Ok(self.durability.written_seq.fetch_add(1, Ordering::SeqCst) + 1)
}
/// Make commit `sequence` durable. Callable **without** the writer mutex.
fn barrier(&self, sequence: u64) -> Result<()> {
self.gate.barrier(&self.durability, sequence, self.fsync)
}
/// Write `framed` and make it durable before returning — the whole
/// acknowledgement barrier in one call, for the paths that hold the writer
/// mutex throughout anyway (opening a fresh log, sealing a secret) and gain
/// nothing from group commit.
fn commit(&self, framed: &[u8]) -> Result<()> {
let sequence = self.stage(framed)?;
self.barrier(sequence)
}
}
// ───────────────────────────── the lock ──────────────────────────────────────
/// The `O_EXCL` lock file that makes "one writer" true across processes.
///
/// `flock` would release itself when a crashed process's descriptors close; an
/// `O_EXCL` file does not, and needs no `unsafe` and no `libc`, which this
/// crate forbids outright.
///
/// # Recovery is at startup, not an operator verb
///
/// The trade used to be *"a stale lock after a hard kill, cleared by
/// [`ArrowLogStore::force_unlock`]"*, and it does not survive the no-OS rule's premise: one
/// binary as PID 1 in a distroless image, with no shell and nobody to type the
/// verb. A power cut left every repository the server had open holding a lock
/// and the appliance booted into a permanent refusal to serve.
///
/// So the lock file records **who** holds it and [`acquire`](LockFile::acquire)
/// reads that back. The owner is `<pid> <start-time>`: the pid alone is not
/// enough, because pids are recycled and a recycled one reads as "still
/// running". The start time is field 22 of `/proc/<pid>/stat`, in clock ticks
/// since boot, which changes on every process — so a pid that has been reused
/// is detected rather than obeyed. No `unsafe`, no `libc`, no new dependency.
///
/// **The rule is: reclaim only what is provably dead.** Alive, or *undecidable*
/// — no `/proc` on this platform, an unparseable owner line, a lock written by
/// an older gunnar that recorded no start time — both mean refuse.
/// [`ArrowLogStore::force_unlock`] stays for exactly those cases. Reclaiming a
/// lock that is genuinely held lets two writers interleave frames and destroys
/// the log, which is far worse than an appliance that will not start.
#[derive(Debug)]
struct LockFile {
path: PathBuf,
}
/// Who a lock file says holds it.
#[derive(Debug, Clone, PartialEq, Eq)]
struct LockOwner {
pid: u32,
/// `None` for a lock file written before the liveness check, which recorded
/// only a pid.
/// It is what makes pid reuse undetectable, so it is treated as
/// undecidable rather than as "no reuse happened".
started: Option<u64>,
}
impl LockOwner {
fn parse(text: &str) -> Option<Self> {
let mut fields = text.split_whitespace();
let pid = fields.next()?.parse().ok()?;
let started = fields.next().and_then(|f| f.parse().ok());
Some(LockOwner { pid, started })
}
fn line(&self) -> String {
match self.started {
Some(started) => format!("{} {}\n", self.pid, started),
None => format!("{}\n", self.pid),
}
}
fn this_process() -> Self {
let pid = std::process::id();
LockOwner {
pid,
started: match proc_start_time(pid) {
ProcStartTime::Started(ticks) => Some(ticks),
_ => None,
},
}
}
}
/// What `/proc` says about a pid.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ProcStartTime {
/// The process exists and started at this tick.
Started(u64),
/// There is no such process. The one answer that permits a reclaim.
NoSuchProcess,
/// No `/proc`, or a `stat` that did not parse. **Not** the same as dead.
Unknown,
}
/// Read field 22 of `/proc/<pid>/stat` — the process's start time in clock
/// ticks since boot.
///
/// Field 2 is `comm`, which is parenthesised and may itself contain spaces and
/// close-parens, so the split is anchored on the **last** `)` in the line and
/// not on whitespace. Everything after it starts at field 3, which puts field
/// 22 at index 19.
fn proc_start_time(pid: u32) -> ProcStartTime {
// If this platform has no `/proc` at all, a missing `/proc/<pid>/stat` says
// nothing about the process, and reading it as "dead" would reclaim every
// live lock on that platform.
if !Path::new("/proc/self/stat").exists() {
return ProcStartTime::Unknown;
}
let stat = match std::fs::read_to_string(format!("/proc/{pid}/stat")) {
Ok(stat) => stat,
Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
return ProcStartTime::NoSuchProcess
}
Err(_) => return ProcStartTime::Unknown,
};
let Some((_, fields)) = stat.rsplit_once(')') else {
return ProcStartTime::Unknown;
};
// Field 3, the state. A zombie has been killed and is waiting to be
// collected by its parent: it holds no descriptors and can write nothing,
// so it is gone for the purpose of a lock — and its pid cannot be recycled
// while it sits there, so calling it gone is safe as well as true. Without
// this a crashed appliance whose reaper is slow still refuses to start.
if fields.split_whitespace().next() == Some("Z") {
return ProcStartTime::NoSuchProcess;
}
match fields
.split_whitespace()
.nth(19)
.and_then(|ticks| ticks.parse().ok())
{
Some(ticks) => ProcStartTime::Started(ticks),
None => ProcStartTime::Unknown,
}
}
/// Is the lock file's owner still running?
///
/// `false` only when the answer is **certain**. See [`LockFile`].
fn owner_is_gone(owner: &LockOwner) -> bool {
match (proc_start_time(owner.pid), owner.started) {
(ProcStartTime::NoSuchProcess, _) => true,
// The pid is live but is not the process that took the lock: it was
// recycled. This is the case a bare `/proc/<pid>` existence check gets
// wrong, and it gets it wrong in the direction of never recovering.
(ProcStartTime::Started(now), Some(then)) => now != then,
// A live pid whose identity cannot be confirmed, or no `/proc`.
(ProcStartTime::Started(_), None) | (ProcStartTime::Unknown, _) => false,
}
}
impl LockFile {
fn acquire(path: PathBuf) -> Result<Self> {
match Self::create(&path) {
Ok(lock) => Ok(lock),
Err(Error::StoreLocked { .. }) => Self::reclaim_if_stale(path),
Err(err) => Err(err),
}
}
fn create(path: &Path) -> Result<Self> {
match OpenOptions::new().write(true).create_new(true).open(path) {
Ok(mut file) => {
let owner = LockOwner::this_process();
// Not best-effort any more: an owner line that failed to reach
// the disk makes every later recovery undecidable, and a
// silent failure here would only be discovered after a crash.
if let Err(err) = write!(file, "{}", owner.line()).and_then(|()| file.sync_all()) {
tracing::warn!(
path = %path.display(),
%err,
"the store lock could not record its owner; a crash will need \
force_unlock"
);
}
Ok(LockFile {
path: path.to_path_buf(),
})
}
Err(err) if err.kind() == std::io::ErrorKind::AlreadyExists => {
Err(Error::StoreLocked {
path: path.to_path_buf(),
})
}
Err(err) => Err(Error::Io(err)),
}
}
fn reclaim_if_stale(path: PathBuf) -> Result<Self> {
let owner = std::fs::read_to_string(&path)
.ok()
.and_then(|text| LockOwner::parse(&text));
let Some(owner) = owner else {
tracing::warn!(
path = %path.display(),
"the store lock names no owner this gunnar can read, so it cannot be \
shown to be stale; refusing"
);
return Err(Error::StoreLocked { path });
};
if !owner_is_gone(&owner) {
return Err(Error::StoreLocked { path });
}
tracing::warn!(
path = %path.display(),
pid = owner.pid,
"the store lock was left by a process that is gone; reclaiming it"
);
match std::fs::remove_file(&path) {
Ok(()) => {}
Err(err) if err.kind() == std::io::ErrorKind::NotFound => {}
Err(err) => return Err(Error::Io(err)),
}
// Deliberately one attempt: if a second starter reclaimed the same
// stale lock a microsecond earlier, it now holds a live lock and this
// one must refuse. Retrying would be a loop that ends in two writers.
Self::create(&path)
}
}
impl Drop for LockFile {
fn drop(&mut self) {
if let Err(err) = std::fs::remove_file(&self.path) {
tracing::warn!(path = %self.path.display(), %err, "could not release the store lock");
}
}
}
// ───────────────────────────── the store ─────────────────────────────────────
/// The bytes of a store's secret log when there is no filesystem at all.
///
/// Hand the same handle back to [`ArrowLogStore::open_memory`] to simulate a
/// restart: the store is rebuilt by replaying these bytes, exactly as the file
/// backing replays a file.
#[derive(Clone, Debug, Default)]
pub struct MemoryLog {
secrets: Arc<Mutex<Vec<u8>>>,
}
struct Inner {
hash_kind: gix_hash::Kind,
actor: String,
policy: CompactionPolicy,
secrets_log: Mutex<Log>,
secrets_state: RwLock<BTreeMap<String, Vec<u8>>>,
_lock: Option<LockFile>,
}
/// One repository's secrets, as an Arrow IPC log.
///
/// Cheap to clone: every clone is the same store, and the exclusive lock lives
/// until the last handle is dropped.
#[derive(Clone)]
pub struct ArrowLogStore {
inner: Arc<Inner>,
}
impl std::fmt::Debug for ArrowLogStore {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ArrowLogStore")
.field("hash_kind", &self.inner.hash_kind)
.field("lock", &self.inner._lock)
.finish()
}
}
impl ArrowLogStore {
/// Open the logs in `dir`, creating them if absent, and take the exclusive lock.
///
/// Fails with [`Error::StoreLocked`] if another writer already holds it, and
/// with [`Error::HashKindMismatch`] if the log on disk belongs to a
/// repository of the other hash kind.
pub fn open(dir: impl AsRef<Path>, hash_kind: gix_hash::Kind) -> Result<Self> {
Self::open_inner(dir.as_ref(), hash_kind, true, CompactionPolicy::default())
}
/// [`ArrowLogStore::open`] with a chosen compaction policy.
///
/// The operator lever for threshold-based secret-log compaction. Replaced or
/// deleted secrets are compacted immediately regardless of this policy.
pub fn open_with_policy(
dir: impl AsRef<Path>,
hash_kind: gix_hash::Kind,
policy: CompactionPolicy,
) -> Result<Self> {
Self::open_inner(dir.as_ref(), hash_kind, true, policy)
}
fn open_inner(
dir: &Path,
hash_kind: gix_hash::Kind,
fsync: bool,
policy: CompactionPolicy,
) -> Result<Self> {
std::fs::create_dir_all(dir)?;
// The lock comes first: nothing may read a log it is not entitled to
// append to, because the replayed state is what CAS is checked against.
let lock = LockFile::acquire(dir.join(LOCK_FILE))?;
let open_file = |name: &str| -> Result<Backing> {
let path = dir.join(name);
let file = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(&path)?;
Ok(Backing::File { file, path })
};
Self::assemble(
hash_kind,
open_file(SECRETS_LOG)?,
Some(lock),
fsync,
policy,
)
}
/// A store with no file behind it. This is the no-OS shape, and what the
/// no-filesystem conformance run uses.
///
/// The framing, encoding and replay code is the same as the file backing's;
/// what this shape cannot speak to is durability, since there is nothing for
/// an fsync to mean. The durability claim is made by the file tests only.
pub fn in_memory(hash_kind: gix_hash::Kind) -> Result<(Self, MemoryLog)> {
let log = MemoryLog::default();
let store = Self::open_memory(&log, hash_kind)?;
Ok((store, log))
}
/// Rebuild a store by replaying `log`, the in-memory restart.
pub fn open_memory(log: &MemoryLog, hash_kind: gix_hash::Kind) -> Result<Self> {
Self::open_memory_with_policy(log, hash_kind, CompactionPolicy::default())
}
/// [`ArrowLogStore::open_memory`] with a chosen compaction policy.
pub fn open_memory_with_policy(
log: &MemoryLog,
hash_kind: gix_hash::Kind,
policy: CompactionPolicy,
) -> Result<Self> {
Self::assemble(
hash_kind,
Backing::Memory {
bytes: log.secrets.clone(),
},
None,
true,
policy,
)
}
fn assemble(
hash_kind: gix_hash::Kind,
secrets_backing: Backing,
lock: Option<LockFile>,
fsync: bool,
policy: CompactionPolicy,
) -> Result<Self> {
let (secrets_log, secret_batches) = Log::open(secrets_backing, secret_schema(), fsync)?;
let secrets_batches_on_open = secrets_log.batches;
let store = ArrowLogStore {
inner: Arc::new(Inner {
hash_kind,
actor: DEFAULT_ACTOR.to_owned(),
policy,
secrets_log: Mutex::new(secrets_log),
secrets_state: RwLock::new(replay_secrets(&secret_batches)?),
_lock: lock,
}),
};
if policy.should_compact(secrets_batches_on_open) {
if let Err(err) = store.compact_secrets_now() {
tracing::warn!(%err, "could not fold an oversized secret log at open");
}
}
Ok(store)
}
/// Remove a lock file left behind by a process that was killed.
///
/// Only ever correct when no gunnar is running against `dir`. The lock is
/// what keeps Arrow stream appends single-writer; breaking one that is
/// genuinely held lets two writers interleave frames and destroy the log.
///
/// # This is no longer the normal recovery path
///
/// [`open`](ArrowLogStore::open) reclaims a lock whose owner is **provably**
/// gone, so an appliance with no shell restarts after a power cut on its
/// own. What is left for this verb is the undecidable cases, where breaking
/// the lock is a judgement no program should make unasked: a platform with
/// no `/proc`, a lock file written by a gunnar older than the liveness
/// check, or one
/// whose owner line never reached the disk.
pub fn force_unlock(dir: impl AsRef<Path>) -> Result<()> {
match std::fs::remove_file(dir.as_ref().join(LOCK_FILE)) {
Ok(()) => Ok(()),
Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(err) => Err(Error::Io(err)),
}
}
/// The hash kind this repository uses.
pub fn hash_kind(&self) -> gix_hash::Kind {
self.inner.hash_kind
}
/// The compaction policy in force.
pub fn compaction_policy(&self) -> CompactionPolicy {
self.inner.policy
}
/// Fold the secret log: current values only, no history, no archive.
fn compact_secrets_now(&self) -> Result<()> {
let mut log = self
.inner
.secrets_log
.lock()
.map_err(|_| poisoned("secret log"))?;
compact_secrets(&self.inner, &mut log)
}
/// A handle implementing [`SecretStore`].
pub fn secrets(&self) -> ArrowLogSecretStore {
ArrowLogSecretStore {
inner: self.inner.clone(),
}
}
}
// ───────────────────────────── shared helpers ────────────────────────────────
fn now_unix_ms() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_millis() as i64)
.unwrap_or(0)
}
/// An Arrow fault, as a [`gunnar_store::Error`].
///
/// `gunnar-store` used to carry `impl From<arrow_schema::ArrowError> for Error`
/// so a `?` did this implicitly. It does not any more: that `From` was the last
/// Arrow in the contract crate, and keeping one impl there would have kept the
/// dependency edge this module's move to `gunnar-core` exists to cut (see
/// `gunnar_store`'s crate header). One function, called at every site the `?`
/// used to cover, rather than a second opinion about what an Arrow fault is.
pub(crate) fn arrow_err(err: arrow_schema::ArrowError) -> Error {
Error::backend(format!("arrow: {err}"))
}
/// The one door between bytes on disk and arrow-ipc's `StreamReader`: znippy's
/// IPC guard. arrow-ipc trusts the lengths a message declares, so a hostile
/// file does not give an `Err` there, it panics or aborts the process (a 1 TiB
/// `bodyLength` is allocated before a byte of it is read). The guard checks
/// every length arrow is about to trust against the bytes that are there; it
/// reads the metadata only and copies nothing. `stream` is a framed prefix
/// ([`valid_prefix_len`]), so a torn tail never reaches it: what it refuses is a
/// well-framed message that lies, which is an error and not a truncation.
pub(crate) fn ipc_guard(stream: &[u8]) -> Result<()> {
znippy_common::ipc_guard::check_stream(stream)
.map_err(|e| Error::backend(format!("arrow: refused by the IPC guard: {e}")))
}
pub(crate) fn column<'a, T: 'static>(batch: &'a RecordBatch, name: &str) -> Result<&'a T> {
batch
.column_by_name(name)
.and_then(|c| c.as_any().downcast_ref::<T>())
.ok_or_else(|| {
Error::backend(format!(
"the Arrow log has no usable `{name}` column; the file is not a gunnar log"
))
})
}
// ───────────────────────────── compaction ────────────────────────────────────
/// The same fold for secrets, with **no archive and no window**.
///
/// Retaining superseded secret values would make a rotated host key immortal,
/// which is the opposite of what rotation is for.
fn compact_secrets(inner: &Inner, log: &mut Log) -> Result<()> {
let snapshot = {
let state = inner.secrets_state.read().map_err(|_| poisoned("secret"))?;
let mut keys = StringBuilder::new();
let mut values = BinaryBuilder::new();
let mut actors = StringBuilder::new();
let mut ats = Int64Builder::new();
let at = now_unix_ms();
for (key, value) in state.iter() {
keys.append_value(key);
values.append_value(value);
actors.append_value(COMPACTION_ACTOR);
ats.append_value(at);
}
RecordBatch::try_new(
log.schema.clone(),
vec![
Arc::new(keys.finish()),
Arc::new(values.finish()),
Arc::new(actors.finish()),
Arc::new(ats.finish()),
],
)
.map_err(arrow_err)?
};
log.rewrite(&[snapshot])
}
// ───────────────────────────── secrets ───────────────────────────────────────
/// Secrets in an Arrow IPC log: the SSH host key and the admin bootstrap
/// material, on a box with no home directory.
#[derive(Clone)]
pub struct ArrowLogSecretStore {
inner: Arc<Inner>,
}
impl std::fmt::Debug for ArrowLogSecretStore {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str("ArrowLogSecretStore")
}
}
fn build_secret_batch(
schema: &SchemaRef,
actor: &str,
at: i64,
key: &str,
value: Option<&[u8]>,
) -> Result<RecordBatch> {
let mut keys = StringBuilder::new();
let mut values = BinaryBuilder::new();
let mut actors = StringBuilder::new();
let mut ats = Int64Builder::new();
keys.append_value(key);
match value {
Some(bytes) => values.append_value(bytes),
None => values.append_null(),
}
actors.append_value(actor);
ats.append_value(at);
RecordBatch::try_new(
schema.clone(),
vec![
Arc::new(keys.finish()),
Arc::new(values.finish()),
Arc::new(actors.finish()),
Arc::new(ats.finish()),
],
)
.map_err(arrow_err)
}
fn replay_secrets(batches: &[RecordBatch]) -> Result<BTreeMap<String, Vec<u8>>> {
let mut state = BTreeMap::new();
for batch in batches {
let keys = column::<StringArray>(batch, COL_KEY)?;
let values = column::<BinaryArray>(batch, COL_VALUE)?;
for row in 0..batch.num_rows() {
let key = keys.value(row).to_owned();
if values.is_null(row) {
state.remove(&key);
} else {
state.insert(key, values.value(row).to_vec());
}
}
}
Ok(state)
}
impl ArrowLogSecretStore {
fn write(&self, key: &str, value: Option<&[u8]>) -> Result<()> {
let mut log = self
.inner
.secrets_log
.lock()
.map_err(|_| poisoned("secret log"))?;
self.write_locked(&mut log, key, value)
}
fn write_locked(&self, log: &mut Log, key: &str, value: Option<&[u8]>) -> Result<()> {
let batch = build_secret_batch(&log.schema, &self.inner.actor, now_unix_ms(), key, value)?;
// Append AND barrier, both under the writer mutex — deliberately *not*
// the group commit the ref log does.
//
// A secret is written when a host key is minted or rotated, which is
// rare and never concurrent, so there is nothing to group and releasing
// the lock first would buy nothing while widening the window in which a
// rotated secret is still readable. `append` only stages; forgetting
// this line is a secret that is acknowledged and not on the platter,
// which is what `a_secret_is_durable_before_the_write_returns` exists to
// catch.
let sequence = log.append(&batch)?;
log.barrier(sequence)?;
let superseded = {
let mut state = self
.inner
.secrets_state
.write()
.map_err(|_| poisoned("secret"))?;
match value {
Some(bytes) => state.insert(key.to_owned(), bytes.to_vec()).is_some(),
None => state.remove(key).is_some(),
}
};
// A superseded secret is compacted out **now**, not when a batch count
// says so.
//
// This module's own header says "a rotated SSH host key must actually
// stop existing". It did not: the log is append-only and the only
// trigger was `should_compact(log.batches)`, so the previous value sat
// in `secrets.arrow` for up to `after_batches` further writes — on a
// store that sees a handful of writes in its lifetime, that is
// *forever*. Found by `gunnar-core`'s sealing-migration test, which sealed a
// legacy plaintext host key and then found the plaintext still in the
// file.
//
// Refs deliberately keep their history; secrets are the one log where
// the old value is the defect. The fold is a rewrite of a map with a
// handful of entries, so doing it on every replacement costs nothing
// that a threshold was protecting.
if superseded {
compact_secrets(&self.inner, log).map_err(|err| {
Error::backend(format!(
"the new value for the secret {key} was stored, but the PREVIOUS value \
could not be compacted out of the log and is still on disk: {err}"
))
})?;
} else if self.inner.policy.should_compact(log.batches) {
if let Err(err) = compact_secrets(&self.inner, log) {
tracing::warn!(%err, "could not compact the secret log; it keeps growing");
}
}
Ok(())
}
fn current(&self, key: &str) -> Result<Option<Vec<u8>>> {
Ok(self
.inner
.secrets_state
.read()
.map_err(|_| poisoned("secret"))?
.get(key)
.cloned())
}
}
impl SecretStore for ArrowLogSecretStore {
fn get(&self, key: &str) -> Result<Option<Vec<u8>>> {
self.current(key)
}
fn put(&self, key: &str, value: &[u8]) -> Result<()> {
self.write(key, Some(value))
}
fn delete(&self, key: &str) -> Result<()> {
// A tombstone for a key that is not there would grow the log for
// nothing, and "delete what is absent" is explicitly not an error.
if self.current(key)?.is_none() {
return Ok(());
}
self.write(key, None)
}
fn get_or_create(&self, key: &str, mk: &mut dyn FnMut() -> Result<Vec<u8>>) -> Result<Vec<u8>> {
// The writer mutex is taken *before* the existence check, so the loser of
// a race blocks here and then finds the winner's value. `mk` runs once.
let mut log = self
.inner
.secrets_log
.lock()
.map_err(|_| poisoned("secret log"))?;
if let Some(existing) = self.current(key)? {
return Ok(existing);
}
let value = mk()?;
self.write_locked(&mut log, key, Some(&value))?;
Ok(value)
}
}
// ───────────────────────────── tests ─────────────────────────────────────────
#[cfg(test)]
mod tests {
use super::*;
use gunnar_store::SecretStoreExt;
const KIND: gix_hash::Kind = gix_hash::Kind::Sha256;
fn dir() -> tempfile::TempDir {
tempfile::tempdir().expect("a temporary directory")
}
#[test]
fn the_in_memory_secret_log_replays_across_a_rebuild() {
let (store, log) = ArrowLogStore::in_memory(KIND).expect("in-memory store");
store.secrets().put("k", b"v").expect("put");
drop(store);
let rebuilt = ArrowLogStore::open_memory(&log, KIND).expect("replay");
assert_eq!(
rebuilt.secrets().get("k").expect("get"),
Some(b"v".to_vec())
);
}
#[test]
fn a_secret_delete_is_a_tombstone_that_survives_a_reopen() {
let dir = dir();
{
let store = ArrowLogStore::open(dir.path(), KIND).expect("open");
store.secrets().put("k", b"v1").expect("put");
store.secrets().delete("k").expect("delete");
}
let store = ArrowLogStore::open(dir.path(), KIND).expect("reopen");
assert!(store.secrets().get("k").expect("get").is_none());
store.secrets().put("k", b"v2").expect("put again");
assert_eq!(store.secrets().get("k").expect("get"), Some(b"v2".to_vec()));
}
#[test]
fn a_rotated_secret_leaves_no_trace_of_the_previous_value_on_disk() {
let dir = dir();
let old = b"OLD-SECRET-VALUE-that-must-not-survive-rotation";
let new = b"NEW-SECRET-VALUE-after-the-rotation";
let store = ArrowLogStore::open(dir.path(), KIND).expect("open");
store.secrets().put("ssh/host_ed25519", old).expect("put");
let contains = |needle: &[u8]| {
let bytes = std::fs::read(dir.path().join(SECRETS_LOG)).expect("read the secret log");
assert!(!bytes.is_empty(), "the secret log is empty");
bytes.windows(needle.len()).any(|w| w == needle)
};
assert!(contains(old));
store
.secrets()
.put("ssh/host_ed25519", new)
.expect("rotate");
assert!(!contains(old), "the previous secret survived rotation");
assert!(contains(new), "the new value is not in the log");
assert_eq!(
store.secrets().get("ssh/host_ed25519").expect("get"),
Some(new.to_vec())
);
store.secrets().delete("ssh/host_ed25519").expect("delete");
assert!(
!contains(new),
"a deleted secret's value is still in the log"
);
}
#[test]
fn compacting_secrets_does_not_retain_superseded_values() {
let dir = dir();
let store = ArrowLogStore::open_with_policy(dir.path(), KIND, CompactionPolicy::new(4, 0))
.expect("open");
for generation in 0..12 {
store
.secrets()
.put(
"host-key",
format!("secret-generation-{generation:02}").as_bytes(),
)
.expect("put");
}
let haystack = String::from_utf8_lossy(
&std::fs::read(dir.path().join(SECRETS_LOG)).expect("read the secret log"),
)
.into_owned();
assert!(!haystack.contains("secret-generation-00"));
assert!(haystack.contains("secret-generation-11"));
assert!(!dir.path().join(REFS_ARCHIVE_DIR).exists());
drop(store);
let reopened =
ArrowLogStore::open_with_policy(dir.path(), KIND, CompactionPolicy::new(4, 0))
.expect("reopen");
assert_eq!(
reopened.secrets().get("host-key").expect("get"),
Some(b"secret-generation-11".to_vec())
);
}
#[test]
fn get_or_create_does_not_remint_across_a_reopen() {
let dir = dir();
let first = {
let store = ArrowLogStore::open(dir.path(), KIND).expect("open");
store
.secrets()
.get_or_create_with(gunnar_store::SSH_HOST_KEY, || Ok(b"host-key-1".to_vec()))
.expect("mint")
};
let store = ArrowLogStore::open(dir.path(), KIND).expect("reopen");
let second = store
.secrets()
.get_or_create_with(gunnar_store::SSH_HOST_KEY, || {
panic!("the host key was re-minted after a restart")
})
.expect("read back");
assert_eq!(first, second);
}
#[test]
fn a_secret_is_durable_before_the_write_returns() {
let dir = dir();
let store = ArrowLogStore::open(dir.path(), KIND).expect("open");
store
.secrets()
.put(gunnar_store::SSH_HOST_KEY, b"a-host-key")
.expect("write a secret");
let on_disk = std::fs::metadata(dir.path().join(SECRETS_LOG))
.expect("stat")
.len();
let durability = &store
.inner
.secrets_log
.lock()
.expect("the writer")
.durability;
let durable = durability.durable.load(Ordering::SeqCst);
let written = durability.written.load(Ordering::SeqCst);
assert_eq!(written, on_disk);
assert_eq!(durable, written);
assert!(durability.syncs.load(Ordering::SeqCst) >= 2);
}
}