znippy_plugin_git/

pushlog.rs

//! The append-only **push log**: the one mechanism `__gunnar_refs__` and
//! `__gunnar_secrets__` are both built from.
//!
//! ## Why there is no database here
//!
//! D18 removed redb: it was a database wedged between two archive formats, and
//! it bought nothing that the formats did not already provide. What a ref update
//! actually needs is *atomicity* — a push is visible whole or not at all — and
//! Arrow IPC already frames its messages. So the transaction boundary is the
//! **IPC frame boundary**, and the rule is one `RecordBatch` per push.
//!
//! ## The on-disk shape
//!
//! The live log is a concatenation of **self-contained** Arrow IPC streams, one
//! per push:
//!
//! ```text
//! [ schema | RecordBatch | EOS ]  ← push 1, fsynced
//! [ schema | RecordBatch | EOS ]  ← push 2, fsynced
//! [ schema | RecordBatch          ← push 3, TORN by a crash
//! ```
//!
//! Repeating the schema per frame is what buys the property: every frame is
//! independently parseable, so recovery needs no side-car, no length table and
//! no journal.
//!
//! **What that costs, measured** (refs schema, 4 refs per push, 2026-08-03):
//! a frame is 2056 bytes — 448 of schema (21.8%) and 1600 of batch (77.8%,
//! i.e. 400 bytes per ref). So the schema repetition is the *smaller* half of
//! the overhead; the larger half is Arrow's 64-byte buffer alignment paid six
//! times over on a four-row batch. Merging frames into one stream would
//! therefore recover only ~21.8%.
//!
//! The real win is compaction — folding many small batches into one large one,
//! which amortises the padding as well as the schema. `gunnar-store`'s
//! `arrow_log.rs` already does exactly this (`CompactionPolicy`, a staged
//! rewrite plus rename, and a reflog archive for the history it retires), and
//! its single-stream live log carries the schema once rather than per push.
//! That is the shape to adopt here rather than reinvent; see the note in the
//! commit that added this measurement. [`scan_frames`] reads forward while
//! frames parse and stops at the first one that does not — push 3 above is
//! discarded whole, pushes 1 and 2 survive intact. That is the crash story, and
//! it is a property of the framing rather than of any code that has to run.
//!
//! A partially-written frame can never be mistaken for a complete one: Arrow
//! prefixes every message with a continuation marker and a length, so a truncated
//! frame ends in the middle of a message body and the reader errors instead of
//! yielding a half-populated batch.
//!
//! ## Live log vs sealed archive
//!
//! znippy seals later. Until then the log is the file above. At seal, the
//! recovered batches are folded into **one** reserved Arrow section that keeps
//! one RecordBatch per push — so the push boundaries survive into the sealed
//! archive, and DuckDB / Polars / DataFusion read the section straight out of its
//! manifest byte range with no gunnar code.

use std::fs::{File, OpenOptions};
use std::io::{Cursor, Write};
use std::path::{Path, PathBuf};
use std::sync::Arc;

use anyhow::{Result, anyhow};
use znippy_common::arrow::datatypes::Schema;
use znippy_common::arrow::ipc::reader::StreamReader;
use znippy_common::arrow::ipc::writer::StreamWriter;
use znippy_common::arrow::record_batch::RecordBatch;
use znippy_common::{ReservedSection, read_reserved_section_bytes};

/// What a scan of a log recovered, and what it had to throw away.
#[derive(Debug, Default)]
pub struct PushLogScan {
    /// The complete frames, in push order. One entry per push.
    pub pushes: Vec<RecordBatch>,
    /// Bytes at the tail that did not form a complete frame — a push that was
    /// interrupted. Zero for a cleanly closed log.
    ///
    /// Deliberately reported rather than silently dropped: "the last push did
    /// not land" is information the caller may need, and a scan that quietly
    /// swallowed it would be indistinguishable from a clean log.
    pub torn_tail_bytes: u64,
}

