server/lib/src/audit/arrow_ipc.rs
//! **The Arrow IPC audit index** — the searchable, columnar copy of the one audit
//! log (lane HOL-AUDIT, 2026-10-30; the Arrow backend the lean cut `9521bf06`
//! deleted, brought back as an INDEX).
//!
//! # The chain is the truth; this is derived
//!
//! [`FileAuditLog`](super::FileAuditLog) appends hash-chained JSON lines to
//! `<dir>/audit.jsonl` ([`super::chain`]). That file is the record, and nothing
//! here ever writes it. The index lives beside it, under `<dir>/index/`:
//!
//! | file | what |
//! |---|---|
//! | `MANIFEST.json` | the layout ([`LAYOUT`]), how far into the chain the index reaches (`offset`, `next_seq`, `head`) and its segments |
//! | `seg-<first seq>.arrow` | Arrow IPC **file** segments, one row per event, in chain order ([`index_schema`]) |
//! | `.lock` | the lock every refresh and read takes (`flock`, so other processes too) |
//!
//! Every row keeps its `seq` and chain `hash`, so a reader can check any answer
//! against a verified chain (`holger audit verify`, `AuditService.VerifyAuditChain`).
//! Checkpoint lines are links of the chain but not events: they move `next_seq`
//! and `head` and add no row.
//!
//! # Rebuilt from the chain, never the other way
//!
//! [`AuditIndex::table`] first catches the index up with the chain: it reads the
//! chain from the byte `offset` the manifest names (complete lines only, so a
//! line another process is writing is picked up next time) and writes the new rows
//! as a new segment. The index is REBUILT from the whole chain when it is missing,
//! unreadable, of another layout, or no longer the prefix of the chain it says it
//! is (the file is shorter than `offset`, the next line does not link to `head`,
//! or the tail moved without the file growing — a rewritten log). Deleting
//! `<dir>/index/` is always safe: the next read rebuilds it.
//!
//! Small segments are merged into their predecessor while the predecessor holds
//! fewer than four times their rows, so a long-running server keeps a logarithmic
//! number of segments, however often it is read.
use std::fs::{self, File, OpenOptions};
use std::io::{BufRead, BufReader, Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use arrow_array::{
Array, ArrayRef, RecordBatch, StringArray, TimestampNanosecondArray, UInt16Array, UInt64Array,
};
use arrow_schema::{DataType, Field, Schema, SchemaRef, TimeUnit};
use super::{chain, AuditError, AuditEvent, StoredRecord, AUDIT_FILE};
/// The index layout this code reads and writes. A manifest naming another layout
/// is not guessed at: the index is rebuilt.
pub const LAYOUT: &str = "holger.audit.index/v1";
/// The index directory under the audit directory.
pub const INDEX_DIR: &str = "index";
/// The manifest under the index directory.
pub const MANIFEST: &str = "MANIFEST.json";
/// Rows per Arrow record batch when the index is (re)built.
const BATCH_ROWS: usize = 65_536;
/// **The index schema.** Column order is the contract an outside Arrow reader
/// relies on; [`schema_string`] states it.
pub fn index_schema() -> SchemaRef {
Arc::new(Schema::new(vec![
Field::new("seq", DataType::UInt64, false),
Field::new("hash", DataType::Utf8, false),
Field::new("ts", DataType::Timestamp(TimeUnit::Nanosecond, None), false),
Field::new("ident", DataType::Utf8, false),
Field::new("action", DataType::Utf8, false),
Field::new("repo", DataType::Utf8, false),
Field::new("artifact", DataType::Utf8, false),
Field::new("source_ip", DataType::Utf8, false),
Field::new("status", DataType::UInt16, false),
Field::new("bytes", DataType::UInt64, false),
Field::new("detail", DataType::Utf8, false),
Field::new("sha256", DataType::Utf8, false),
Field::new("trail", DataType::Utf8, false),
]))
}
/// The schema in one line, `name:type` per column, for a manifest or an export
/// reply, so a registration and the data cannot disagree about column order.
pub fn schema_string() -> String {
index_schema()
.fields()
.iter()
.map(|f| format!("{}:{}", f.name(), f.data_type()))
.collect::<Vec<_>>()
.join(",")
}
fn arrow_err(e: impl std::fmt::Display) -> AuditError {
AuditError::Arrow(e.to_string())
}
/// One row of the index, borrowed from its batch.
#[derive(Debug, Clone, Copy)]
pub struct Row<'a> {
pub seq: u64,
pub hash: &'a str,
pub ts_nanos: i64,
pub ident: &'a str,
pub action: &'a str,
pub repo: &'a str,
pub artifact: &'a str,
pub source_ip: &'a str,
pub status: u16,
pub bytes: u64,
pub detail: &'a str,
pub sha256: &'a str,
pub trail: &'a str,
}
impl Row<'_> {
/// The row as the log holds it.
pub fn to_record(&self) -> StoredRecord {
StoredRecord {
seq: self.seq,
hash: self.hash.to_string(),
event: AuditEvent {
ts_nanos: self.ts_nanos,
ident: self.ident.to_string(),
action: super::AuditAction::from_wire(self.action),
repo: self.repo.to_string(),
artifact: self.artifact.to_string(),
source_ip: self.source_ip.to_string(),
status: self.status,
bytes: self.bytes,
detail: self.detail.to_string(),
sha256: self.sha256.to_string(),
trail: super::TrailKind::from_wire(self.trail),
},
}
}
}
/// **The records as Arrow batches**: what every query, aggregate and export reads.
#[derive(Debug, Clone, Default)]
pub struct AuditTable {
pub batches: Vec<RecordBatch>,
}
struct Cols<'a> {
seq: &'a UInt64Array,
hash: &'a StringArray,
ts: &'a TimestampNanosecondArray,
ident: &'a StringArray,
action: &'a StringArray,
repo: &'a StringArray,
artifact: &'a StringArray,
ip: &'a StringArray,
status: &'a UInt16Array,
bytes: &'a UInt64Array,
detail: &'a StringArray,
sha256: &'a StringArray,
trail: &'a StringArray,
}
fn col<'a, T: 'static>(b: &'a RecordBatch, name: &str) -> Result<&'a T, AuditError> {
b.column_by_name(name)
.and_then(|c| c.as_any().downcast_ref::<T>())
.ok_or_else(|| {
arrow_err(format!(
"the index has no column `{name}` of the expected type"
))
})
}
impl<'a> Cols<'a> {
fn of(b: &'a RecordBatch) -> Result<Self, AuditError> {
Ok(Self {
seq: col(b, "seq")?,
hash: col(b, "hash")?,
ts: col(b, "ts")?,
ident: col(b, "ident")?,
action: col(b, "action")?,
repo: col(b, "repo")?,
artifact: col(b, "artifact")?,
ip: col(b, "source_ip")?,
status: col(b, "status")?,
bytes: col(b, "bytes")?,
detail: col(b, "detail")?,
sha256: col(b, "sha256")?,
trail: col(b, "trail")?,
})
}
fn row(&self, i: usize) -> Row<'a> {
Row {
seq: self.seq.value(i),
hash: self.hash.value(i),
ts_nanos: self.ts.value(i),
ident: self.ident.value(i),
action: self.action.value(i),
repo: self.repo.value(i),
artifact: self.artifact.value(i),
source_ip: self.ip.value(i),
status: self.status.value(i),
bytes: self.bytes.value(i),
detail: self.detail.value(i),
sha256: self.sha256.value(i),
trail: self.trail.value(i),
}
}
}
impl AuditTable {
/// A table of `records` (the in-memory sink's snapshot; tests).
pub fn of_records(records: &[StoredRecord]) -> Result<Self, AuditError> {
let mut batches = Vec::new();
for chunk in records.chunks(BATCH_ROWS) {
batches.push(batch_of(chunk)?);
}
Ok(Self { batches })
}
/// How many rows the table holds.
pub fn rows(&self) -> u64 {
self.batches.iter().map(|b| b.num_rows() as u64).sum()
}
/// Visit every row in chain order.
pub fn for_each_row(&self, mut f: impl FnMut(Row<'_>)) -> Result<(), AuditError> {
for b in &self.batches {
let c = Cols::of(b)?;
for i in 0..b.num_rows() {
f(c.row(i));
}
}
Ok(())
}
/// The rows `keep` selects, as a new table (an export).
pub fn select(
&self,
mut keep: impl FnMut(&Row<'_>) -> bool,
) -> Result<Vec<StoredRecord>, AuditError> {
let mut out = Vec::new();
self.for_each_row(|r| {
if keep(&r) {
out.push(r.to_record());
}
})?;
Ok(out)
}
}
/// One Arrow batch of `records`, in [`index_schema`].
pub fn batch_of(records: &[StoredRecord]) -> Result<RecordBatch, AuditError> {
let s = |f: &dyn Fn(&StoredRecord) -> &str| -> ArrayRef {
Arc::new(StringArray::from_iter_values(records.iter().map(f)))
};
let cols: Vec<ArrayRef> = vec![
Arc::new(UInt64Array::from_iter_values(records.iter().map(|r| r.seq))),
s(&|r| r.hash.as_str()),
Arc::new(TimestampNanosecondArray::from_iter_values(
records.iter().map(|r| r.event.ts_nanos),
)),
s(&|r| r.event.ident.as_str()),
s(&|r| r.event.action.as_str()),
s(&|r| r.event.repo.as_str()),
s(&|r| r.event.artifact.as_str()),
s(&|r| r.event.source_ip.as_str()),
Arc::new(UInt16Array::from_iter_values(
records.iter().map(|r| r.event.status),
)),
Arc::new(UInt64Array::from_iter_values(
records.iter().map(|r| r.event.bytes),
)),
s(&|r| r.event.detail.as_str()),
s(&|r| r.event.sha256.as_str()),
s(&|r| r.event.trail.map(|t| t.as_str()).unwrap_or("")),
];
RecordBatch::try_new(index_schema(), cols).map_err(arrow_err)
}
/// Rows of a parsed chain line, as the index stores them. The action is kept as
/// the line wrote it ([`AuditEvent::from_json_value`] maps an unknown one to
/// `other`, which is what a reader of the row sees too).
fn record_of_line(v: &serde_json::Value, seq: u64) -> StoredRecord {
StoredRecord {
seq,
hash: v
.get("hash")
.and_then(|x| x.as_str())
.unwrap_or_default()
.to_string(),
event: AuditEvent::from_json_value(v),
}
}
/// Write `batches` as one Arrow IPC file at `path` (temp + rename, synced).
fn write_segment(path: &Path, batches: &[RecordBatch]) -> Result<(), AuditError> {
let tmp = path.with_extension("arrow.tmp");
{
let f = File::create(&tmp)?;
let mut w =
arrow_ipc::writer::FileWriter::try_new(f, &index_schema()).map_err(arrow_err)?;
for b in batches {
w.write(b).map_err(arrow_err)?;
}
w.finish().map_err(arrow_err)?;
let f = w.into_inner().map_err(arrow_err)?;
f.sync_all()?;
}
fs::rename(&tmp, path)?;
Ok(())
}
/// Every batch of one segment file.
fn read_segment(path: &Path) -> Result<Vec<RecordBatch>, AuditError> {
let f = File::open(path)?;
let r = arrow_ipc::reader::FileReader::try_new(f, None).map_err(arrow_err)?;
let mut out = Vec::new();
for b in r {
out.push(b.map_err(arrow_err)?);
}
Ok(out)
}
/// One segment as the manifest names it.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct Segment {
/// File name, relative to the index directory.
pub file: String,
pub first_seq: u64,
pub last_seq: u64,
pub rows: u64,
pub first_ts: i64,
pub last_ts: i64,
}
/// **The manifest**: how far into the chain the index reaches, and its segments.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct Manifest {
pub layout: String,
/// Bytes of `audit.jsonl` the index covers (always the end of a whole line).
pub offset: u64,
/// The `seq` of the next chain line.
pub next_seq: u64,
/// The hash of the last chain line covered ([`chain::GENESIS`] for none).
pub head: String,
/// Event rows in all segments.
pub rows: u64,
pub segments: Vec<Segment>,
}
impl Manifest {
fn empty() -> Self {
Self {
layout: LAYOUT.to_string(),
offset: 0,
next_seq: 0,
head: chain::GENESIS.to_string(),
rows: 0,
segments: Vec::new(),
}
}
}
/// What one refresh did.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Refresh {
/// The index was rebuilt from the whole chain, and why.
pub rebuilt: Option<String>,
/// Event rows added by this refresh.
pub added: u64,
/// Event rows the index holds now.
pub rows: u64,
/// The chain position the index covers now.
pub next_seq: u64,
}
/// **The index of one audit directory.** Cheap to make; every call takes the
/// index lock for its whole duration.
#[derive(Debug, Clone)]
pub struct AuditIndex {
audit_dir: PathBuf,
}
/// An exclusive `flock` on a file, released on drop.
struct Flock(File);
impl Flock {
fn exclusive(path: &Path) -> Result<Self, AuditError> {
let f = OpenOptions::new()
.create(true)
.truncate(false)
.read(true)
.write(true)
.open(path)?;
rustix::fs::flock(&f, rustix::fs::FlockOperation::LockExclusive)
.map_err(|e| AuditError::Io(e.into()))?;
Ok(Self(f))
}
}
impl Drop for Flock {
fn drop(&mut self) {
let _ = rustix::fs::flock(&self.0, rustix::fs::FlockOperation::Unlock);
}
}
/// The chain lines read past an offset: the rows, and where the reading stopped.
struct Tail {
records: Vec<StoredRecord>,
end_offset: u64,
next_seq: u64,
head: String,
}
impl AuditIndex {
/// The index of the audit directory `audit_dir`.
pub fn open(audit_dir: impl AsRef<Path>) -> Self {
Self {
audit_dir: audit_dir.as_ref().to_path_buf(),
}
}
/// `<audit dir>/index`.
pub fn dir(&self) -> PathBuf {
self.audit_dir.join(INDEX_DIR)
}
fn chain_path(&self) -> PathBuf {
self.audit_dir.join(AUDIT_FILE)
}
fn take_lock(&self) -> Result<Flock, AuditError> {
fs::create_dir_all(self.dir())?;
Flock::exclusive(&self.dir().join(".lock"))
}
/// The manifest as it stands, if there is a readable one of this layout.
pub fn manifest(&self) -> Option<Manifest> {
let bytes = fs::read(self.dir().join(MANIFEST)).ok()?;
let m: Manifest = serde_json::from_slice(&bytes).ok()?;
(m.layout == LAYOUT).then_some(m)
}
fn write_manifest(&self, m: &Manifest) -> Result<(), AuditError> {
let path = self.dir().join(MANIFEST);
let tmp = self.dir().join("MANIFEST.json.tmp");
let mut f = File::create(&tmp)?;
f.write_all(&serde_json::to_vec_pretty(m).map_err(|e| AuditError::Other(e.to_string()))?)?;
f.sync_all()?;
fs::rename(&tmp, &path)?;
Ok(())
}
/// Read whole chain lines from byte `from` (the line at `from` must link to
/// `head` at `seq`, else `Err(None)`-style: the caller rebuilds).
fn read_from(
&self,
from: u64,
seq: u64,
head: &str,
) -> Result<Result<Tail, String>, AuditError> {
let mut f = match File::open(self.chain_path()) {
Ok(f) => f,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
return Ok(if from == 0 {
Ok(Tail {
records: Vec::new(),
end_offset: 0,
next_seq: 0,
head: chain::GENESIS.into(),
})
} else {
Err("the chain file is gone".into())
});
}
Err(e) => return Err(e.into()),
};
let len = f.metadata()?.len();
if len < from {
return Ok(Err(format!(
"the chain file is {len} bytes, shorter than the {from} the index covers"
)));
}
f.seek(SeekFrom::Start(from))?;
let mut r = BufReader::new(f.take(len - from));
let (mut offset, mut next_seq, mut head) = (from, seq, head.to_string());
let mut records = Vec::new();
let mut line = String::new();
loop {
line.clear();
let n = r.read_line(&mut line)?;
if n == 0 || !line.ends_with('\n') {
break; // EOF, or a line still being written: picked up next time.
}
offset += n as u64;
if line.trim().is_empty() {
continue;
}
let v: serde_json::Value = match serde_json::from_str(&line) {
Ok(v) => v,
Err(e) => {
return Ok(Err(format!(
"the chain line at byte {} does not parse ({e})",
offset - n as u64
)))
}
};
let line_seq = v.get("seq").and_then(|s| s.as_u64());
if let Some(prev) = v.get("prev").and_then(|p| p.as_str()) {
if prev != head || line_seq != Some(next_seq) {
return Ok(Err(format!(
"the chain line at byte {} (seq {}) does not continue the index (seq {next_seq}, head {head})",
offset - n as u64,
line_seq.map(|s| s.to_string()).unwrap_or_else(|| "none".into())
)));
}
}
let this_seq = line_seq.unwrap_or(next_seq);
head = v
.get("hash")
.and_then(|h| h.as_str())
.map(str::to_string)
.unwrap_or_else(|| nornir_hash::sha256_hex(line.trim_end().as_bytes()));
next_seq = this_seq + 1;
if !chain::is_checkpoint(&v) {
records.push(record_of_line(&v, this_seq));
}
}
Ok(Ok(Tail {
records,
end_offset: offset,
next_seq,
head,
}))
}
fn segment_name(first_seq: u64) -> String {
format!("seg-{first_seq:020}.arrow")
}
fn segment_of(&self, records: &[StoredRecord]) -> Result<Option<Segment>, AuditError> {
let (Some(first), Some(last)) = (records.first(), records.last()) else {
return Ok(None);
};
let mut batches = Vec::new();
for chunk in records.chunks(BATCH_ROWS) {
batches.push(batch_of(chunk)?);
}
let file = Self::segment_name(first.seq);
write_segment(&self.dir().join(&file), &batches)?;
Ok(Some(Segment {
file,
first_seq: first.seq,
last_seq: last.seq,
rows: records.len() as u64,
first_ts: records.iter().map(|r| r.event.ts_nanos).min().unwrap_or(0),
last_ts: records.iter().map(|r| r.event.ts_nanos).max().unwrap_or(0),
}))
}
/// Wipe and rebuild from the whole chain. Lock held by the caller.
fn rebuild_locked(&self) -> Result<Manifest, AuditError> {
for e in fs::read_dir(self.dir())?.flatten() {
let name = e.file_name().to_string_lossy().into_owned();
if name.starts_with("seg-") || name == MANIFEST {
let _ = fs::remove_file(e.path());
}
}
let tail = match self.read_from(0, 0, chain::GENESIS)? {
Ok(t) => t,
// A chain that does not link from the start (a pre-chain line, a
// break) is indexed as it reads: rows are taken line by line.
Err(_) => self.read_unlinked()?,
};
let mut m = Manifest::empty();
if let Some(seg) = self.segment_of(&tail.records)? {
m.segments.push(seg);
}
m.rows = tail.records.len() as u64;
m.offset = tail.end_offset;
m.next_seq = tail.next_seq;
m.head = tail.head;
self.write_manifest(&m)?;
Ok(m)
}
/// Every whole line of a chain that does not link (it is broken, or predates
/// the chain): indexed as written, so a query still answers what the file
/// says. `holger audit verify` names the break.
fn read_unlinked(&self) -> Result<Tail, AuditError> {
let f = File::open(self.chain_path())?;
let len = f.metadata()?.len();
let mut r = BufReader::new(f.take(len));
let (mut offset, mut next_seq, mut head) = (0u64, 0u64, chain::GENESIS.to_string());
let mut records = Vec::new();
let mut line = String::new();
loop {
line.clear();
let n = r.read_line(&mut line)?;
if n == 0 || !line.ends_with('\n') {
break;
}
offset += n as u64;
let Ok(v) = serde_json::from_str::<serde_json::Value>(&line) else {
continue;
};
let this_seq = v.get("seq").and_then(|s| s.as_u64()).unwrap_or(next_seq);
head = v
.get("hash")
.and_then(|h| h.as_str())
.map(str::to_string)
.unwrap_or_else(|| nornir_hash::sha256_hex(line.trim_end().as_bytes()));
next_seq = this_seq + 1;
if !chain::is_checkpoint(&v) {
records.push(record_of_line(&v, this_seq));
}
}
Ok(Tail {
records,
end_offset: offset,
next_seq,
head,
})
}
/// Catch up (or rebuild). Lock held by the caller.
fn refresh_locked(&self) -> Result<(Manifest, Refresh), AuditError> {
// An empty chain (the zero-byte file a start opens) has nothing to index, and a
// fresh box must stay fresh (`twin::held_state`): no manifest, no segment.
if fs::metadata(self.chain_path())
.map(|m| m.len())
.unwrap_or(0)
== 0
{
if self.dir().join(MANIFEST).exists() {
self.rebuild_locked()?;
let _ = fs::remove_file(self.dir().join(MANIFEST));
}
return Ok((Manifest::empty(), Refresh::default()));
}
let Some(mut m) = self.manifest() else {
let m = self.rebuild_locked()?;
let r = Refresh {
rebuilt: Some("there was no readable index".into()),
added: m.rows,
rows: m.rows,
next_seq: m.next_seq,
};
return Ok((m, r));
};
if let Some(missing) = m
.segments
.iter()
.find(|s| !self.dir().join(&s.file).is_file())
{
let why = format!("segment {} is missing", missing.file);
let m = self.rebuild_locked()?;
return Ok((
m.clone(),
Refresh {
rebuilt: Some(why),
added: m.rows,
rows: m.rows,
next_seq: m.next_seq,
},
));
}
let len = fs::metadata(self.chain_path())
.map(|md| md.len())
.unwrap_or(0);
if len == m.offset {
// Nothing new. The tail must still be the one the index ends at: a log
// rewritten to the same length is a different log.
let (seq, head) = chain::read_tail(&self.chain_path())?;
if seq == m.next_seq && head == m.head {
let r = Refresh {
rebuilt: None,
added: 0,
rows: m.rows,
next_seq: m.next_seq,
};
return Ok((m, r));
}
let m = self.rebuild_locked()?;
return Ok((
m.clone(),
Refresh {
rebuilt: Some("the chain's tail moved without the file growing".into()),
added: m.rows,
rows: m.rows,
next_seq: m.next_seq,
},
));
}
let tail = match self.read_from(m.offset, m.next_seq, &m.head)? {
Ok(t) => t,
Err(why) => {
let m = self.rebuild_locked()?;
return Ok((
m.clone(),
Refresh {
rebuilt: Some(why),
added: m.rows,
rows: m.rows,
next_seq: m.next_seq,
},
));
}
};
let added = tail.records.len() as u64;
if let Some(seg) = self.segment_of(&tail.records)? {
m.segments.push(seg);
}
m.rows += added;
m.offset = tail.end_offset;
m.next_seq = tail.next_seq;
m.head = tail.head;
self.merge_small(&mut m)?;
self.write_manifest(&m)?;
let r = Refresh {
rebuilt: None,
added,
rows: m.rows,
next_seq: m.next_seq,
};
Ok((m, r))
}
/// Merge the newest segment into its predecessor while the predecessor holds
/// fewer than four times its rows. The manifest is written by the caller; the
/// old files go after it is.
fn merge_small(&self, m: &mut Manifest) -> Result<(), AuditError> {
let mut dropped = Vec::new();
while m.segments.len() >= 2 {
let n = m.segments.len();
let (a, b) = (&m.segments[n - 2], &m.segments[n - 1]);
if a.rows >= b.rows.saturating_mul(4) {
break;
}
let mut batches = read_segment(&self.dir().join(&a.file))?;
batches.extend(read_segment(&self.dir().join(&b.file))?);
let merged = Segment {
file: format!("seg-{:020}-{:020}.arrow", a.first_seq, b.last_seq),
first_seq: a.first_seq,
last_seq: b.last_seq,
rows: a.rows + b.rows,
first_ts: a.first_ts.min(b.first_ts),
last_ts: a.last_ts.max(b.last_ts),
};
write_segment(&self.dir().join(&merged.file), &batches)?;
dropped.push(a.file.clone());
dropped.push(b.file.clone());
m.segments.truncate(n - 2);
m.segments.push(merged);
}
if !dropped.is_empty() {
self.write_manifest(m)?;
let live: std::collections::BTreeSet<&str> =
m.segments.iter().map(|s| s.file.as_str()).collect();
for f in dropped.iter().filter(|f| !live.contains(f.as_str())) {
let _ = fs::remove_file(self.dir().join(f));
}
}
Ok(())
}
/// **Catch the index up with the chain** (rebuilding it when it is not the
/// chain's prefix), and say what happened.
pub fn refresh(&self) -> Result<Refresh, AuditError> {
let _g = self.take_lock()?;
Ok(self.refresh_locked()?.1)
}
/// **Rebuild the index from the whole chain**, whatever it holds now.
pub fn rebuild(&self) -> Result<Refresh, AuditError> {
let _g = self.take_lock()?;
let m = self.rebuild_locked()?;
Ok(Refresh {
rebuilt: Some("asked for".into()),
added: m.rows,
rows: m.rows,
next_seq: m.next_seq,
})
}
/// **Every record, as Arrow batches**, the index caught up first.
pub fn table(&self) -> Result<AuditTable, AuditError> {
let _g = self.take_lock()?;
let (m, _) = self.refresh_locked()?;
let mut batches = Vec::new();
for s in &m.segments {
batches.extend(read_segment(&self.dir().join(&s.file))?);
}
Ok(AuditTable { batches })
}
}
/// **Write `records` as one Arrow IPC file** (an export): the bytes.
pub fn arrow_file_bytes(records: &[StoredRecord]) -> Result<Vec<u8>, AuditError> {
let mut buf = Vec::new();
{
let mut w =
arrow_ipc::writer::FileWriter::try_new(&mut buf, &index_schema()).map_err(arrow_err)?;
for chunk in records.chunks(BATCH_ROWS) {
w.write(&batch_of(chunk)?).map_err(arrow_err)?;
}
w.finish().map_err(arrow_err)?;
}
Ok(buf)
}
/// **Read an Arrow IPC file export back** (a reader's side; tests).
pub fn read_arrow_file_bytes(bytes: &[u8]) -> Result<Vec<StoredRecord>, AuditError> {
let r = arrow_ipc::reader::FileReader::try_new(std::io::Cursor::new(bytes.to_vec()), None)
.map_err(arrow_err)?;
let mut t = AuditTable::default();
for b in r {
t.batches.push(b.map_err(arrow_err)?);
}
t.select(|_| true)
}
#[cfg(test)]
mod tests {
use super::super::{AuditAction, AuditEvent, AuditLog, FileAuditLog};
use super::*;
fn tmp(tag: &str) -> PathBuf {
std::env::temp_dir().join(format!(
"holger-index-{tag}-{}-{}",
std::process::id(),
super::super::now_nanos()
))
}
fn ev(i: u64) -> AuditEvent {
AuditEvent::new(
format!("u{}", i % 3),
AuditAction::Download,
"r",
format!("a@{i}"),
format!("10.0.0.{}:9", i % 5),
200,
i,
)
}
fn seqs(t: &AuditTable) -> Vec<u64> {
let mut v = Vec::new();
t.for_each_row(|r| v.push(r.seq)).unwrap();
v
}
/// ★ The index is built from the chain, caught up as the chain grows, and
/// rebuilt from it when deleted: the same rows, the same chain places.
#[test]
fn the_index_is_rebuilt_from_the_chain() {
let dir = tmp("rebuild");
let log = FileAuditLog::new(&dir).unwrap().with_checkpoint_every(4);
for i in 0..10 {
log.record(ev(i)).unwrap();
}
let idx = AuditIndex::open(&dir);
let first = idx.table().unwrap();
assert_eq!(first.rows(), 10, "checkpoints are links, not rows");
let want: Vec<u64> = FileAuditLog::read_records(&dir)
.unwrap()
.iter()
.map(|r| r.seq)
.collect();
assert_eq!(seqs(&first), want);
assert!(idx.dir().join(MANIFEST).is_file());
for i in 10..13 {
log.record(ev(i)).unwrap();
}
let r = idx.refresh().unwrap();
assert_eq!((r.rebuilt.clone(), r.added, r.rows), (None, 3, 13), "{r:?}");
std::fs::remove_dir_all(idx.dir()).unwrap();
let again = idx.table().unwrap();
assert_eq!(again.rows(), 13);
let all: Vec<StoredRecord> = again.select(|_| true).unwrap();
assert_eq!(
all,
FileAuditLog::read_records(&dir).unwrap(),
"rebuilt row for row from the chain"
);
let _ = std::fs::remove_dir_all(&dir);
}
/// A chain rewritten under the index (same length or not) is not trusted as
/// its prefix: the index is rebuilt from what the chain says now.
#[test]
fn a_rewritten_chain_rebuilds_the_index() {
let dir = tmp("rewritten");
let log = FileAuditLog::new(&dir).unwrap().with_checkpoint_every(0);
for i in 0..4 {
log.record(ev(i)).unwrap();
}
let idx = AuditIndex::open(&dir);
assert_eq!(idx.table().unwrap().rows(), 4);
// A fresh log of two other records in its place.
std::fs::remove_file(dir.join(AUDIT_FILE)).unwrap();
let other = FileAuditLog::new(&dir).unwrap().with_checkpoint_every(0);
other
.record(AuditEvent::new(
"x",
AuditAction::Upload,
"q",
"z",
"",
200,
1,
))
.unwrap();
other
.record(AuditEvent::new(
"x",
AuditAction::Upload,
"q",
"z2",
"",
200,
1,
))
.unwrap();
let r = idx.refresh().unwrap();
assert!(r.rebuilt.is_some(), "{r:?}");
let t = idx.table().unwrap();
assert_eq!(
t.select(|_| true).unwrap(),
FileAuditLog::read_records(&dir).unwrap()
);
let _ = std::fs::remove_dir_all(&dir);
}
/// Many small refreshes keep few segments.
#[test]
fn small_segments_are_merged() {
let dir = tmp("merge");
let log = FileAuditLog::new(&dir).unwrap().with_checkpoint_every(0);
let idx = AuditIndex::open(&dir);
for i in 0..64 {
log.record(ev(i)).unwrap();
idx.refresh().unwrap();
}
let m = idx.manifest().unwrap();
assert_eq!(m.rows, 64);
assert!(m.segments.len() <= 8, "{} segments", m.segments.len());
let files = std::fs::read_dir(idx.dir())
.unwrap()
.flatten()
.filter(|e| e.file_name().to_string_lossy().ends_with(".arrow"))
.count();
assert_eq!(files, m.segments.len(), "merged-away files are removed");
assert_eq!(seqs(&idx.table().unwrap()), (0..64).collect::<Vec<_>>());
let _ = std::fs::remove_dir_all(&dir);
}
/// ★ An empty chain (the zero-byte file a start opens) leaves no index bytes:
/// a fresh box stays fresh (`twin::held_state`), and its reads answer nothing.
#[test]
fn an_empty_chain_writes_no_index() {
let dir = tmp("empty");
let _log = FileAuditLog::new(&dir).unwrap();
let idx = AuditIndex::open(&dir);
assert_eq!(idx.refresh().unwrap().rows, 0);
assert_eq!(idx.table().unwrap().rows(), 0);
let written: Vec<String> = std::fs::read_dir(idx.dir())
.map(|d| {
d.flatten()
.filter(|e| e.metadata().map(|m| m.len() > 0).unwrap_or(false))
.map(|e| e.file_name().to_string_lossy().into_owned())
.collect()
})
.unwrap_or_default();
assert!(
written.is_empty(),
"no index bytes for an empty chain: {written:?}"
);
let _ = std::fs::remove_dir_all(&dir);
}
}