Add additional doc comments; rename checkpoint API names for clarity.

Author: AaronRM

rust/otap-dataflow/crates/quiver/src/wal/checkpoint_sidecar.rs

@@ -1,11 +1,22 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
-// Provides read/write helpers for the `checkpoint.offset` sidecar that tracks the
-// global WAL consumer checkpoint (data bytes excluding headers) still required for crash
-// recovery. The sidecar lets new processes resume from a known safe offset
-// without rescanning the entire WAL. It carries a CRC so we can discard corrupted
-// metadata safely.
+//! Crash-safe checkpoint offset persistence.
+//!
+//! The checkpoint sidecar is a tiny file (`checkpoint.offset`) that tracks how
+//! much of the WAL has been durably consumed. It survives crashes so restarts
+//! can resume from the last known safe offset without rescanning the entire log.
+//!
+//! # Format (24 bytes)
+//!
+//! ```text
+//! ┌────────────┬─────────┬──────────┬─────────────────────┬──────────┐
+//! │ magic (8)  │ ver (2) │ rsv (2)  │ global_data_off (8) │ crc (4)  │
+//! └────────────┴─────────┴──────────┴─────────────────────┴──────────┘
+//! ```
+//!
+//! Writes use atomic rename (`write_to` → `.tmp` → `rename`) plus parent
+//! directory fsync to ensure durability across power loss.
 
 use std::fs::OpenOptions;
 use std::io::{Read, Write};

rust/otap-dataflow/crates/quiver/src/wal/header.rs

@@ -1,6 +1,21 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
+//! WAL file header encoding and validation.
+//!
+//! Every WAL file starts with a fixed 30-byte header:
+//!
+//! ```text
+//! ┌────────────┬─────────┬──────────┬──────────────────┐
+//! │ magic (10) │ ver (2) │ rsv (2)  │ segment_hash (16)│
+//! └────────────┴─────────┴──────────┴──────────────────┘
+//! ```
+//!
+//! - **magic**: `b"QUIVER\0WAL"` identifies the file type
+//! - **version**: Format version (currently 1)
+//! - **reserved**: Zero padding for future use
+//! - **segment_hash**: MD5 of segment configuration; mismatches reject the file
+
 use std::io::{Read, Seek, SeekFrom, Write};
 
 use super::{WAL_MAGIC, WalError};

rust/otap-dataflow/crates/quiver/src/wal/mod.rs

@@ -1,6 +1,62 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
+//! Write-Ahead Log (WAL) for Quiver crash recovery.
+//!
+//! This module provides durable, append-only storage for Arrow record batches.
+//! On crash, the WAL replays uncommitted data to restore in-memory state.
+//!
+//! # Quick Start
+//!
+//! ```ignore
+//! // Writing
+//! let mut writer = WalWriter::open(options)?;
+//! let offset = writer.append_bundle(&bundle)?;
+//!
+//! // Reading (for replay)
+//! let mut reader = WalReader::open(&path)?;
+//! let mut cursor = WalConsumerCheckpoint::default();
+//! for entry in reader.iter_from(0)? {
+//!     let bundle = entry?;
+//!     // ... rebuild state from bundle ...
+//!     cursor.increment(&bundle);  // in-memory only
+//! }
+//!
+//! // Checkpointing (after downstream confirms durability)
+//! writer.checkpoint_cursor(&cursor)?;  // persists + enables cleanup
+//! ```
+//!
+//! # Module Organization
+//!
+//! | File                    | Purpose                                          |
+//! |-------------------------|--------------------------------------------------|
+//! | `writer.rs`             | Append entries, rotate files, manage checkpoints |
+//! | `reader.rs`             | Iterate entries, decode payloads, track progress |
+//! | `header.rs`             | WAL file header format (magic, version, config)  |
+//! | `checkpoint_sidecar.rs` | Crash-safe checkpoint offset persistence         |
+//! | `tests.rs`              | Integration tests and crash simulation           |
+//!
+//! # On-Disk Layout
+//!
+//! ```text
+//! wal/
+//! ├── quiver.wal           # Active WAL file (append target)
+//! ├── quiver.wal.1         # Rotated file (oldest)
+//! ├── quiver.wal.2         # Rotated file
+//! └── checkpoint.offset    # Consumer progress (24 bytes, CRC-protected)
+//! ```
+//!
+//! # Key Concepts
+//!
+//! - **Entry**: One [`RecordBundle`] serialized with CRC32 integrity check
+//! - **Rotation**: When the active file exceeds `rotation_target_bytes`, it's
+//!   renamed to `quiver.wal.N` and a fresh file starts
+//! - **Checkpoint**: Consumers call [`WalConsumerCheckpoint::increment()`] while
+//!   iterating (in-memory), then [`WalWriter::checkpoint_cursor()`] to persist
+//! - **Purge**: Rotated files are deleted once fully covered by the checkpoint
+//!
+//! See [`writer`] module docs for detailed lifecycle documentation.
+
 use std::io;
 
 use arrow_schema::ArrowError;
