gunnar_core/

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();
            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}"))
}

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);
    }
}