impl PushLogScan {
    pub fn is_clean(&self) -> bool {
        self.torn_tail_bytes == 0
    }
}

/// Serialize one push as a self-contained Arrow IPC stream frame.
pub fn encode_frame(schema: &Arc<Schema>, batch: &RecordBatch) -> Result<Vec<u8>> {
    let mut buf = Vec::new();
    {
        let mut w = StreamWriter::try_new(&mut buf, schema)
            .map_err(|e| anyhow!("push log: opening a frame: {e}"))?;
        w.write(batch).map_err(|e| anyhow!("push log: writing a frame: {e}"))?;
        w.finish().map_err(|e| anyhow!("push log: closing a frame: {e}"))?;
    }
    Ok(buf)
}

/// Arrow's end-of-stream marker: the continuation marker `0xFFFFFFFF` followed
/// by a zero metadata length. Its presence at the end of a frame is what proves
/// the writer got all the way through `finish()`.
const IPC_EOS: [u8; 8] = [0xFF, 0xFF, 0xFF, 0xFF, 0x00, 0x00, 0x00, 0x00];

/// Read forward over concatenated frames, stopping at the first that does not
/// parse. Never errors on a torn tail — that is the expected state after a crash
/// and the whole reason the framing was chosen.
///
/// ## Why "did the reader error?" is not enough
///
/// Arrow's `StreamReader` treats an unexpected EOF as a clean end of stream, so
/// a frame truncated *inside* its record-batch message parses without error and
/// simply yields no batch. Trusting the reader's verdict alone therefore
/// accepted a torn push as a complete-but-empty frame and advanced past it —
/// the torn bytes were silently swallowed, `torn_tail_bytes` reported 0, and a
/// crashed push was indistinguishable from a clean log.
///
/// Two positive checks close that, and both must hold for a frame to count:
///
/// 1. it yielded **exactly one** `RecordBatch` — the format's invariant is one
///    push, one batch, so zero batches is a truncation and two is not our frame;
/// 2. its last eight bytes are [`IPC_EOS`] — proof the writer reached `finish()`
///    rather than the reader reaching the end of the file.
pub fn scan_frames(bytes: &[u8]) -> PushLogScan {
    let mut out = PushLogScan::default();
    let mut off = 0usize;

    while off < bytes.len() {
        // Only the whole, well-formed messages of the next frame reach arrow
        // (the IPC guard's valid prefix stops at its end-of-stream marker): a
        // hostile frame then reads as a torn one, never as an arrow panic or
        // abort, and the checks below see exactly the bytes the reader consumed.
        let rest = znippy_common::ipc_guard::valid_prefix(&bytes[off..]);
        let Ok(mut reader) = StreamReader::try_new(Cursor::new(rest), None) else {
            break;
        };
        let mut batches = Vec::new();
        let mut errored = false;
        loop {
            match reader.next() {
                Some(Ok(b)) => batches.push(b),
                Some(Err(_)) => {
                    errored = true;
                    break;
                }
                None => break,
            }
        }
        if errored {
            break;
        }
        // The reader is unbuffered over a `Cursor`, so its position is exactly
        // the bytes this frame consumed.
        let consumed = reader.get_ref().position() as usize;

        // A frame that consumed nothing would spin forever; one that did not end
        // in EOS, or did not carry exactly one batch, was torn mid-write.
        if consumed < IPC_EOS.len()
            || batches.len() != 1
            || rest[consumed - IPC_EOS.len()..consumed] != IPC_EOS
        {
            break;
        }

        out.pushes.append(&mut batches);
        off += consumed;
    }

    out.torn_tail_bytes = (bytes.len() - off) as u64;
    out
}

/// When a log should be compacted.
///
/// Compaction folds many small frames into one large one. The point is NOT the
/// schema — measured, that is only 21.8% of a frame — but Arrow's 64-byte
/// buffer alignment, which a four-row batch pays six times over and a
/// twenty-thousand-row batch pays once.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct CompactionPolicy {
    /// Compact once the log holds at least this many frames. Zero disables it.
    pub after_frames: usize,
}

