krabka_audit/
spool.rs

1//! Durable-across-process-crash local spool for the AU-5 degraded path.
2//!
3//! The spool holds exactly the chained audit records that are not yet written
4//! to the topic, in order. By default, `append` calls `fsync` before it returns
5//! success. Callers that explicitly choose a wider sync cadence can batch that
6//! cost and use [`Spool::sync`] when they need a durable acknowledgement.
7//! `open` heals and durably truncates a torn tail frame from a crash during an
8//! append.
9//!
10//! File: a header (the magic `KAUD`, then the `i16`
11//! version), then frames. Frame: `[u32 len][record]`. Record: `[u8
12//! class_tag][u32 value_len][value][u32 header_count]([u32 klen][k][u32
13//! vlen][v])*`. This module uses synchronous `std::fs`, because the path is
14//! degraded and low-frequency. It treats a truncated or corrupt tail frame as
15//! end-of-data, but a missing or unknown file header as a hard error.
16//!
17//! The sidecars carry the same header: `audit.losses` then holds the `u64`
18//! generation and the `u64` count, `audit.replay-offset` the `u64` offset,
19//! and `audit.replay-poison` the `u64` offset and one frame. Every offset and
20//! byte count here is into the frames, after the header.
21
22use {
23    self::codec::{HEADER_LEN, decode_record, encode_frame, file_header, strip_file_header},
24    crate::{
25        event::AuditEventClass,
26        ids::{MaxSpoolBytes, RecordCount, SpoolBytes},
27        sink::{AuditError, AuditRecord},
28    },
29    krabka_units::{
30        fmt::Human as _,
31        prelude::{ByteSize, ByteSizeExt as _},
32    },
33    krabka_verified::{
34        audit::{AuditLosses, settle_loss_batch},
35        spool_append_decision,
36    },
37    std::{
38        fmt::Display,
39        fs::{self, File, OpenOptions},
40        io::{ErrorKind, Read, Seek, SeekFrom, Write},
41        num::NonZeroU64,
42        path::{Path, PathBuf},
43        sync::{Arc, Mutex, PoisonError},
44    },
45    tracing::field::Empty,
46};
47
48mod codec;
49mod resume;
50
51#[cfg(test)]
52pub(crate) use self::codec::HEADER_LEN as FILE_HEADER_LEN;
53
54#[cfg(test)]
55mod test_support;
56
57const SPOOL_FILE: &str = "audit.spool";
58const SPOOL_TMP: &str = "audit.spool.tmp";
59const LOSS_STATE_FILE: &str = "audit.losses";
60const LOSS_STATE_TMP: &str = "audit.losses.tmp";
61const LOSS_STATE_LEN: usize = 16;
62const REPLAY_OFFSET_FILE: &str = "audit.replay-offset";
63const REPLAY_OFFSET_TMP: &str = "audit.replay-offset.tmp";
64const REPLAY_POISON_FILE: &str = "audit.replay-poison";
65const REPLAY_POISON_TMP: &str = "audit.replay-poison.tmp";
66
67pub(crate) fn add_loss_state(generation: u64, count: u64, added: u64) -> (u64, u64) {
68    let generation = if count == 0 {
69        generation.saturating_add(1)
70    } else {
71        generation
72    };
73    (generation, count.saturating_add(added))
74}
75
76/// What a scan of the spool file found.
77struct ScannedSpool {
78    /// The records that decoded, in order.
79    records: Vec<AuditRecord>,
80    /// Encoded frame lengths corresponding one-for-one with `records`.
81    frame_lengths: Vec<u64>,
82    /// The byte length of the file after the last complete frame.
83    valid_bytes: SpoolBytes,
84}
85
86/// One record read from the durable replay cursor.
87///
88/// Keeping the on-disk frame length alongside the decoded record means a
89/// successful replay can advance the cursor without serializing the record a
90/// second time.
91pub(crate) struct ReplayRecord {
92    pub(crate) record: AuditRecord,
93    pub(crate) frame_len: u64,
94}
95
96/// Replay is deliberately record-bounded. A healthy sink can consume several
97/// batches in one pass, while a failing sink makes us retain at most this many
98/// decoded records between retries.
99pub(crate) const REPLAY_BATCH_RECORDS: usize = 64;
100
101#[derive(Debug, Clone, Copy, PartialEq, Eq)]
102pub(crate) enum ReplayRecovery {
103    ClearPoison,
104    RequireExplicitRecovery,
105}
106
107pub(crate) fn replay_recovery(
108    valid_frame: bool,
109    poison_offset: Option<u64>,
110    replay_offset: u64,
111) -> ReplayRecovery {
112    if valid_frame && poison_offset.is_some_and(|offset| offset < replay_offset) {
113        ReplayRecovery::ClearPoison
114    } else {
115        ReplayRecovery::RequireExplicitRecovery
116    }
117}
118
119/// Writer-persisted count of fail-open records awaiting a chain marker.
120#[derive(Debug)]
121pub(crate) struct PendingLosses {
122    state: Mutex<AuditLosses>,
123    path: Option<PathBuf>,
124}
125
126impl PendingLosses {
127    pub(crate) fn memory() -> Arc<Self> {
128        Arc::new(Self {
129            state: Mutex::new(AuditLosses::default()),
130            path: None,
131        })
132    }
133
134    fn open(dir: &Path) -> Result<Arc<Self>, AuditError> {
135        let path = dir.join(LOSS_STATE_FILE);
136        let state = if path.exists() {
137            let bytes = fs::read(&path).map_err(io)?;
138            let bytes = strip_file_header(LOSS_STATE_FILE, &bytes)?;
139            // `LOSS_STATE_LEN` bytes: the generation, then exactly eight count bytes.
140            let Some((generation, count)) =
141                bytes
142                    .split_first_chunk::<8>()
143                    .and_then(|(generation, count)| {
144                        Some((*generation, <[u8; 8]>::try_from(count).ok()?))
145                    })
146            else {
147                return Err(AuditError::Io(format!(
148                    "invalid audit loss state length {}",
149                    bytes.len()
150                )));
151            };
152            AuditLosses {
153                generation: u64::from_be_bytes(generation),
154                count: u64::from_be_bytes(count),
155            }
156        } else {
157            let state = AuditLosses::default();
158            persist_loss_state(&path, state)?;
159            state
160        };
161        Ok(Arc::new(Self {
162            state: Mutex::new(state),
163            path: Some(path),
164        }))
165    }
166
167    pub(crate) fn add(&self, count: u64) {
168        let mut state = self.state.lock().unwrap_or_else(PoisonError::into_inner);
169        (state.generation, state.count) = add_loss_state(state.generation, state.count, count);
170    }
171
172    #[cfg(test)]
173    pub(crate) fn count(&self) -> u64 {
174        self.state
175            .lock()
176            .unwrap_or_else(PoisonError::into_inner)
177            .count
178    }
179
180    #[cfg(test)]
181    fn state(&self) -> AuditLosses { *self.state.lock().unwrap_or_else(PoisonError::into_inner) }
182
183    pub(crate) fn snapshot(&self) -> Option<AuditLosses> {
184        let state = *self.state.lock().unwrap_or_else(PoisonError::into_inner);
185        (state.count > 0).then_some(state)
186    }
187
188    /// Settle `batch` once its marker is durable. Losses added since the
189    /// snapshot stay pending in a fresh generation (`settle_loss_batch`).
190    pub(crate) fn commit(&self, batch: AuditLosses) {
191        self.update_and_persist(|state| {
192            if state.generation != batch.generation {
193                return false;
194            }
195            *state = settle_loss_batch(*state, batch);
196            true
197        });
198    }
199
200    pub(crate) fn persist(&self) -> Result<(), AuditError> {
201        let state = *self.state.lock().unwrap_or_else(PoisonError::into_inner);
202        self.path
203            .as_ref()
204            .map_or(Ok(()), |path| persist_loss_state(path, state))
205    }
206
207    pub(crate) fn persist_with<T>(
208        &self,
209        write_marker: impl FnOnce(AuditLosses) -> Result<T, AuditError>,
210    ) -> Result<Option<T>, AuditError> {
211        let Some(batch) = self.snapshot() else {
212            return Ok(None);
213        };
214        self.persist()?;
215        let result = write_marker(batch)?;
216        self.commit(batch);
217        Ok(Some(result))
218    }
219
220    /// Settle the persisted pending losses against the durable marker of
221    /// their generation, if the spool holds one: a crash fell between the
222    /// marker's sync and the commit's sidecar write. The marker settles only
223    /// the count it reports (`settle_loss_batch`).
224    fn reconcile(&self, records: &[AuditRecord]) {
225        self.update_and_persist(|state| {
226            if state.count == 0 {
227                return false;
228            }
229            let Some(batch) = records
230                .iter()
231                .filter_map(loss_marker_batch)
232                .find(|batch| batch.generation == state.generation)
233            else {
234                return false;
235            };
236            *state = settle_loss_batch(*state, batch);
237            true
238        });
239    }
240
241    fn update_and_persist(&self, update: impl FnOnce(&mut AuditLosses) -> bool) {
242        let state = {
243            let mut state = self.state.lock().unwrap_or_else(PoisonError::into_inner);
244            if !update(&mut state) {
245                return;
246            }
247            *state
248        };
249        self.persist_or_warn(state);
250    }
251
252    fn persist_or_warn(&self, state: AuditLosses) {
253        if let Some(path) = &self.path
254            && let Err(error) = persist_loss_state(path, state)
255        {
256            tracing::error!(%error, "failed to persist pending audit loss count");
257        }
258    }
259}
260
261/// The losses a records-lost marker reports, or `None` for any other record.
262fn loss_marker_batch(record: &AuditRecord) -> Option<AuditLosses> {
263    if record.class != AuditEventClass::RecordsLost {
264        return None;
265    }
266    let value = serde_json::from_slice::<serde_json::Value>(&record.value).ok()?;
267    Some(AuditLosses {
268        generation: value.get("loss_generation")?.as_u64()?,
269        count: value.get("records_lost")?.as_u64()?,
270    })
271}
272
273fn io<E: Display>(e: E) -> AuditError { AuditError::Io(e.to_string()) }
274
275fn persist_loss_state(path: &Path, state: AuditLosses) -> Result<(), AuditError> {
276    let mut bytes = [0_u8; LOSS_STATE_LEN];
277    bytes[..8].copy_from_slice(&state.generation.to_be_bytes());
278    bytes[8..].copy_from_slice(&state.count.to_be_bytes());
279    persist_with_header(path, LOSS_STATE_TMP, &bytes)
280}
281
282fn read_or_create_u64(path: &Path, tmp_name: &str) -> Result<u64, AuditError> {
283    match fs::read(path) {
284        Ok(bytes) => <[u8; 8]>::try_from(strip_file_header(&file_name(path), &bytes)?)
285            .map(u64::from_be_bytes)
286            .map_err(|_| AuditError::Io(format!("invalid state length for {}", path.display()))),
287        Err(error) if error.kind() == ErrorKind::NotFound => {
288            persist_u64(path, tmp_name, 0)?;
289            Ok(0)
290        }
291        Err(error) => Err(io(error)),
292    }
293}
294
295fn persist_u64(path: &Path, tmp_name: &str, value: u64) -> Result<(), AuditError> {
296    persist_with_header(path, tmp_name, &value.to_be_bytes())
297}
298
299/// The file name of `path`, for an error that names the file.
300fn file_name(path: &Path) -> String {
301    path.file_name().map_or_else(
302        || path.display().to_string(),
303        |name| name.to_string_lossy().into_owned(),
304    )
305}
306
307/// Atomically replace `path` with the file header followed by `body`.
308fn persist_with_header(path: &Path, tmp_name: &str, body: &[u8]) -> Result<(), AuditError> {
309    let mut bytes = Vec::with_capacity(HEADER_LEN + body.len());
310    bytes.extend_from_slice(&file_header());
311    bytes.extend_from_slice(body);
312    persist_bytes(path, tmp_name, &bytes)
313}
314
315fn persist_bytes(path: &Path, tmp_name: &str, bytes: &[u8]) -> Result<(), AuditError> {
316    let tmp = path.with_file_name(tmp_name);
317    let mut file = OpenOptions::new()
318        .write(true)
319        .create(true)
320        .truncate(true)
321        .open(&tmp)
322        .map_err(io)?;
323    file.write_all(bytes).map_err(io)?;
324    file.sync_all().map_err(io)?;
325    fs::rename(tmp, path).map_err(io)?;
326    sync_parent(path)
327}
328
329/// Append-only durable spool file.
330///
331/// The byte cap and the running total are raw [`MaxSpoolBytes`] and
332/// [`SpoolBytes`] newtypes. The spool adds and compares both of them exactly,
333/// which an `f64`-backed [`ByteSize`] cannot promise. The quantity stays the
334/// boundary type: [`Spool::open`] takes one and [`Spool::size`] returns one.
335#[derive(Debug)]
336pub struct Spool {
337    path: PathBuf,
338    file: File,
339    max_bytes: MaxSpoolBytes,
340    bytes: SpoolBytes,
341    count: RecordCount,
342    recovered_torn_tail: bool,
343    sync_every: NonZeroU64,
344    unsynced: u64,
345    replay_offset: u64,
346    pending_losses: Arc<PendingLosses>,
347}
348
349impl Spool {
350    /// Open the spool and recover its existing contents.
351    ///
352    /// This function creates the directory and the file if they do not exist.
353    #[tracing::instrument(
354        level = "debug",
355        skip_all,
356        fields(dir = %dir.display(), max_size = %max_size.human(), count = Empty, bytes = Empty),
357        err
358    )]
359    /// # Errors
360    /// Returns an error if the spool directory cannot be created. Returns an
361    /// error if the spool file cannot be opened, read, or truncated.
362    pub fn open(dir: &Path, max_size: ByteSize) -> Result<Self, AuditError> {
363        Self::open_with_sync_every(dir, max_size, NonZeroU64::MIN)
364    }
365
366    /// Open the spool with an explicit append-to-`fsync` cadence.
367    ///
368    /// A cadence of `N` syncs after every `N` successful appends. Appends before
369    /// that boundary are visible after a process crash but are not promised to
370    /// survive power loss until [`Self::sync`] succeeds.
371    #[tracing::instrument(
372        level = "debug",
373        skip_all,
374        fields(dir = %dir.display(), max_size = %max_size.human(), sync_every = sync_every.get(), count = Empty, bytes = Empty),
375        err
376    )]
377    /// # Errors
378    /// Returns an error if the spool directory cannot be created. Returns an
379    /// error if the spool file cannot be opened, read, synced, or truncated.
380    pub fn open_with_sync_every(
381        dir: &Path,
382        max_size: ByteSize,
383        sync_every: NonZeroU64,
384    ) -> Result<Self, AuditError> {
385        fs::create_dir_all(dir).map_err(io)?;
386        let pending_losses = PendingLosses::open(dir)?;
387        let path = dir.join(SPOOL_FILE);
388        let replay_offset_path = dir.join(REPLAY_OFFSET_FILE);
389        let replay_offset = read_or_create_u64(&replay_offset_path, REPLAY_OFFSET_TMP)?;
390        if !path.exists() {
391            // Created whole, header included, by rename: a crash leaves either
392            // no spool or one with its header, never a headerless file.
393            persist_with_header(&path, SPOOL_TMP, &[])?;
394        }
395        let file = OpenOptions::new()
396            .read(true)
397            .write(true)
398            .open(&path)
399            .map_err(io)?;
400        let mut s = Self {
401            path,
402            file,
403            max_bytes: MaxSpoolBytes(max_size.bytes_u64()),
404            bytes: SpoolBytes(0),
405            count: RecordCount(0),
406            recovered_torn_tail: false,
407            sync_every,
408            unsynced: 0,
409            replay_offset,
410            pending_losses,
411        };
412        let ScannedSpool {
413            records,
414            frame_lengths,
415            valid_bytes,
416        } = s.scan()?;
417        let physical = s.file.metadata().map_err(io)?.len();
418        s.recovered_torn_tail = header_len_u64() + valid_bytes.0 < physical;
419        if s.recovered_torn_tail {
420            s.file
421                .set_len(header_len_u64() + valid_bytes.0)
422                .map_err(io)?;
423            s.file.sync_all().map_err(io)?;
424            tracing::warn!(
425                physical,
426                valid_bytes = valid_bytes.0,
427                "audit spool: truncated torn tail frame on open"
428            );
429        }
430        s.bytes = valid_bytes;
431        let unread = s.unread_record_count(&frame_lengths, valid_bytes)?;
432        s.count = RecordCount(u64::try_from(unread).unwrap_or(u64::MAX));
433        s.reconcile_replay_poison()?;
434        s.pending_losses.reconcile(&records);
435        if s.count.0 == 0 && s.replay_offset > 0 {
436            s.truncate()?;
437        }
438        let span = tracing::Span::current();
439        span.record("count", s.count.0);
440        span.record("bytes", s.bytes.0);
441        Ok(s)
442    }
443
444    #[must_use]
445    pub fn is_empty(&self) -> bool { self.count.0 == 0 }
446
447    #[must_use]
448    pub fn count(&self) -> RecordCount { self.count }
449
450    pub(crate) fn pending_losses(&self) -> Arc<PendingLosses> { Arc::clone(&self.pending_losses) }
451
452    /// How many bytes the spool currently holds.
453    #[must_use]
454    pub fn size(&self) -> ByteSize { ByteSize::from_bytes(self.bytes.0) }
455
456    /// Whether [`Spool::open`] found a torn tail frame and cut it off.
457    ///
458    /// A crash partway through an append leaves a frame whose length prefix
459    /// promises more bytes than follow it. `open` heals that, and this reports
460    /// whether it had to -- healing a spool that was intact would mean
461    /// discarding good bytes, so the two cases are worth telling apart.
462    #[must_use]
463    pub fn recovered_torn_tail(&self) -> bool { self.recovered_torn_tail }
464
465    /// Append a record to the spool.
466    ///
467    /// Returns `Ok(false)` if the record would make the spool exceed the
468    /// configured cap.
469    #[tracing::instrument(
470        level = "debug",
471        skip_all,
472        fields(class = ?record.class, value_bytes = record.value.len(), count = self.count.0),
473        err
474    )]
475    /// # Errors
476    /// Returns an error if the seek, write, rollback, or configured sync on the
477    /// spool file fails.
478    pub fn append(&mut self, record: &AuditRecord) -> Result<bool, AuditError> {
479        self.append_inner(record)
480    }
481
482    /// Persist a loss marker within the configured cap.
483    pub(crate) fn append_loss_marker(&mut self, record: &AuditRecord) -> Result<(), AuditError> {
484        if !self.append(record)? {
485            return Err(AuditError::Unavailable("audit spool is full".into()));
486        }
487        self.sync()
488    }
489
490    pub(crate) fn begin_replay(&self, record: &AuditRecord) -> Result<(), AuditError> {
491        let path = self.path.with_file_name(REPLAY_POISON_FILE);
492        let mut poison = self.replay_offset.to_be_bytes().to_vec();
493        poison.extend_from_slice(&encode_frame(record)?);
494        persist_with_header(&path, REPLAY_POISON_TMP, &poison)
495    }
496
497    pub(crate) fn commit_replay_frame(&mut self, frame_len: u64) -> Result<(), AuditError> {
498        let replay_offset = self
499            .replay_offset
500            .checked_add(frame_len)
501            .ok_or_else(|| AuditError::Io("audit replay offset overflow".into()))?;
502        persist_u64(
503            &self.path.with_file_name(REPLAY_OFFSET_FILE),
504            REPLAY_OFFSET_TMP,
505            replay_offset,
506        )?;
507        self.replay_offset = replay_offset;
508        self.count.0 = self.count.0.saturating_sub(1);
509        self.clear_replay_poison()
510    }
511
512    #[cfg(test)]
513    fn commit_replay(&mut self, record: &AuditRecord) -> Result<(), AuditError> {
514        self.commit_replay_frame(u64::try_from(encode_frame(record)?.len()).unwrap_or(u64::MAX))
515    }
516
517    pub(crate) fn abort_replay(&self) -> Result<(), AuditError> { self.clear_replay_poison() }
518
519    fn clear_replay_poison(&self) -> Result<(), AuditError> {
520        let path = self.path.with_file_name(REPLAY_POISON_FILE);
521        match fs::remove_file(&path) {
522            Ok(()) => sync_parent(&path),
523            Err(error) if error.kind() == ErrorKind::NotFound => Ok(()),
524            Err(error) => Err(io(error)),
525        }
526    }
527
528    fn reconcile_replay_poison(&self) -> Result<(), AuditError> {
529        let path = self.path.with_file_name(REPLAY_POISON_FILE);
530        let bytes = match fs::read(&path) {
531            Ok(bytes) => bytes,
532            Err(error) if error.kind() == ErrorKind::NotFound => return Ok(()),
533            Err(error) => return Err(io(error)),
534        };
535        let bytes = strip_file_header(REPLAY_POISON_FILE, &bytes)?;
536        let poison_offset = bytes
537            .get(..8)
538            .and_then(|prefix| <[u8; 8]>::try_from(prefix).ok())
539            .map(u64::from_be_bytes);
540        let valid_frame = bytes
541            .get(8..12)
542            .and_then(|prefix| <[u8; 4]>::try_from(prefix).ok())
543            .map(u32::from_be_bytes)
544            .and_then(|len| usize::try_from(len).ok())
545            .is_some_and(|len| len.checked_add(12) == Some(bytes.len()));
546        match replay_recovery(valid_frame, poison_offset, self.replay_offset) {
547            ReplayRecovery::ClearPoison => self.clear_replay_poison(),
548            ReplayRecovery::RequireExplicitRecovery => {
549                Err(AuditError::Poisoned(path.display().to_string()))
550            }
551        }
552    }
553
554    fn append_inner(&mut self, record: &AuditRecord) -> Result<bool, AuditError> {
555        let frame = encode_frame(record)?;
556        let frame_len = u64::try_from(frame.len()).unwrap_or(u64::MAX);
557        let decision = spool_append_decision(
558            self.bytes.0,
559            frame_len,
560            self.max_bytes.0,
561            self.unsynced,
562            self.sync_every.get(),
563        );
564        if !decision.accepted {
565            return Ok(false);
566        }
567        let old_len = self.file.metadata().map_err(io)?.len();
568        self.file.seek(SeekFrom::End(0)).map_err(io)?;
569        if let Err(error) = self.file.write_all(&frame) {
570            return Err(self.rollback_append(old_len, error));
571        }
572        if decision.sync
573            && let Err(error) = self.file.sync_all()
574        {
575            return Err(self.rollback_append(old_len, error));
576        }
577        self.unsynced = decision.next_unsynced;
578        self.bytes = SpoolBytes(decision.new_bytes);
579        self.count.0 = self.count.0.saturating_add(1);
580        Ok(true)
581    }
582
583    /// Force all appended records to stable storage.
584    ///
585    /// Callers using a sync cadence greater than one use this before reporting
586    /// a durable acknowledgement.
587    /// # Errors
588    /// Returns an error if the spool file cannot be synced.
589    pub fn sync(&mut self) -> Result<(), AuditError> {
590        self.file.sync_all().map_err(io)?;
591        self.unsynced = 0;
592        Ok(())
593    }
594
595    fn rollback_append<E: Display>(&mut self, old_len: u64, error: E) -> AuditError {
596        let original = error.to_string();
597        let rollback = self
598            .file
599            .set_len(old_len)
600            .and_then(|()| self.file.seek(SeekFrom::Start(old_len)).map(drop))
601            .and_then(|()| self.file.sync_all());
602        match rollback {
603            Ok(()) => AuditError::Io(original),
604            Err(rollback) => AuditError::Io(format!(
605                "{original}; failed to roll back partial append: {rollback}"
606            )),
607        }
608    }
609
610    /// Scan the spool file for the decoded records and the logical length.
611    ///
612    /// The logical length is the byte offset immediately after the last
613    /// complete frame that decoded successfully. This method treats a truncated
614    /// or corrupt tail frame as end-of-data. `valid_bytes` then points to the
615    /// position immediately before that torn frame.
616    #[tracing::instrument(
617        level = "debug",
618        skip_all,
619        fields(records = Empty, valid_bytes = Empty),
620        err
621    )]
622    fn scan(&self) -> Result<ScannedSpool, AuditError> {
623        let buf = fs::read(&self.path).map_err(io)?;
624        let mut out = Vec::new();
625        let mut frame_lengths = Vec::new();
626        let mut cur: &[u8] = strip_file_header(SPOOL_FILE, &buf)?;
627        let mut valid_bytes = SpoolBytes(0);
628        while let Some((prefix, rest)) = cur.split_first_chunk::<4>() {
629            let len = usize::try_from(u32::from_be_bytes(*prefix)).unwrap_or(0);
630            let Some((frame, tail)) = rest.split_at_checked(len) else {
631                break; // truncated tail frame
632            };
633            match decode_record(frame) {
634                Some(rec) => {
635                    out.push(rec);
636                    let frame_len = u64::try_from(4 + len).unwrap_or(0);
637                    frame_lengths.push(frame_len);
638                    valid_bytes += SpoolBytes(frame_len);
639                }
640                None => break, // corrupt frame: stop (visible)
641            }
642            cur = tail;
643        }
644        let span = tracing::Span::current();
645        span.record("records", out.len());
646        span.record("valid_bytes", valid_bytes.0);
647        Ok(ScannedSpool {
648            records: out,
649            frame_lengths,
650            valid_bytes,
651        })
652    }
653
654    fn unread_record_count(
655        &self,
656        frame_lengths: &[u64],
657        valid_bytes: SpoolBytes,
658    ) -> Result<usize, AuditError> {
659        if self.replay_offset > valid_bytes.0 {
660            if valid_bytes.0 == 0 {
661                return Ok(0);
662            }
663            return Err(AuditError::Io(format!(
664                "audit replay offset {} exceeds spool length {}",
665                self.replay_offset, valid_bytes.0
666            )));
667        }
668        let mut offset = 0_u64;
669        let mut unread = 0_usize;
670        let mut at_boundary = self.replay_offset == 0;
671        for frame_len in frame_lengths {
672            if offset == self.replay_offset {
673                at_boundary = true;
674            }
675            if at_boundary {
676                unread = unread.saturating_add(1);
677            }
678            offset = offset.saturating_add(*frame_len);
679        }
680        if offset == self.replay_offset {
681            at_boundary = true;
682        }
683        if !at_boundary {
684            return Err(AuditError::Io(format!(
685                "audit replay offset {} is not a frame boundary",
686                self.replay_offset
687            )));
688        }
689        Ok(unread)
690    }
691
692    /// Read at most one replay batch, starting exactly at the cursor validated
693    /// by `open`. The committed physical prefix is neither read nor decoded.
694    pub(crate) fn read_replay_batch(&self) -> Result<Vec<ReplayRecord>, AuditError> {
695        let mut file = File::open(&self.path).map_err(io)?;
696        file.seek(SeekFrom::Start(
697            header_len_u64()
698                .checked_add(self.replay_offset)
699                .ok_or_else(|| AuditError::Io("audit replay seek overflow".into()))?,
700        ))
701        .map_err(io)?;
702
703        let mut records = Vec::with_capacity(
704            usize::try_from(self.count.0)
705                .unwrap_or(usize::MAX)
706                .min(REPLAY_BATCH_RECORDS),
707        );
708        for _ in 0..REPLAY_BATCH_RECORDS {
709            let mut prefix = [0_u8; 4];
710            match file.read(&mut prefix) {
711                Ok(0) => break,
712                Ok(read) => {
713                    let unread = prefix.get_mut(read..).ok_or_else(|| {
714                        AuditError::Io("audit spool read past the frame prefix".into())
715                    })?;
716                    file.read_exact(unread).map_err(io)?;
717                }
718                Err(error) => return Err(io(error)),
719            }
720            let len = usize::try_from(u32::from_be_bytes(prefix)).unwrap_or(0);
721            let mut body = vec![0_u8; len];
722            file.read_exact(&mut body).map_err(io)?;
723            let record = decode_record(&body)
724                .ok_or_else(|| AuditError::Io("corrupt audit spool frame during replay".into()))?;
725            records.push(ReplayRecord {
726                record,
727                frame_len: u64::try_from(4 + len).unwrap_or(u64::MAX),
728            });
729        }
730        Ok(records)
731    }
732
733    /// Read every record from the start of the spool, in order.
734    /// # Errors
735    /// Returns an error if the spool file cannot be opened or read.
736    pub fn read_all(&self) -> Result<Vec<AuditRecord>, AuditError> {
737        let ScannedSpool {
738            records,
739            frame_lengths,
740            valid_bytes,
741        } = self.scan()?;
742        let unread = self.unread_record_count(&frame_lengths, valid_bytes)?;
743        let first_unread = records.len().saturating_sub(unread);
744        Ok(records.into_iter().skip(first_unread).collect())
745    }
746
747    /// Clear the spool.
748    #[tracing::instrument(level = "debug", skip_all, fields(count = self.count.0, bytes = self.bytes.0), err)]
749    /// # Errors
750    /// Returns an error if the spool file cannot be truncated, or if the seek
751    /// or sync that follows fails.
752    pub fn truncate(&mut self) -> Result<(), AuditError> {
753        // Down to the header, which a spool file always keeps.
754        self.file.set_len(header_len_u64()).map_err(io)?;
755        self.bytes = SpoolBytes(0);
756        self.count = RecordCount(0);
757        self.unsynced = 0;
758        self.file
759            .seek(SeekFrom::Start(header_len_u64()))
760            .map_err(io)?;
761        self.file.sync_all().map_err(io)?;
762        persist_u64(
763            &self.path.with_file_name(REPLAY_OFFSET_FILE),
764            REPLAY_OFFSET_TMP,
765            0,
766        )?;
767        self.replay_offset = 0;
768        Ok(())
769    }
770}
771
772/// [`HEADER_LEN`] as a file offset.
773fn header_len_u64() -> u64 {
774    // A six-byte header always fits; the saturation is never reached.
775    u64::try_from(HEADER_LEN).unwrap_or(u64::MAX)
776}
777
778// cargo-mutants: I/O-only wrapper with no in-process signal (fsyncs directory entry).
779#[cfg(unix)]
780#[cfg_attr(test, mutants::skip)]
781fn sync_parent(path: &Path) -> Result<(), AuditError> {
782    File::open(path.parent().unwrap_or_else(|| Path::new(".")))
783        .map_err(io)?
784        .sync_all()
785        .map_err(io)
786}
787
788// cargo-mutants: I/O-only wrapper with no in-process signal (stats directory).
789#[cfg(not(unix))]
790#[cfg_attr(test, mutants::skip)]
791fn sync_parent(path: &Path) -> Result<(), AuditError> {
792    // Windows exposes no directory handle to fsync, so the directory entry is
793    // already durable by the time the file write returns. Stat the parent so a
794    // vanished spool directory still surfaces here rather than silently later.
795    path.parent()
796        .unwrap_or_else(|| Path::new("."))
797        .metadata()
798        .map_err(io)
799        .map(drop)
800}
801
802#[cfg(test)]
803mod tests {
804    use {
805        super::*,
806        crate::{
807            chain::{GENESIS_HEAD, chain_hash},
808            event::{
809                AuditEndpoint, AuditEvent, AuditEventClass, AuditOutcome, AuditPrincipal,
810                PrivilegedPhase,
811            },
812            ocsf::ProductInfo,
813            spool::test_support::{
814                LossySpool, ROOMY_CAP, SeededSpool, chained_record, check_empty_file,
815                reopen_after_losses, seeded_spool, spool_with_losses,
816            },
817            test_support::TempSpool,
818        },
819        assert2::check,
820        krabka_units::prelude::ByteSize,
821        std::{
822            env,
823            fs::{self, Permissions},
824            path::Path,
825            process::{self, Command, Stdio},
826        },
827    };
828
829    fn append_torn_frame(path: &Path, partial_body: &[u8]) {
830        use std::io::Write;
831
832        let mut file = fs::OpenOptions::new().append(true).open(path).unwrap();
833        file.write_all(&100u32.to_be_bytes()).unwrap();
834        file.write_all(partial_body).unwrap();
835    }
836
837    #[test]
838    fn append_then_read_round_trips_records_with_headers() {
839        let dir = tempfile::tempdir().unwrap();
840        let mut s = Spool::open(dir.path(), ROOMY_CAP).unwrap();
841        check!(s.is_empty());
842        let r0 = chained_record(0, &GENESIS_HEAD, b"{\"i\":0}");
843        let r1 = chained_record(1, &chain_hash(&GENESIS_HEAD, 0, b"{\"i\":0}"), b"{\"i\":1}");
844        check!((s.append(&r0).unwrap(), s.append(&r1).unwrap()) == (true, true));
845        let read = s.read_all().unwrap();
846        check!((s.count().0, s.is_empty(), read) == (2, false, vec![r0.clone(), r1.clone()]));
847    }
848
849    /// The privileged-action event shares the `ApiActivity` class tag, so the
850    /// frame format carries it with no change. This proves the round trip on a
851    /// record the broker actually builds, headers and OCSF value included.
852    #[test]
853    fn privileged_action_record_round_trips_through_the_frame_format() {
854        let product = ProductInfo {
855            vendor_name: "Krabka".into(),
856            name: "krabka-broker".into(),
857            version: "0".into(),
858        };
859        let event = AuditEvent::PrivilegedAction {
860            outcome: AuditOutcome::Success,
861            phase: PrivilegedPhase::Approved,
862            action: "unclean_elect_leaders".into(),
863            target: "orders-3".into(),
864            proposal_id: "bg-7".into(),
865            principal: AuditPrincipal {
866                name: "User:bob".into(),
867                auth_method: "MTls".into(),
868            },
869            counterparties: vec![AuditPrincipal {
870                name: "User:alice".into(),
871                auth_method: "MTls".into(),
872            }],
873            approver_set_fingerprint: "f00dcafe".into(),
874            key_id: "op-1".into(),
875            signature: vec![0xde, 0xad, 0xbe, 0xef],
876            signature_verified: true,
877            signed_at_ms: 0,
878            source: AuditEndpoint {
879                ip: "10.0.0.4".into(),
880                port: 9092,
881            },
882            reason: "incident 42".into(),
883            time_ms: 10,
884        };
885        let mut rec = AuditRecord::from_event(&event, &product);
886        rec.push_chain_headers(0, &GENESIS_HEAD);
887
888        let dir = tempfile::tempdir().unwrap();
889        let mut s = Spool::open(dir.path(), ROOMY_CAP).unwrap();
890        check!(s.append(&rec).unwrap());
891        // Reopen, so the read goes through the on-disk frames and not through
892        // any state `append` kept in memory.
893        let s = Spool::open(dir.path(), ROOMY_CAP).unwrap();
894        check!(s.read_all().unwrap() == vec![rec.clone()]);
895        check!(rec.class == AuditEventClass::ApiActivity);
896    }
897
898    #[test]
899    fn overflow_is_rejected_not_truncated() {
900        let dir = tempfile::tempdir().unwrap();
901        // tiny cap: first record fits, second does not
902        let r = chained_record(0, &GENESIS_HEAD, b"0123456789");
903        let one = {
904            let mut s = Spool::open(dir.path(), ROOMY_CAP).unwrap();
905            s.append(&r).unwrap();
906            s.size()
907        };
908        let dir2 = tempfile::tempdir().unwrap();
909        let mut s = Spool::open(dir2.path(), one).unwrap(); // exactly one record fits
910        check!(s.append(&r).unwrap()); // accepted
911        check!(!s.append(&r).unwrap()); // rejected (would exceed the cap)
912        check!((s.count().0, s.read_all().unwrap().len()) == (1, 1)); // not corrupted
913    }
914
915    #[test]
916    fn append_rejects_size_overflow_without_writing() {
917        let dir = tempfile::tempdir().unwrap();
918        let mut spool = Spool::open(dir.path(), ByteSize::from_bytes(u64::MAX)).unwrap();
919        spool.bytes = SpoolBytes(u64::MAX);
920
921        check!(
922            !spool
923                .append(&chained_record(0, &GENESIS_HEAD, b"x"))
924                .unwrap()
925        );
926        check!((spool.bytes.0, spool.count.0) == (u64::MAX, 0));
927        check_empty_file(dir.path());
928    }
929
930    #[test]
931    fn replay_cursor_keeps_only_unacknowledged_records_after_reopen() {
932        let dir = tempfile::tempdir().unwrap();
933        let mut s = Spool::open(dir.path(), ROOMY_CAP).unwrap();
934        let r0 = chained_record(0, &GENESIS_HEAD, b"a");
935        let r1 = chained_record(1, &GENESIS_HEAD, b"b");
936        let r2 = chained_record(2, &GENESIS_HEAD, b"c");
937        s.append(&r0).unwrap();
938        s.append(&r1).unwrap();
939        s.append(&r2).unwrap();
940        let physical_size = s.size();
941        s.begin_replay(&r0).unwrap();
942        s.commit_replay(&r0).unwrap();
943        drop(s);
944
945        let mut s = Spool::open(dir.path(), ROOMY_CAP).unwrap();
946        check!(s.size() > ByteSize::ZERO); // `*=` mutant would leave bytes at 0
947        check!(s.size() == physical_size);
948        check!((s.count().0, s.read_all().unwrap()) == (2, vec![r1, r2]));
949        s.truncate().unwrap();
950        check!((s.is_empty(), s.read_all().unwrap().is_empty()) == (true, true));
951    }
952
953    #[test]
954    fn replay_batches_are_bounded_and_seek_past_the_committed_prefix() {
955        let dir = tempfile::tempdir().unwrap();
956        let mut spool = Spool::open(dir.path(), ROOMY_CAP).unwrap();
957        for sequence in 0..(REPLAY_BATCH_RECORDS + 3) {
958            check!(
959                spool
960                    .append(&chained_record(
961                        u64::try_from(sequence).unwrap(),
962                        &GENESIS_HEAD,
963                        &[u8::try_from(sequence).unwrap()],
964                    ))
965                    .unwrap()
966            );
967        }
968
969        let first = spool.read_replay_batch().unwrap();
970        check!(first.len() == REPLAY_BATCH_RECORDS);
971        for replay in first {
972            spool.commit_replay_frame(replay.frame_len).unwrap();
973        }
974        check!(spool.count().0 == 3);
975
976        // Damage a byte in the already committed prefix. A replay reader that
977        // rescans from the file header would now fail; the cursor reader seeks
978        // directly to the previously validated frame boundary.
979        let mut file = OpenOptions::new()
980            .write(true)
981            .open(dir.path().join(SPOOL_FILE))
982            .unwrap();
983        file.seek(SeekFrom::Start(header_len_u64() + 4)).unwrap();
984        file.write_all(&[0xff]).unwrap();
985
986        let remaining = spool.read_replay_batch().unwrap();
987        check!(remaining.len() == 3);
988        check!(remaining[0].record.value == vec![u8::try_from(REPLAY_BATCH_RECORDS).unwrap()]);
989    }
990
991    /// Reproducible large-backlog probe used for the replay-read performance
992    /// evidence. It is ignored because it writes about 75 MiB and measures
993    /// wall-clock time rather than asserting a machine-dependent threshold.
994    #[test]
995    #[ignore = "manual audit spool performance probe"]
996    fn measure_replay_batch_across_large_backlogs() {
997        use std::{io::BufWriter, time::Instant};
998
999        for backlog in [1_024_usize, 8_192] {
1000            let dir = tempfile::tempdir().unwrap();
1001            drop(Spool::open(dir.path(), ByteSize::from_bytes(u64::MAX)).unwrap());
1002            let record = chained_record(0, &GENESIS_HEAD, &vec![b'x'; 8 * 1_024]);
1003            let frame = encode_frame(&record).unwrap();
1004            let committed = backlog - REPLAY_BATCH_RECORDS;
1005            let replay_offset = u64::try_from(committed * frame.len()).unwrap();
1006
1007            let file = File::create(dir.path().join(SPOOL_FILE)).unwrap();
1008            let mut writer = BufWriter::new(file);
1009            writer.write_all(&file_header()).unwrap();
1010            for _ in 0..backlog {
1011                writer.write_all(&frame).unwrap();
1012            }
1013            writer.into_inner().unwrap().sync_all().unwrap();
1014            persist_u64(
1015                &dir.path().join(REPLAY_OFFSET_FILE),
1016                REPLAY_OFFSET_TMP,
1017                replay_offset,
1018            )
1019            .unwrap();
1020
1021            let started = Instant::now();
1022            let spool = Spool::open(dir.path(), ByteSize::from_bytes(u64::MAX)).unwrap();
1023            let recovery = started.elapsed();
1024            let started = Instant::now();
1025            let batch = spool.read_replay_batch().unwrap();
1026            let replay = started.elapsed();
1027            let replay_bytes: u64 = batch.iter().map(|record| record.frame_len).sum();
1028            let peak_kib = fs::read_to_string("/proc/self/status")
1029                .ok()
1030                .and_then(|status| {
1031                    status.lines().find_map(|line| {
1032                        line.strip_prefix("VmHWM:")?
1033                            .split_whitespace()
1034                            .next()?
1035                            .parse::<u64>()
1036                            .ok()
1037                    })
1038                });
1039            let recovery_mib_per_second = u128::from(spool.size().bytes_u64())
1040                .saturating_mul(1_000_000_000)
1041                / recovery.as_nanos().max(1)
1042                / (1_u128 << 20);
1043            eprintln!(
1044                "backlog={backlog} spool_bytes={} recovery_us={} recovery_mib_per_second={recovery_mib_per_second} replay_records={} replay_bytes={replay_bytes} replay_us={} process_peak_kib={peak_kib:?}",
1045                spool.size().bytes_u64(),
1046                recovery.as_micros(),
1047                batch.len(),
1048                replay.as_micros(),
1049            );
1050        }
1051    }
1052
1053    #[test]
1054    fn append_accepts_record_that_exactly_fills_to_max() {
1055        let probe = tempfile::tempdir().unwrap();
1056        let r = chained_record(0, &GENESIS_HEAD, b"payload");
1057        let one = {
1058            let mut s = Spool::open(probe.path(), ROOMY_CAP).unwrap();
1059            s.append(&r).unwrap();
1060            s.size()
1061        };
1062        // Cap at exactly two records: the 2nd append fills to max and MUST be
1063        // accepted (bytes + frame == max, not >). The `+ -> *` mutant computes
1064        // bytes * frame, which exceeds max and would wrongly reject.
1065        let dir = tempfile::tempdir().unwrap();
1066        let mut s = Spool::open(dir.path(), one * 2.0).unwrap();
1067        check!(s.append(&r).unwrap());
1068        check!(s.append(&r).unwrap());
1069        check!(s.count() == 2);
1070    }
1071
1072    #[test]
1073    fn size_reports_the_bytes_actually_on_disk() {
1074        // Anchors the `SpoolBytes` -> `ByteSize` accessor to the real file
1075        // length, so a scale slip at that seam cannot hide behind a matching
1076        // slip at the `open` seam.
1077        let dir = tempfile::tempdir().unwrap();
1078        let mut s = Spool::open(dir.path(), ROOMY_CAP).unwrap();
1079        s.append(&chained_record(0, &GENESIS_HEAD, b"payload"))
1080            .unwrap();
1081        let on_disk = fs::metadata(dir.path().join(SPOOL_FILE)).unwrap().len();
1082        check!(s.size() == ByteSize::from_bytes(on_disk - header_len_u64()));
1083    }
1084
1085    #[test]
1086    fn open_heals_torn_tail_frame() {
1087        let SeededSpool {
1088            dir,
1089            record: r0,
1090            spool,
1091        } = seeded_spool(b"good");
1092        drop(spool);
1093
1094        // Simulate a crash mid-append: a length prefix claiming 100 bytes, only 3 follow.
1095        append_torn_frame(&dir.path().join("audit.spool"), b"abc");
1096        // Reopen heals the torn tail; the good record survives and appends continue contiguously.
1097        let mut s = Spool::open(dir.path(), ROOMY_CAP).unwrap();
1098        assert2::check!((s.count().0, s.read_all().unwrap()) == (1, vec![r0.clone()]));
1099        let r1 = chained_record(1, &GENESIS_HEAD, b"more");
1100        assert2::check!(s.append(&r1).unwrap());
1101        assert2::check!(s.read_all().unwrap() == vec![r0, r1]);
1102    }
1103
1104    /// `open` heals a torn tail and reports that it did. Opening an intact
1105    /// spool must report the opposite: healing one that was whole would mean
1106    /// cutting good bytes off the end.
1107    #[test]
1108    fn open_reports_whether_it_healed_a_torn_tail() {
1109        let SeededSpool { dir, spool, .. } = seeded_spool(b"good");
1110        drop(spool);
1111        let spool_file = dir.path().join(SPOOL_FILE);
1112
1113        let intact_len = fs::metadata(&spool_file).unwrap().len();
1114        {
1115            let intact = Spool::open(dir.path(), ROOMY_CAP).unwrap();
1116            check!(!intact.recovered_torn_tail());
1117        }
1118        check!(fs::metadata(&spool_file).unwrap().len() == intact_len);
1119
1120        // A crash mid-append: a length prefix promising 100 bytes, 3 behind it.
1121        append_torn_frame(&spool_file, b"abc");
1122        let healed = Spool::open(dir.path(), ROOMY_CAP).unwrap();
1123        check!(healed.recovered_torn_tail());
1124        check!(fs::metadata(&spool_file).unwrap().len() == intact_len);
1125    }
1126
1127    #[test]
1128    fn reopen_recovers_existing_contents() {
1129        let dir = tempfile::tempdir().unwrap();
1130        let r0 = chained_record(0, &GENESIS_HEAD, b"x");
1131        {
1132            let mut s = Spool::open(dir.path(), ROOMY_CAP).unwrap();
1133            s.append(&r0).unwrap();
1134        }
1135        let s2 = Spool::open(dir.path(), ROOMY_CAP).unwrap();
1136        check!((s2.count().0, s2.read_all().unwrap()) == (1, vec![r0]));
1137    }
1138
1139    #[test]
1140    fn cadence_and_explicit_sync_reset_the_unsynced_counter() {
1141        let dir = tempfile::tempdir().unwrap();
1142        let mut spool =
1143            Spool::open_with_sync_every(dir.path(), ROOMY_CAP, NonZeroU64::new(2).unwrap())
1144                .unwrap();
1145        check!(
1146            spool
1147                .append(&chained_record(0, &GENESIS_HEAD, b"a"))
1148                .unwrap()
1149        );
1150        check!(spool.unsynced == 1);
1151        check!(
1152            spool
1153                .append(&chained_record(1, &GENESIS_HEAD, b"b"))
1154                .unwrap()
1155        );
1156        check!(spool.unsynced == 0);
1157        check!(
1158            spool
1159                .append(&chained_record(2, &GENESIS_HEAD, b"c"))
1160                .unwrap()
1161        );
1162        check!(spool.unsynced == 1);
1163        spool.sync().unwrap();
1164        check!(spool.unsynced == 0);
1165    }
1166
1167    #[test]
1168    fn acknowledged_record_survives_child_crash_and_torn_tail() {
1169        const CHILD_DIR: &str = "KRABKA_AUDIT_SPOOL_CRASH_DIR";
1170        if let Some(dir) = env::var_os(CHILD_DIR) {
1171            let dir = PathBuf::from(dir);
1172            let mut spool = Spool::open(&dir, ROOMY_CAP).unwrap();
1173            let record = chained_record(0, &GENESIS_HEAD, b"acknowledged");
1174            check!(spool.append(&record).unwrap());
1175            drop(spool);
1176
1177            // Model a process dying after the next frame's prefix and only a
1178            // fragment of its body reached the file.
1179            append_torn_frame(&dir.join(SPOOL_FILE), b"torn");
1180            process::abort();
1181        }
1182
1183        let dir = tempfile::tempdir().unwrap();
1184        let status = Command::new(env::current_exe().unwrap())
1185            .arg("acknowledged_record_survives_child_crash_and_torn_tail")
1186            .env(CHILD_DIR, dir.path())
1187            .stdout(Stdio::null())
1188            .stderr(Stdio::null())
1189            .status()
1190            .unwrap();
1191        check!(!status.success());
1192
1193        let spool = Spool::open(dir.path(), ROOMY_CAP).unwrap();
1194        let record = chained_record(0, &GENESIS_HEAD, b"acknowledged");
1195        check!(spool.recovered_torn_tail());
1196        check!(spool.read_all().unwrap() == vec![record]);
1197    }
1198
1199    #[test]
1200    fn pending_loss_survives_reopen_and_reconciles_a_durable_marker() {
1201        let LossySpool { dir, spool, losses } = spool_with_losses(ByteSize::from_bytes(0), 3);
1202        losses.persist().unwrap();
1203        let batch = losses.snapshot().unwrap();
1204        drop(losses);
1205        drop(spool);
1206
1207        let mut reopened = Spool::open(dir.path(), ByteSize::from_bytes(0)).unwrap();
1208        check!(reopened.pending_losses().count() == 3);
1209        let mut marker = AuditRecord::records_lost(batch.count, batch.generation);
1210        marker.push_chain_headers(0, &GENESIS_HEAD);
1211        check!(reopened.append_loss_marker(&marker).is_err());
1212        check!(reopened.size() == ByteSize::ZERO);
1213        drop(reopened);
1214
1215        let mut reopened = Spool::open(dir.path(), ROOMY_CAP).unwrap();
1216        reopened.append_loss_marker(&marker).unwrap();
1217        // Model a crash after the marker fsync but before clearing the sidecar.
1218        drop(reopened);
1219
1220        let reopened = Spool::open(dir.path(), ByteSize::from_bytes(0)).unwrap();
1221        check!(reopened.pending_losses().count() == 0);
1222        check!(reopened.read_all().unwrap() == vec![marker]);
1223    }
1224
1225    #[test]
1226    fn uncertain_replay_poison_survives_reopen_until_record_is_removed() {
1227        let dir = tempfile::tempdir().unwrap();
1228        let record = chained_record(0, &GENESIS_HEAD, b"uncertain");
1229        let mut spool = Spool::open(dir.path(), ROOMY_CAP).unwrap();
1230        check!(spool.append(&record).unwrap());
1231        spool.begin_replay(&record).unwrap();
1232
1233        check!(matches!(
1234            Spool::open(dir.path(), ROOMY_CAP),
1235            Err(AuditError::Poisoned(_))
1236        ));
1237
1238        spool.commit_replay(&record).unwrap();
1239        drop(spool);
1240        let spool = Spool::open(dir.path(), ROOMY_CAP).unwrap();
1241        check!(spool.is_empty());
1242        check!(!dir.path().join(REPLAY_POISON_FILE).exists());
1243    }
1244
1245    #[test]
1246    fn commit_and_persist_pending_losses() {
1247        let LossySpool { dir, spool, losses } = spool_with_losses(ROOMY_CAP, 5);
1248        losses.persist().unwrap();
1249        let batch = losses.snapshot().unwrap();
1250        let wrong_batch = AuditLosses {
1251            generation: batch.generation + 1,
1252            count: batch.count,
1253        };
1254        losses.commit(wrong_batch);
1255        check!(losses.count() == 5);
1256
1257        losses.commit(batch);
1258        check!(losses.count() == 0);
1259
1260        let reopened = reopen_after_losses(dir.path(), spool, losses);
1261        check!(reopened.pending_losses().count() == 0);
1262    }
1263
1264    #[test]
1265    fn reconcile_requires_matching_class_and_generation() {
1266        let LossySpool {
1267            dir: _dir,
1268            spool: _spool,
1269            losses,
1270        } = spool_with_losses(ROOMY_CAP, 4);
1271        let generation = losses.snapshot().unwrap().generation;
1272
1273        let payload =
1274            serde_json::to_vec(&serde_json::json!({ "loss_generation": generation })).unwrap();
1275        let app_event = chained_record(0, &GENESIS_HEAD, &payload);
1276        losses.reconcile(&[app_event]);
1277        check!(losses.count() == 4);
1278
1279        let wrong_gen_marker = AuditRecord::records_lost(4, generation + 1);
1280        losses.reconcile(&[wrong_gen_marker]);
1281        check!(losses.count() == 4);
1282
1283        let matching_marker = AuditRecord::records_lost(4, generation);
1284        losses.reconcile(&[matching_marker]);
1285        check!(losses.count() == 0);
1286    }
1287
1288    /// A loss `AuditHandle::emit` adds while the writer's marker is in flight
1289    /// is neither settled by that marker nor left in its generation.
1290    #[test]
1291    fn commit_carries_a_concurrent_loss_into_a_fresh_generation() {
1292        let LossySpool { dir, spool, losses } = spool_with_losses(ROOMY_CAP, 3);
1293        let batch = losses.persist_with(Ok).unwrap().unwrap();
1294        check!(
1295            losses.state()
1296                == AuditLosses {
1297                    generation: batch.generation,
1298                    count: 0,
1299                }
1300        );
1301
1302        losses.add(2);
1303        let batch = losses.snapshot().unwrap();
1304        losses.persist().unwrap();
1305        losses.add(1);
1306        losses.commit(batch);
1307        let carried = AuditLosses {
1308            generation: batch.generation + 1,
1309            count: 1,
1310        };
1311        check!(losses.state() == carried);
1312        check!(losses.snapshot() == Some(carried));
1313
1314        let reopened = reopen_after_losses(dir.path(), spool, losses);
1315        check!(reopened.pending_losses().state() == carried);
1316    }
1317
1318    /// Crash between the marker's fsync and the commit, with a concurrent loss
1319    /// already in the sidecar: reconciliation settles the marker's count only,
1320    /// and a later reopen does not settle the remainder a second time.
1321    #[test]
1322    fn reconcile_settles_only_the_markers_count() {
1323        let LossySpool {
1324            dir,
1325            mut spool,
1326            losses,
1327        } = spool_with_losses(ROOMY_CAP, 3);
1328        let batch = losses.snapshot().unwrap();
1329        losses.persist().unwrap();
1330        losses.add(1);
1331        losses.persist().unwrap();
1332        let mut marker = AuditRecord::records_lost(batch.count, batch.generation);
1333        marker.push_chain_headers(0, &GENESIS_HEAD);
1334        spool.append_loss_marker(&marker).unwrap();
1335        drop(losses);
1336        drop(spool);
1337
1338        let carried = AuditLosses {
1339            generation: batch.generation + 1,
1340            count: 1,
1341        };
1342        for _ in 0..2 {
1343            let reopened = Spool::open(dir.path(), ROOMY_CAP).unwrap();
1344            check!(reopened.pending_losses().state() == carried);
1345            check!(reopened.read_all().unwrap() == vec![marker.clone()]);
1346        }
1347    }
1348
1349    #[cfg(unix)]
1350    fn is_root() -> bool {
1351        if let Ok(status) = fs::read_to_string("/proc/self/status") {
1352            for line in status.lines() {
1353                if let Some(rest) = line.strip_prefix("Uid:") {
1354                    let fields: Vec<&str> = rest.split_whitespace().collect();
1355                    if let Some(euid) = fields.get(1) {
1356                        return *euid == "0";
1357                    }
1358                }
1359            }
1360        }
1361        false
1362    }
1363
1364    #[test]
1365    #[cfg(unix)]
1366    fn read_or_create_u64_propagates_non_not_found_error() {
1367        use std::os::unix::fs::PermissionsExt;
1368        let dir = tempfile::tempdir().unwrap();
1369        let path = dir.path().join("forbidden");
1370        fs::write(&path, [0u8; 8]).unwrap();
1371        fs::set_permissions(&path, Permissions::from_mode(0o000)).unwrap();
1372        if is_root() || fs::File::open(&path).is_ok() {
1373            return;
1374        }
1375        let result = read_or_create_u64(&path, "tmp");
1376        check!(matches!(result, Err(AuditError::Io(_))));
1377    }
1378
1379    #[test]
1380    fn abort_replay_clears_poison_and_clear_poison_handles_absent() {
1381        let SeededSpool { dir, record, spool } = seeded_spool(b"data");
1382
1383        check!(spool.clear_replay_poison().is_ok());
1384
1385        spool.begin_replay(&record).unwrap();
1386        check!(dir.path().join(REPLAY_POISON_FILE).exists());
1387
1388        spool.abort_replay().unwrap();
1389        check!(!dir.path().join(REPLAY_POISON_FILE).exists());
1390    }
1391
1392    #[test]
1393    fn reconcile_replay_poison_detects_torn_frame() {
1394        let dir = tempfile::tempdir().unwrap();
1395        let poison_path = dir.path().join(REPLAY_POISON_FILE);
1396        let mut corrupted = file_header().to_vec();
1397        corrupted.extend_from_slice(&0u64.to_be_bytes());
1398        corrupted.extend_from_slice(&100u32.to_be_bytes());
1399        corrupted.extend_from_slice(b"abcd");
1400        fs::write(&poison_path, &corrupted).unwrap();
1401
1402        let replay_offset_path = dir.path().join(REPLAY_OFFSET_FILE);
1403        fs::write(&replay_offset_path, with_header(&100u64.to_be_bytes())).unwrap();
1404
1405        let res = Spool::open(dir.path(), ROOMY_CAP);
1406        check!(matches!(res, Err(AuditError::Poisoned(_))));
1407    }
1408
1409    #[test]
1410    fn replayed_spool_is_truncated_on_reopen() {
1411        let SeededSpool {
1412            dir,
1413            record,
1414            mut spool,
1415        } = seeded_spool(b"hello");
1416        spool.commit_replay(&record).unwrap();
1417        drop(spool);
1418
1419        let reopened = Spool::open(dir.path(), ROOMY_CAP).unwrap();
1420        check!(reopened.size() == ByteSize::ZERO);
1421        check_empty_file(dir.path());
1422    }
1423
1424    #[test]
1425    fn unread_records_rejects_replay_offset_past_valid_bytes() {
1426        let SeededSpool { dir, spool, .. } = seeded_spool(b"valid");
1427        let valid_len = spool.size().bytes_u64();
1428        drop(spool);
1429
1430        let replay_offset_path = dir.path().join(REPLAY_OFFSET_FILE);
1431        fs::write(
1432            &replay_offset_path,
1433            with_header(&(valid_len + 100).to_be_bytes()),
1434        )
1435        .unwrap();
1436
1437        let res = Spool::open(dir.path(), ROOMY_CAP);
1438        check!(matches!(res, Err(AuditError::Io(_))));
1439    }
1440
1441    #[test]
1442    #[cfg(unix)]
1443    fn clear_replay_poison_propagates_permission_denied() {
1444        use std::os::unix::fs::PermissionsExt;
1445        let SeededSpool { dir, record, spool } = seeded_spool(b"data");
1446        spool.begin_replay(&record).unwrap();
1447
1448        fs::set_permissions(dir.path(), Permissions::from_mode(0o555)).unwrap();
1449        if is_root() || fs::write(dir.path().join("probe"), b"probe").is_ok() {
1450            let _ = fs::set_permissions(dir.path(), Permissions::from_mode(0o755));
1451            let _ = fs::remove_file(dir.path().join("probe"));
1452            return;
1453        }
1454        let res = spool.clear_replay_poison();
1455        let _ = fs::set_permissions(dir.path(), Permissions::from_mode(0o755));
1456        check!(matches!(res, Err(AuditError::Io(_))));
1457    }
1458
1459    #[test]
1460    #[cfg(unix)]
1461    fn reconcile_replay_poison_propagates_read_error() {
1462        use std::os::unix::fs::PermissionsExt;
1463        let dir = tempfile::tempdir().unwrap();
1464        let poison_path = dir.path().join(REPLAY_POISON_FILE);
1465        fs::write(&poison_path, [0u8; 16]).unwrap();
1466        fs::set_permissions(&poison_path, Permissions::from_mode(0o000)).unwrap();
1467        if is_root() || fs::File::open(&poison_path).is_ok() {
1468            let _ = fs::set_permissions(&poison_path, Permissions::from_mode(0o644));
1469            return;
1470        }
1471
1472        let res = Spool::open(dir.path(), ROOMY_CAP);
1473        let _ = fs::set_permissions(&poison_path, Permissions::from_mode(0o644));
1474        check!(matches!(res, Err(AuditError::Io(_))));
1475    }
1476
1477    /// `body` behind the current file header, as the spool writes it.
1478    fn with_header(body: &[u8]) -> Vec<u8> {
1479        let mut bytes = file_header().to_vec();
1480        bytes.extend_from_slice(body);
1481        bytes
1482    }
1483
1484    /// One record, frame by frame, as `audit.spool` holds it. These bytes are
1485    /// the 1.x on-disk format: a change to them breaks every 1.x spool.
1486    const GOLDEN_SPOOL: &[u8] = &[
1487        b'K', b'A', b'U', b'D', 0x00, 0x00, // magic, version 0
1488        0x00, 0x00, 0x00, 0x16, // frame length 22
1489        0x03, // class tag: application_lifecycle
1490        0x00, 0x00, 0x00, 0x02, b'{', b'}', // value
1491        0x00, 0x00, 0x00, 0x01, // one header
1492        0x00, 0x00, 0x00, 0x01, b'k', // key
1493        0x00, 0x00, 0x00, 0x02, b'v', b'1', // value
1494    ];
1495
1496    const FUTURE_HEADER: &[u8] = &[b'K', b'A', b'U', b'D', 0x00, 0x01];
1497    const NEGATIVE_HEADER: &[u8] = &[b'K', b'A', b'U', b'D', 0xff, 0xff];
1498
1499    fn golden_record() -> AuditRecord {
1500        AuditRecord {
1501            class: AuditEventClass::ApplicationLifecycle,
1502            value: b"{}".to_vec(),
1503            headers: vec![("k".into(), b"v1".to_vec())],
1504        }
1505    }
1506
1507    /// A sidecar file and the golden bytes it must hold.
1508    struct ExpectedSidecar {
1509        file: &'static str,
1510        want: Vec<u8>,
1511    }
1512
1513    fn spooled_golden_record() -> TempSpool {
1514        let dir = tempfile::tempdir().unwrap();
1515        let mut spool = Spool::open(dir.path(), ROOMY_CAP).unwrap();
1516        check!(spool.append(&golden_record()).unwrap());
1517        TempSpool { dir, spool }
1518    }
1519
1520    #[test]
1521    fn spool_file_is_the_golden_bytes() {
1522        let TempSpool { dir, spool: _spool } = spooled_golden_record();
1523
1524        check!(fs::read(dir.path().join(SPOOL_FILE)).unwrap() == GOLDEN_SPOOL);
1525    }
1526
1527    #[test]
1528    fn golden_spool_file_reads_back_as_its_record() {
1529        let dir = tempfile::tempdir().unwrap();
1530        fs::write(dir.path().join(SPOOL_FILE), GOLDEN_SPOOL).unwrap();
1531        let spool = Spool::open(dir.path(), ROOMY_CAP).unwrap();
1532
1533        check!(
1534            (spool.recovered_torn_tail(), spool.read_all().unwrap())
1535                == (false, vec![golden_record()])
1536        );
1537    }
1538
1539    /// The sidecars, as a spool that has spooled the golden record, started
1540    /// its replay and lost three records in generation 1 leaves them.
1541    #[test]
1542    fn sidecar_files_are_the_golden_bytes() {
1543        let TempSpool { dir, spool } = spooled_golden_record();
1544        spool.pending_losses().add(3);
1545        spool.pending_losses().persist().unwrap();
1546        spool.begin_replay(&golden_record()).unwrap();
1547
1548        let header: &[u8] = &[b'K', b'A', b'U', b'D', 0x00, 0x00];
1549        let cases: [ExpectedSidecar; 3] = [
1550            ExpectedSidecar {
1551                file: LOSS_STATE_FILE,
1552                want: [header, &1u64.to_be_bytes(), &3u64.to_be_bytes()].concat(),
1553            },
1554            ExpectedSidecar {
1555                file: REPLAY_OFFSET_FILE,
1556                want: [header, &0u64.to_be_bytes()].concat(),
1557            },
1558            ExpectedSidecar {
1559                file: REPLAY_POISON_FILE,
1560                want: [header, &0u64.to_be_bytes(), &GOLDEN_SPOOL[6..]].concat(),
1561            },
1562        ];
1563        for ExpectedSidecar { file, want } in cases {
1564            check!(fs::read(dir.path().join(file)).unwrap() == want, "{file}");
1565        }
1566    }
1567
1568    /// The sidecars read back: the loss state and replay offset come back
1569    /// whole, and a poison file left behind by a replay the offset already
1570    /// passed is cleared.
1571    #[test]
1572    fn golden_sidecar_files_read_back() {
1573        let dir = tempfile::tempdir().unwrap();
1574        let header: &[u8] = &[b'K', b'A', b'U', b'D', 0x00, 0x00];
1575        let frame_len = u64::try_from(GOLDEN_SPOOL.len() - 6).unwrap();
1576        let golden_two = [GOLDEN_SPOOL, &GOLDEN_SPOOL[6..]].concat();
1577        fs::write(dir.path().join(SPOOL_FILE), &golden_two).unwrap();
1578        fs::write(
1579            dir.path().join(LOSS_STATE_FILE),
1580            [header, &1u64.to_be_bytes(), &3u64.to_be_bytes()].concat(),
1581        )
1582        .unwrap();
1583        fs::write(
1584            dir.path().join(REPLAY_OFFSET_FILE),
1585            [header, &frame_len.to_be_bytes()].concat(),
1586        )
1587        .unwrap();
1588        fs::write(
1589            dir.path().join(REPLAY_POISON_FILE),
1590            [header, &0u64.to_be_bytes(), &GOLDEN_SPOOL[6..]].concat(),
1591        )
1592        .unwrap();
1593
1594        let spool = Spool::open(dir.path(), ROOMY_CAP).unwrap();
1595        check!(
1596            (
1597                spool.pending_losses().state(),
1598                spool.replay_offset,
1599                spool.read_all().unwrap(),
1600                dir.path().join(REPLAY_POISON_FILE).exists(),
1601            ) == (
1602                AuditLosses {
1603                    generation: 1,
1604                    count: 3,
1605                },
1606                frame_len,
1607                vec![golden_record()],
1608                false,
1609            )
1610        );
1611    }
1612
1613    /// Every file in the spool directory refuses a header it does not know:
1614    /// none at all (the 0.x layout, which started straight with the data) is
1615    /// `found: None`, and a version other than 0 is named.
1616    #[yare::parameterized(
1617        spool_v0_frame = { SPOOL_FILE, GOLDEN_SPOOL[6..].to_vec(), None },
1618        spool_empty = { SPOOL_FILE, Vec::new(), None },
1619        spool_short_header = { SPOOL_FILE, b"KAU".to_vec(), None },
1620        spool_future_version = {
1621            SPOOL_FILE,
1622            [FUTURE_HEADER, &GOLDEN_SPOOL[6..]].concat(),
1623            Some(1),
1624        },
1625        spool_negative_version = { SPOOL_FILE, NEGATIVE_HEADER.to_vec(), Some(-1) },
1626        losses_empty_version = { LOSS_STATE_FILE, [0u8; 16].to_vec(), None },
1627        losses_future_version = { LOSS_STATE_FILE, [FUTURE_HEADER, &[0u8; 16]].concat(), Some(1) },
1628        replay_offset_empty_version = { REPLAY_OFFSET_FILE, [0u8; 8].to_vec(), None },
1629        replay_offset_future_version = {
1630            REPLAY_OFFSET_FILE,
1631            [FUTURE_HEADER, &[0u8; 8]].concat(),
1632            Some(1),
1633        },
1634        replay_poison_empty_version = {
1635            REPLAY_POISON_FILE,
1636            [&0u64.to_be_bytes()[..], &GOLDEN_SPOOL[6..]].concat(),
1637            None,
1638        },
1639        replay_poison_future_version = {
1640            REPLAY_POISON_FILE,
1641            [FUTURE_HEADER, &0u64.to_be_bytes(), &GOLDEN_SPOOL[6..]].concat(),
1642            Some(1),
1643        },
1644    )]
1645    fn open_refuses_a_missing_or_unknown_file_header(
1646        file: &str,
1647        contents: Vec<u8>,
1648        found: Option<i16>,
1649    ) {
1650        let dir = tempfile::tempdir().unwrap();
1651        fs::write(dir.path().join(file), &contents).unwrap();
1652        let got = match Spool::open(dir.path(), ROOMY_CAP) {
1653            Err(AuditError::UnsupportedSpoolFormat { file, found }) => Some((file, found)),
1654            _ => None,
1655        };
1656        check!(
1657            got == Some((file.to_owned(), found)),
1658            "{file} {contents:x?}"
1659        );
1660    }
1661
1662    #[yare::parameterized(
1663        no_header = {
1664            None,
1665            "audit spool file audit.spool: has no version header; it predates krabka 1.0, so the audit spool directory must be emptied (the node reformatted) before this broker starts",
1666        },
1667        future_version = {
1668            Some(3),
1669            "audit spool file audit.spool: has version 3, which this build does not read",
1670        },
1671    )]
1672    fn unsupported_spool_format_names_the_file_and_the_version(found: Option<i16>, want: &str) {
1673        let error = AuditError::UnsupportedSpoolFormat {
1674            file: SPOOL_FILE.into(),
1675            found,
1676        };
1677        check!(error.to_string() == want);
1678    }
1679}
krabka_audit/spool.rs

krabka_audit/
spool.rs