@@ -31,6 +87,9 @@ pub(crate) const SLOT_HEADER_LEN: usize = 2 + SCHEMA_FINGERPRINT_LEN + 4 + 4;
 pub(crate) type WalResult<T> = Result<T, WalError>;
 
 /// Errors produced while reading or writing WAL data.
+///
+/// Most variants include context about where the failure occurred.
+/// [`WalError::Io`] wraps underlying filesystem errors.
 #[derive(Error, Debug)]
 pub enum WalError {
     /// Underlying filesystem failure.

rust/otap-dataflow/crates/quiver/src/wal/reader.rs

@@ -4,9 +4,38 @@
 //! Read-side companion to the WAL writer.
 //!
 //! The reader validates headers, streams entries starting at arbitrary offsets,
-//! and exposes helper types such as [`WalConsumerCheckpoint`] so higher layers can
-//! describe how much of the log is safe to reclaim. Like the writer, it is
-//! currently exercised by tests until the runtime wires in replay logic.
+//! and exposes helper types for tracking replay progress.
+//!
+//! # Entry Format
+//!
+//! Each WAL entry has this layout:
+//!
+//! ```text
+//! ┌──────────┬────────────────┬─────────────────┬──────────┐
+//! │ len (4)  │ entry_hdr (25) │ slot_data (var) │ crc (4)  │
+//! └──────────┴────────────────┴─────────────────┴──────────┘
+//! ```
+//!
+//! - **len**: Size of `entry_hdr + slot_data` (excludes len and crc fields)
+//! - **entry_hdr**: Type (1), timestamp (8), sequence (8), slot_bitmap (8)
+//! - **slot_data**: For each set bit: slot_id (2), fingerprint (32), rows (4),
+//!   payload_len (4), Arrow IPC bytes (payload_len)
+//! - **crc**: CRC32 over `entry_hdr + slot_data`
+//!
+//! # Usage
+//!
+//! ```ignore
+//! let mut reader = WalReader::open("wal/quiver.wal")?;
+//! let mut cursor = WalConsumerCheckpoint::default();
+//!
+//! for result in reader.iter_from(0)? {
+//!     let bundle = result?;
+//!     println!("seq={} slots={}", bundle.sequence, bundle.slots.len());
+//!     cursor.increment(&bundle);  // in-memory only
+//! }
+//! // cursor now points past the last entry
+//! // call writer.checkpoint_cursor(&cursor) to persist
+//! ```
 #![allow(dead_code)]
 
 #[cfg(test)]
@@ -188,12 +217,16 @@ pub(crate) struct DecodedWalSlot {
     pub payload: Vec<u8>,
 }
 
-/// Opaque cursor describing how much of the WAL has been durably processed.
+/// Opaque cursor describing how much of the WAL has been seen.
 ///
-/// Operators should not interpret the internal fields directly. Instead:
+/// This is an **in-memory only** cursor. Updating it has no durability effect.
+/// To persist progress and allow WAL cleanup, pass the cursor to
+/// [`WalWriter::checkpoint_cursor()`].
+///
+/// Usage:
 /// 1. Start with `WalConsumerCheckpoint::default()` (beginning of WAL)
-/// 2. Call `cursor.advance(&bundle)` after processing each entry
-/// 3. Pass the cursor to `advance_consumer_checkpoint()` to persist progress
+/// 2. Call `cursor.increment(&bundle)` after processing each entry (in-memory)
+/// 3. Call `writer.checkpoint_cursor(&cursor)` to persist and enable cleanup
 ///
 /// The writer validates that checkpoints land on entry boundaries and rejects
 /// stale or regressed values.
@@ -209,7 +242,7 @@ impl WalConsumerCheckpoint {
     /// Creates a checkpoint positioned immediately after the given bundle.
     ///
     /// This is equivalent to `WalConsumerCheckpoint::default()` followed by
-    /// `advance(bundle)`, but clearer when you only need to checkpoint a
+    /// `increment(bundle)`, but clearer when you only need to checkpoint a
     /// single entry.
     pub fn after(bundle: &WalRecordBundle) -> Self {
         Self {
@@ -218,18 +251,21 @@ impl WalConsumerCheckpoint {
         }
     }
 
-    /// Advances the cursor to cover the provided bundle.
+    /// Moves the cursor past the provided bundle (in-memory only).
+    ///
+    /// This does **not** persist the checkpoint or trigger WAL cleanup.
+    /// Call [`WalWriter::checkpoint_cursor()`] to make progress durable.
     ///
     /// Typical usage in a replay loop:
     /// ```ignore
     /// let mut cursor = WalConsumerCheckpoint::default();
     /// for bundle in reader.iter_from(0)? {
     ///     process(&bundle?);
-    ///     cursor.advance(&bundle);
+    ///     cursor.increment(&bundle);  // in-memory only
     /// }
-    /// writer.advance_consumer_checkpoint(&cursor)?;
+    /// writer.checkpoint_cursor(&cursor)?;  // persist + cleanup
     /// ```
-    pub fn advance(&mut self, bundle: &WalRecordBundle) {
+    pub fn increment(&mut self, bundle: &WalRecordBundle) {
         self.safe_offset = bundle.next_offset;
         self.safe_sequence = bundle.sequence;
     }

rust/otap-dataflow/crates/quiver/src/wal/tests.rs

@@ -542,9 +542,7 @@ fn wal_writer_records_cursor_without_truncating() {
         safe_offset: first_entry.next_offset,
         ..WalConsumerCheckpoint::default()
     };
-    writer
-        .advance_consumer_checkpoint(&cursor)
-        .expect("record cursor");
+    writer.checkpoint_cursor(&cursor).expect("record cursor");
     drop(writer);
 
     let len_after = std::fs::metadata(&wal_path).expect("metadata").len();
@@ -602,16 +600,16 @@ fn wal_writer_enforces_safe_offset_boundaries() {
         safe_sequence: first_entry.sequence,
     };
 
-    match writer.advance_consumer_checkpoint(&cursor) {
+    match writer.checkpoint_cursor(&cursor) {
         Err(WalError::InvalidConsumerCheckpoint(message)) => {
             assert_eq!(message, "safe offset splits entry boundary")
         }
         other => panic!("expected invalid cursor error, got {other:?}"),
     }
 
-    cursor.advance(&first_entry);
+    cursor.increment(&first_entry);
     writer
-        .advance_consumer_checkpoint(&cursor)
+        .checkpoint_cursor(&cursor)
         .expect("record succeeds with aligned cursor");
     drop(writer);
 
@@ -639,9 +637,7 @@ fn wal_writer_persists_consumer_checkpoint_sidecar() {
         safe_offset: file_len,
         ..WalConsumerCheckpoint::default()
     };
-    writer
-        .advance_consumer_checkpoint(&cursor)
-        .expect("record cursor");
+    writer.checkpoint_cursor(&cursor).expect("record cursor");
     drop(writer);
 
     let sidecar_path = wal_path.parent().expect("dir").join("checkpoint.offset");
@@ -803,7 +799,7 @@ fn wal_writer_enforces_size_cap_and_purges_rotations() {
         ..WalConsumerCheckpoint::default()
     };
     writer
-        .advance_consumer_checkpoint(&cursor)
+        .checkpoint_cursor(&cursor)
         .expect("record cursor purges rotated chunks");
 
     assert!(
@@ -857,9 +853,7 @@ fn wal_writer_ignores_invalid_checkpoint_sidecar() {
         safe_offset: file_len,
         ..WalConsumerCheckpoint::default()
     };
-    writer
-        .advance_consumer_checkpoint(&cursor)
-        .expect("record cursor");
+    writer.checkpoint_cursor(&cursor).expect("record cursor");
     drop(writer);
 
     let state = CheckpointSidecar::read_from(&sidecar_path).expect("sidecar");
@@ -1407,7 +1401,7 @@ fn wal_reader_iter_from_respects_offsets() {
     assert_eq!(entry_one.next_offset, entry_two.offset.position);
 
     let mut cursor = WalConsumerCheckpoint::default();
-    cursor.advance(&entry_one);
+    cursor.increment(&entry_one);
     assert_eq!(cursor.safe_offset, entry_one.next_offset);
     assert_eq!(cursor.safe_sequence, entry_one.sequence);
 
@@ -1607,7 +1601,7 @@ fn wal_consumer_checkpoint_recovers_after_partial_entry() {
     let mut iter = reader.iter_from(0).expect("iterator");
     let first_entry = iter.next().expect("first entry").expect("ok");
     let mut cursor = WalConsumerCheckpoint::default();
-    cursor.advance(&first_entry);
+    cursor.increment(&first_entry);
     drop(reader);
 
     // Simulate a crash that truncates the file in the middle of the second
@@ -1811,7 +1805,7 @@ fn run_crash_case(case: CrashCase) {
         case.name
     );
     writer_test_support::inject_crash(case.injection);
-    let err = match writer.advance_consumer_checkpoint(&cursor) {
+    let err = match writer.checkpoint_cursor(&cursor) {
         Ok(_) => panic!("{}: crash injection did not trigger", case.name),
         Err(err) => err,
     };
@@ -1841,7 +1835,7 @@ fn wal_cursor_after_entries(path: &Path, entry_count: usize) -> WalConsumerCheck
                 )
             })
             .expect("entry ok while building cursor");
-        cursor.advance(&bundle);
+        cursor.increment(&bundle);
     }
     cursor
 }
@@ -2061,7 +2055,7 @@ fn wal_writer_rejects_checkpoint_sequence_regression() {
         safe_sequence: entries[1].sequence,
     };
     writer
-        .advance_consumer_checkpoint(&cursor_at_second)
+        .checkpoint_cursor(&cursor_at_second)
         .expect("advance to second entry");
 
     // Now try to regress to the first entry (sequence=0)
@@ -2070,7 +2064,7 @@ fn wal_writer_rejects_checkpoint_sequence_regression() {
         safe_sequence: entries[0].sequence, // sequence=0, which is less than 1
     };
 
-    match writer.advance_consumer_checkpoint(&cursor_at_first) {
+    match writer.checkpoint_cursor(&cursor_at_first) {
         Err(WalError::InvalidConsumerCheckpoint(msg)) => {
             assert!(
                 msg.contains("regressed"),
@@ -2105,7 +2099,7 @@ fn wal_writer_rejects_checkpoint_offset_regression() {
         safe_sequence: entries[1].sequence,
     };
     writer
-        .advance_consumer_checkpoint(&cursor_at_second)
+        .checkpoint_cursor(&cursor_at_second)
         .expect("advance to second entry");
 
     // Now try to advance with a higher sequence but lower offset
@@ -2115,7 +2109,7 @@ fn wal_writer_rejects_checkpoint_offset_regression() {
         safe_sequence: entries[1].sequence + 1, // higher sequence to pass that check
     };
 
-    match writer.advance_consumer_checkpoint(&bad_cursor) {
+    match writer.checkpoint_cursor(&bad_cursor) {
         Err(WalError::InvalidConsumerCheckpoint(msg)) => {
             assert!(
                 msg.contains("regressed"),
@@ -2218,7 +2212,7 @@ fn wal_recovery_clamps_stale_sidecar_offset() {
         safe_sequence: offset.sequence,
     };
     writer
-        .advance_consumer_checkpoint(&cursor)
+        .checkpoint_cursor(&cursor)
         .expect("advance checkpoint to end of WAL");
     drop(writer);