impl CompactionPolicy {
    pub const fn never() -> Self {
        Self { after_frames: 0 }
    }

    pub fn should_compact(&self, frames: usize) -> bool {
        self.after_frames != 0 && frames >= self.after_frames
    }
}

impl Default for CompactionPolicy {
    fn default() -> Self {
        Self { after_frames: 1024 }
    }
}

/// What one compaction did.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CompactionReport {
    pub frames_before: usize,
    pub frames_after: usize,
    pub rows: usize,
    pub bytes_before: u64,
    pub bytes_after: u64,
}

/// The scratch name a compaction stages under, 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.
fn sync_parent_dir(path: &Path) -> Result<()> {
    let parent = match path.parent() {
        Some(p) if !p.as_os_str().is_empty() => p.to_path_buf(),
        _ => PathBuf::from("."),
    };
    File::open(parent)?.sync_all()?;
    Ok(())
}

/// Where a compaction is allowed to stop. Only the crash test constructs
/// anything but [`Finish::Swap`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Finish {
    /// Stage the replacement, fsync it, and rename it into place.
    Swap,
    /// Stage and fsync, then stop. What a crash mid-compaction leaves on disk:
    /// a complete replacement that nothing points at yet.
    StopBeforeSwap,
}

/// An append-only log file. One [`append`](Self::append) call is one push.
pub struct PushLog {
    path: PathBuf,
    schema: Arc<Schema>,
    /// How many times this handle has read the whole file — the only `O(log)`
    /// operation it has. Applied output, not a switch: a caller that claims a
    /// push costs one append can assert this did not move, which is a fact a
    /// clock cannot establish. See [`crate::refs::RefLog`].
    scans: std::sync::atomic::AtomicU64,
}

impl PushLog {
    pub fn new(path: impl Into<PathBuf>, schema: Arc<Schema>) -> Self {
        Self {
            path: path.into(),
            schema,
            scans: std::sync::atomic::AtomicU64::new(0),
        }
    }

    /// Full passes over the file this handle has paid, since it was made.
    pub fn scans(&self) -> u64 {
        self.scans.load(std::sync::atomic::Ordering::Relaxed)
    }

    pub fn path(&self) -> &Path {
        &self.path
    }

    pub fn schema(&self) -> &Arc<Schema> {
        &self.schema
    }

    /// Append one push and fsync it. Returns the byte offset the frame starts
    /// at.
    ///
    /// The frame is built **fully in memory first**, so the single `write_all`
    /// is the only thing the crash window covers; and the `sync_all` is what
    /// makes "the frame is complete on disk" — the durability claim this format
    /// rests on — actually true rather than merely likely.
    pub fn append(&self, batch: &RecordBatch) -> Result<u64> {
        self.append_with(batch, true)
    }

    /// [`append`](Self::append), with the barrier as an argument.
    ///
    /// `durable = false` skips the `fsync` and **nothing else**: the frame is
    /// still built whole in memory and written with one `write_all`, so it is
    /// still whole-or-absent — that property comes from the framing, not from
    /// the barrier. What is given up is only *when* it reaches the platter.
    ///
    /// The one caller is [`crate::refs::RefLog::push_rebuildable`], for a frame
    /// whose content is re-derivable from a fact that is already durable
    /// somewhere else. Anything a client was told landed goes through
    /// [`append`](Self::append).
    pub fn append_with(&self, batch: &RecordBatch, durable: bool) -> Result<u64> {
        if batch.schema() != self.schema {
            return Err(anyhow!(
                "push log {}: batch schema does not match the log schema",
                self.path.display()
            ));
        }
        let frame = encode_frame(&self.schema, batch)?;
        let mut f = OpenOptions::new()
            .create(true)
            .append(true)
            .open(&self.path)
            .map_err(|e| anyhow!("push log {}: {e}", self.path.display()))?;
        let offset = f.metadata()?.len();
        f.write_all(&frame)?;
        if durable {
            f.sync_all()?;
        }
        Ok(offset)
    }

