plugins/native/znippy-plugin-git/src/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();
}
}