    /// Recover the log. A missing file is an empty, clean log — a repository
    /// that has never been pushed to is not an error.
    pub fn scan(&self) -> Result<PushLogScan> {
        self.scans.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        match std::fs::read(&self.path) {
            Ok(b) => Ok(scan_frames(&b)),
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(PushLogScan::default()),
            Err(e) => Err(anyhow!("push log {}: {e}", self.path.display())),
        }
    }

    /// How many complete frames the log holds.
    pub fn frame_count(&self) -> Result<usize> {
        Ok(self.scan()?.pushes.len())
    }

    /// Fold every frame into one, atomically.
    ///
    /// ## What it costs and what it buys
    ///
    /// Rows are concatenated, never dropped: `push_seq` and `updated_ms` ride
    /// in the columns, so the full history survives and [`crate::refs::fold`]
    /// gives the identical answer before and after. What is given up is the
    /// *frame* boundary as a record of where one push ended — after compaction
    /// a frame is a run of pushes, and `push_seq` is the only thing that says
    /// where the seams were. That is why the ordering key was never the frame
    /// index.
    ///
    /// ## Compatibility — no migration, no version bump
    ///
    /// A compacted log is still exactly what an uncompacted one is: a sequence
    /// of self-contained frames, each carrying one batch and ending in EOS.
    /// There are simply fewer and larger ones. [`scan_frames`] already reads any
    /// number of them and its one-batch-per-frame invariant still holds, so a
    /// log written before this existed reads unchanged and a compacted log
    /// reads on any reader that could read the old one. This is deliberately
    /// **not** a versioned format change.
    ///
    /// ## Crash safety
    ///
    /// The same shape as `gunnar-store`'s `arrow_log.rs` — which owns the
    /// mature version of this problem — and deliberately so, since the code
    /// cannot be shared across the repository boundary: stage the whole
    /// replacement under [`compacting_path`], fsync it, rename it over the live
    /// log, then fsync the parent directory so the rename itself is durable.
    /// The live log is never written in place, so an interruption at *any* byte
    /// leaves the old log serving untouched; the worst outcome is a stray
    /// `.compacting` file that the next compaction overwrites.
    pub fn compact(&self) -> Result<CompactionReport> {
        self.compact_with(Finish::Swap)
    }

    /// [`compact`](Self::compact), stopping where `finish` says. The crash test
    /// is the only caller that passes anything but [`Finish::Swap`].
    pub fn compact_with(&self, finish: Finish) -> Result<CompactionReport> {
        let scan = self.scan()?;
        let frames_before = scan.pushes.len();
        let bytes_before = std::fs::metadata(&self.path).map(|m| m.len()).unwrap_or(0);

        if frames_before == 0 {
            return Ok(CompactionReport {
                frames_before,
                frames_after: frames_before,
                rows: 0,
                bytes_before,
                bytes_after: bytes_before,
            });
        }

        let merged = znippy_common::arrow::compute::concat_batches(&self.schema, scan.pushes.iter())
            .map_err(|e| anyhow!("compacting {}: {e}", self.path.display()))?;
        let rows = merged.num_rows();
        let bytes = encode_frame(&self.schema, &merged)?;

        // Stage the whole replacement first. Nothing points at it yet, so a
        // crash anywhere in here is invisible to a reader.
        let staged = compacting_path(&self.path);
        {
            let mut f = File::create(&staged)?;
            f.write_all(&bytes)?;
            f.sync_all()?;
        }
        if finish == Finish::StopBeforeSwap {
            return Ok(CompactionReport {
                frames_before,
                frames_after: 1,
                rows,
                bytes_before,
                bytes_after: bytes.len() as u64,
            });
        }

        std::fs::rename(&staged, &self.path)?;
        sync_parent_dir(&self.path)?;

        Ok(CompactionReport {
            frames_before,
            frames_after: 1,
            rows,
            bytes_before,
            bytes_after: bytes.len() as u64,
        })
    }

    /// Compact if `policy` says the log has grown enough. Returns `None` when
    /// it did not run.
    pub fn maybe_compact(&self, policy: CompactionPolicy) -> Result<Option<CompactionReport>> {
        if policy == CompactionPolicy::never() {
            return Ok(None);
        }
        let frames = self.frame_count()?;
        if !policy.should_compact(frames) {
            return Ok(None);
        }
        self.compact().map(Some)
    }

    /// Fold the recovered pushes into the reserved section to seal into an
    /// archive, preserving one RecordBatch per push.
    pub fn seal_section(&self, module_name: &str) -> Result<ReservedSection> {
        let scan = self.scan()?;
        Ok(ReservedSection::arrow(module_name, self.schema.clone(), scan.pushes))
    }
}

/// Read a sealed push-log section back out of an archive. `Ok(None)` when the
/// archive carries no such section — distinct from a section with no pushes.
pub fn read_sealed(archive: &Path, module_name: &str) -> Result<Option<Vec<RecordBatch>>> {
    let Some(bytes) = read_reserved_section_bytes(archive, module_name)? else {
        return Ok(None);
    };
    decode_sealed(module_name, &bytes).map(Some)
}

/// The batches of a sealed push-log section's bytes.
fn decode_sealed(module_name: &str, bytes: &[u8]) -> Result<Vec<RecordBatch>> {
    // Through the IPC guard: the section comes out of an archive.
    let reader = znippy_common::ipc_guard::stream_reader(bytes)
        .map_err(|e| anyhow!("{module_name}: {e}"))?;
    let mut out = Vec::new();
    for b in reader {
        out.push(b.map_err(|e| anyhow!("{module_name}: {e}"))?);
    }
    Ok(out)
}

/// Truncate a log to `len` bytes — the crash simulator the tests need, and the
/// only supported way to produce a torn tail deliberately.
#[doc(hidden)]
pub fn truncate_for_test(path: &Path, len: u64) -> Result<()> {
    let f = File::options().write(true).open(path)?;
    f.set_len(len)?;
    Ok(())
}

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

    /// A sealed section is a section of an archive someone else may have
    /// written: hostile bytes are an `Err`, never a panic or an abort.
    #[test]
    #[ignore = "run as a child by a_hostile_sealed_section_is_an_error"]
    fn hostile_child() {
        if crate::hostile_inputs::child_case().is_none() {
            return;
        }
        let mut valid = Vec::new();
        {
            let b = batch(&["refs/heads/main", "refs/tags/v1"], 7);
            let mut w = znippy_common::arrow::ipc::writer::StreamWriter::try_new(&mut valid, &schema()).unwrap();
            w.write(&b).unwrap();
            w.write(&batch(&["x"], 8)).unwrap();
            w.finish().unwrap();
        }
        assert_eq!(decode_sealed("t", &valid).unwrap().len(), 2);
        crate::hostile_inputs::sweep(&valid, |_, b| {
            let _ = decode_sealed("t", b);
        });
    }

    #[test]
    fn a_hostile_sealed_section_is_an_error() {
        crate::hostile_inputs::run_in_child("pushlog::tests::hostile_child", "sealed");
    }
    use znippy_common::arrow::array::{StringArray, UInt64Array};
    use znippy_common::arrow::datatypes::{DataType, Field};

    fn schema() -> Arc<Schema> {
        Arc::new(Schema::new(vec![
            Field::new("name", DataType::Utf8, false),
            Field::new("seq", DataType::UInt64, false),
        ]))
    }

    fn batch(names: &[&str], seq: u64) -> RecordBatch {
        RecordBatch::try_new(
            schema(),
            vec![
                Arc::new(StringArray::from(names.to_vec())),
                Arc::new(UInt64Array::from(vec![seq; names.len()])),
            ],
        )
        .unwrap()
    }

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

    /// Each push is its own frame, and the boundaries survive the round trip —
    /// three pushes come back as three batches, not one merged batch. If the
    /// framing collapsed, `pushes.len()` would be 1 and the per-push transaction
    /// boundary this format rests on would not exist.
    #[test]
    fn one_push_is_one_frame_is_one_batch() {
        let dir = tmpdir("frames");
        let log = PushLog::new(dir.join("refs.log"), schema());
        log.append(&batch(&["a", "b"], 1)).unwrap();
        log.append(&batch(&["c"], 2)).unwrap();
        log.append(&batch(&["d", "e", "f"], 3)).unwrap();

        let scan = log.scan().unwrap();
        assert!(scan.is_clean(), "no crash happened, tail must be clean");
        assert_eq!(scan.pushes.len(), 3, "one RecordBatch per push");
        assert_eq!(scan.pushes[0].num_rows(), 2);
        assert_eq!(scan.pushes[1].num_rows(), 1);
        assert_eq!(scan.pushes[2].num_rows(), 3);

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

    /// The crash test. A push interrupted part-way must lose **that push and
    /// only that push**; every earlier push stays readable.
    ///
    /// The truncation is swept across the whole final frame rather than taken at
    /// one convenient offset, because a recovery that works at one cut point and
    /// not another is not recovery. At every cut the two surviving pushes must
    /// come back byte-identical, and the third must never appear in any partial
    /// form.
    #[test]
    fn a_torn_final_frame_loses_only_that_push() {
        let dir = tmpdir("torn");
        let path = dir.join("refs.log");
        let log = PushLog::new(&path, schema());
        log.append(&batch(&["a", "b"], 1)).unwrap();
        log.append(&batch(&["c"], 2)).unwrap();
        let third_start = log.append(&batch(&["doomed"], 3)).unwrap();
        let full_len = std::fs::metadata(&path).unwrap().len();
        assert!(full_len > third_start, "third frame must have real bytes");

        let intact = std::fs::read(&path).unwrap();

        // Cut anywhere inside the third frame: 1 byte in, through 1 byte short.
        for cut in (third_start + 1)..full_len {
            std::fs::write(&path, &intact).unwrap();
            truncate_for_test(&path, cut).unwrap();

            let scan = log.scan().unwrap();
            assert_eq!(
                scan.pushes.len(),
                2,
                "cut at {cut}: the torn push must vanish whole, leaving exactly the two \
                 that completed (got {} batches)",
                scan.pushes.len()
            );
            assert_eq!(scan.pushes[0].num_rows(), 2, "cut at {cut}: push 1 damaged");
            assert_eq!(scan.pushes[1].num_rows(), 1, "cut at {cut}: push 2 damaged");
            assert_eq!(
                scan.torn_tail_bytes,
                cut - third_start,
                "cut at {cut}: the torn tail must be reported, not silently swallowed"
            );
            let names = scan.pushes[0]
                .column(0)
                .as_any()
                .downcast_ref::<StringArray>()
                .unwrap();
            assert_eq!(names.value(0), "a", "cut at {cut}: push 1 content damaged");
            assert_eq!(names.value(1), "b");
        }

        // Restoring the full bytes brings the third push back — proving the loss
        // was the truncation and not the scanner dropping a trailing frame.
        std::fs::write(&path, &intact).unwrap();
        let scan = log.scan().unwrap();
        assert_eq!(scan.pushes.len(), 3);
        assert!(scan.is_clean());

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

    /// Garbage appended after a clean log is a torn tail, not a parse failure
    /// and not a panic: recovery must be total over arbitrary trailing bytes.
    #[test]
    fn trailing_garbage_is_a_torn_tail_not_an_error() {
        let dir = tmpdir("garbage");
        let path = dir.join("refs.log");
        let log = PushLog::new(&path, schema());
        log.append(&batch(&["a"], 1)).unwrap();

        for junk in [
            &b"\x00"[..],
            &b"\xff\xff\xff\xff"[..],
            &b"\xff\xff\xff\xff\x10\x00\x00\x00partial"[..],
            &[0xAB; 4096][..],
        ] {
            let mut bytes = std::fs::read(&path).unwrap();
            let clean_len = bytes.len();
            bytes.extend_from_slice(junk);
            let scan = scan_frames(&bytes);
            assert_eq!(scan.pushes.len(), 1, "the complete push must survive {junk:?}");
            assert_eq!(scan.torn_tail_bytes, (bytes.len() - clean_len) as u64);
        }

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

    /// An empty / never-written log is clean and empty, not an error. A
    /// repository that has never been pushed to is a normal state.
    #[test]
    fn a_missing_log_is_empty_and_clean() {
        let dir = tmpdir("missing");
        let log = PushLog::new(dir.join("nope.log"), schema());
        let scan = log.scan().unwrap();
        assert!(scan.pushes.is_empty());
        assert!(scan.is_clean());
        std::fs::remove_dir_all(&dir).ok();
    }

    /// Compaction preserves every row and every push_seq, so the fold is
    /// identical before and after — and it must actually shrink the log, which
    /// is the only reason to run it.
    #[test]
    fn compaction_preserves_every_row_and_its_order() {
        let dir = tmpdir("compact");
        let path = dir.join("refs.log");
        let log = PushLog::new(&path, schema());
        for i in 0..200u64 {
            log.append(&batch(&["a", "b", "c", "d"], i)).unwrap();
        }

        let before = log.scan().unwrap();
        let before_rows: usize = before.pushes.iter().map(|b| b.num_rows()).sum();
        assert_eq!(before.pushes.len(), 200);

        let report = log.compact().unwrap();
        assert_eq!(report.frames_before, 200);
        assert_eq!(report.frames_after, 1, "everything must fold into one frame");
        assert_eq!(report.rows, before_rows, "compaction must not drop a row");

        let after = log.scan().unwrap();
        assert!(after.is_clean(), "a compacted log must scan clean");
        assert_eq!(after.pushes.len(), 1);
        assert_eq!(
            after.pushes[0].num_rows(),
            before_rows,
            "the merged batch must carry every row the frames did"
        );

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

    /// The reason compaction exists, as its own claim so that a mutation which
    /// stops it saving space fails *here* rather than tripping a row-count
    /// assertion first. Alignment padding paid 200 times collapses to once.
    #[test]
    fn compaction_actually_shrinks_the_log() {
        let dir = tmpdir("compact_size");
        let log = PushLog::new(dir.join("refs.log"), schema());
        for i in 0..200u64 {
            log.append(&batch(&["a", "b", "c", "d"], i)).unwrap();
        }
        let report = log.compact().unwrap();
        assert!(
            report.bytes_after * 2 < report.bytes_before,
            "compaction saved almost nothing: {} -> {} bytes. It costs a rewrite and the \
             frame boundaries; if it does not pay for them it should not run.",
            report.bytes_before,
            report.bytes_after
        );
        std::fs::remove_dir_all(&dir).ok();
    }

    /// **The crash bar.** A compaction interrupted at ANY byte must leave the
    /// OLD log serving, whole.
    ///
    /// Mirrors `gunnar-store`'s `an_interrupted_compaction_leaves_the_old_log_
    /// serving`, and swept across every byte rather than stopped at one point,
    /// for the same reason the torn-push sweep is: a recovery that holds at one
    /// offset and not another is not a recovery. A compaction that can lose a
    /// ref is worse than the padding it saves.
    ///
    /// Seen red by making `compact_with` write the merged frame straight over
    /// the live log instead of staging and renaming — the sweep then finds cuts
    /// where the log holds a partial frame and the pushes are gone.
    #[test]
    fn an_interrupted_compaction_leaves_the_old_log_serving_at_every_cut() {
        let dir = tmpdir("compact_crash");
        let path = dir.join("refs.log");
        let log = PushLog::new(&path, schema());
        for i in 0..40u64 {
            log.append(&batch(&["x", "y"], i)).unwrap();
        }
        let intact = std::fs::read(&path).unwrap();
        let expected_rows: usize = log.scan().unwrap().pushes.iter().map(|b| b.num_rows()).sum();

        // Stage a compaction and stop before the swap — what a crash leaves.
        let report = log.compact_with(Finish::StopBeforeSwap).unwrap();
        let staged = compacting_path(&path);
        assert!(staged.exists(), "the staged replacement must exist to cut into");
        let staged_bytes = std::fs::read(&staged).unwrap();
        assert!(staged_bytes.len() > 32);

        for cut in 0..staged_bytes.len() {
            // A crash part-way through writing the replacement.
            std::fs::write(&staged, &staged_bytes[..cut]).unwrap();

            // The live log is untouched and still serves every push.
            assert_eq!(
                std::fs::read(&path).unwrap(),
                intact,
                "cut at {cut}: the LIVE log was modified by a compaction that never \
                 completed — it must never be written in place"
            );
            let scan = log.scan().unwrap();
            assert!(scan.is_clean(), "cut at {cut}: the live log stopped scanning clean");
            assert_eq!(
                scan.pushes.len(),
                40,
                "cut at {cut}: the live log lost pushes to an interrupted compaction"
            );
            let rows: usize = scan.pushes.iter().map(|b| b.num_rows()).sum();
            assert_eq!(rows, expected_rows, "cut at {cut}: rows went missing");
        }

        // And finishing the compaction for real still works afterwards.
        std::fs::remove_file(&staged).ok();
        let done = log.compact().unwrap();
        assert_eq!(done.frames_after, 1);
        assert_eq!(done.rows, expected_rows);
        assert_eq!(report.rows, expected_rows);

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

    /// A log written before compaction existed still reads, and a compacted log
    /// reads on the same scanner. No migration, no version bump.
    #[test]
    fn compacted_and_uncompacted_logs_are_the_same_format() {
        let dir = tmpdir("compat");
        let path = dir.join("refs.log");
        let log = PushLog::new(&path, schema());
        for i in 0..8u64 {
            log.append(&batch(&["a"], i)).unwrap();
        }
        let uncompacted = log.scan().unwrap();
        log.compact().unwrap();
        let compacted = log.scan().unwrap();

        // Same rows, same order, read by the same scanner with no flag.
        let flat = |s: &PushLogScan| -> Vec<u64> {
            let mut v = Vec::new();
            for b in &s.pushes {
                let c = b.column(1).as_any().downcast_ref::<UInt64Array>().unwrap();
                v.extend((0..c.len()).map(|i| c.value(i)));
            }
            v
        };
        assert_eq!(flat(&uncompacted), flat(&compacted), "compaction changed the row sequence");

        // Appending after a compaction still works and stays readable.
        log.append(&batch(&["z"], 99)).unwrap();
        let after = log.scan().unwrap();
        assert!(after.is_clean());
        assert_eq!(after.pushes.len(), 2, "a compacted log must still accept appends");
        assert_eq!(*flat(&after).last().unwrap(), 99);

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

    /// A batch whose schema is not the log's is refused at append. Writing it
    /// would produce a frame the scanner parses happily but whose columns no
    /// reader of this log expects.
    #[test]
    fn a_foreign_schema_is_refused_at_append() {
        let dir = tmpdir("schema");
        let log = PushLog::new(dir.join("refs.log"), schema());
        let other = Arc::new(Schema::new(vec![Field::new("x", DataType::Utf8, false)]));
        let foreign =
            RecordBatch::try_new(other, vec![Arc::new(StringArray::from(vec!["v"]))]).unwrap();
        let err = log.append(&foreign).unwrap_err().to_string();
        assert!(err.contains("schema"), "expected a schema error, got: {err}");
        std::fs::remove_dir_all(&dir).ok();
    }
}