### WriteAhead::with_options and start method

Source: https://docs.rs/writeahead/latest/src/writeahead/write_ahead.rs.html

Constructor with_options initializes the struct with given options. The start method creates the log directory, reads directory entries, and begins recovery; it validates files, heals torn tails, and applies startup retention.

```rust
impl<F: FileIo + 'static> WriteAhead<F> {
    pub fn with_options(options: WriteAheadOptions) -> Self {
        Self {
            options,
            readers: RwLock::new(BTreeMap::new()),
            writer_tx: None,
            writer_events: None,
        }
    }

    /// Starts the write ahead log manager, recovering existing state:
    ///
    /// - Every file is validated (magic number, format version).
    /// - Non-active files that missed their seal (crash mid-rotation) are
    ///   healed: torn tails truncated, then sealed.
    /// - The active (highest-id) file gets its torn tail truncated; if it
    ///   was already sealed, a new active file is created instead.
    /// - Files that don't look like log files (`<digits>.log`) are skipped.
    /// - Startup retention is applied before this returns.
    pub fn start(&mut self) -> Result<()> {
        std::fs::create_dir_all(&self.options.log_dir).context("Failed to create log directory")?;

        let mut found: Vec<(u64, PathBuf)> = Vec::new();
        let entries =
            std::fs::read_dir(&self.options.log_dir).context("Failed to read log directory")?;
```

--------------------------------

### Test read before start errors

Source: https://docs.rs/writeahead/latest/src/writeahead/write_ahead.rs.html

Confirms that read, stream creation, and writer acquisition fail if the WAL has not been started.

```rust
#[test]
    fn test_read_before_start_errors() {
        let dir = tempfile::tempdir().unwrap();
        let wal = test_wal(dir.path());
        assert!(wal.read(0, FILE_HEADER_SIZE).is_err());
        assert!(wal.create_stream().is_err());
        assert!(wal.writer().is_err());
    }
```

--------------------------------

### WriteAhead::start

Source: https://docs.rs/writeahead/latest/writeahead/write_ahead/struct.WriteAhead.html

Starts the write ahead log manager, recovering existing state. Validates files, heals torn tails, and applies startup retention.

```APIDOC
## WriteAhead::start

### Description
Starts the write ahead log manager, recovering existing state. Validates files, heals torn tails, and applies startup retention.

### Method
Method

### Signature
`pub fn start(&mut self) -> Result<()>`

### Parameters
None

### Returns
- **Result<()>** - Ok if the manager started successfully, or an error.

```

--------------------------------

### Test create stream from offset

Source: https://docs.rs/writeahead/latest/src/writeahead/write_ahead.rs.html

Demonstrates creating a stream starting from a specific file ID and offset.

```rust
#[tokio::test]
    async fn test_create_stream_from_offset() {
        let dir = tempfile::tempdir().unwrap();
        let mut wal = test_wal(dir.path());
        wal.start().unwrap();

        let ids = wal
            .write_batch(vec![b"one".to_vec(), b"two".to_vec(), b"three".to_vec()])
            .await
            .unwrap();

        let mut stream = wal
            .create_stream_from(ids[1].file_id, ids[1].file_offset)
            .unwrap();
        let (id, record) = stream.next().await.unwrap().unwrap();
        assert_eq!((id, record.as_slice()), (ids[1], b"two".as_slice()));
        assert_eq!(stream.next().await.unwrap().unwrap().1, b"three");
        assert!(stream.next().await.is_none());
    }
```

--------------------------------

### LogFileStream::new

Source: https://docs.rs/writeahead/latest/writeahead/logfile/struct.LogFileStream.html

Creates a new LogFileStream from a Logfile instance, starting at the beginning of the log.

```APIDOC
## LogFileStream::new

### Description
Creates a new LogFileStream from a Logfile instance, starting at the beginning of the log.

### Method
`pub fn new(logfile: Logfile<F>) -> Self`

### Parameters
- `logfile` (Logfile<F>) - Required - The log file to stream records from.

### Returns
- `Self` - A new LogFileStream instance.

```

--------------------------------

### LogFileStream::new_with_offset

Source: https://docs.rs/writeahead/latest/writeahead/logfile/struct.LogFileStream.html

Creates a new LogFileStream from a Logfile instance, starting at a specified offset.

```APIDOC
## LogFileStream::new_with_offset

### Description
Creates a new LogFileStream from a Logfile instance, starting at a specified offset.

### Method
`pub fn new_with_offset(logfile: Logfile<F>, offset: u64) -> Self`

### Parameters
- `logfile` (Logfile<F>) - Required - The log file to stream records from.
- `offset` (u64) - Required - The byte offset to start streaming from.

### Returns
- `Self` - A new LogFileStream instance.

```

--------------------------------

### WriteAhead::create_stream_from

Source: https://docs.rs/writeahead/latest/writeahead/write_ahead/struct.WriteAhead.html

Creates a stream starting at a record offset in a specific log file. The stream gets its own file handles, so it can be sent to another thread and consumed while writes continue.

```APIDOC
## WriteAhead::create_stream_from

### Description
Creates a stream starting at a record offset in a specific log file. The stream gets its own file handles, so it can be sent to another thread and consumed while writes continue.

### Method
Method

### Signature
`pub fn create_stream_from(&self, logfile_id: u64, offset: u64) -> Result<WriteAheadStream<F>>`

### Parameters
- **logfile_id** (u64) - Required - The ID of the log file to start from.
- **offset** (u64) - Required - The byte offset within the file to start from.

### Returns
- **Result<WriteAheadStream<F>>** - A stream of records starting at the specified location.

```

--------------------------------

### WriteAheadOptions and Default implementation

Source: https://docs.rs/writeahead/latest/src/writeahead/write_ahead.rs.html

Configuration options for the WAL: log directory, max file size, preallocation chunk size, and retention. Defaults set log_dir to './write_ahead', max_file_size to 1GB, preallocation to 64MiB, and retention to default.

```rust
#[derive(Debug, Clone)]
pub struct WriteAheadOptions {
    pub log_dir: PathBuf,
    /// Soft limit for the logical record region. The writer rotates before a
    /// commit group would cross it. A group larger than an empty file's
    /// capacity is kept intact in one oversized file, then rotated. The limit
    /// includes the format header; values below FILE_HEADER_SIZE therefore
    /// treat every non-empty group as oversized without truncating the header.
    pub max_file_size: u64,
    /// Sparse allocation-window size for active files. Ahead-of-time growth
    /// avoids persisting an EOF change on every fdatasync while committed-end
    /// metadata keeps the unused tail invisible to readers and recovery.
    /// None disables preallocation; Some(0) is also treated as disabled.
    pub preallocation_chunk_size: Option<u64>,
    pub retention: RetentionOptions,
}

impl Default for WriteAheadOptions {
    fn default() -> Self {
        Self {
            log_dir: PathBuf::from("./write_ahead"),
            max_file_size: 1024 * 1024 * 1024,                // 1GB
            preallocation_chunk_size: Some(64 * 1024 * 1024), // 64MiB
            retention: RetentionOptions::default(),
        }
    }
}
```

--------------------------------

### take

Source: https://docs.rs/writeahead/latest/writeahead/write_ahead/struct.WriteAheadStream.html

Creates a new stream of at most n items of the underlying stream.

```APIDOC
## take

### Description
Creates a new stream of at most n items of the underlying stream.

### Method
take

### Signature
fn take(self, n: usize) -> Take<Self>
where Self: Sized

### Parameters
- **n** (usize) - Required - The maximum number of items to take.

### Returns
Take<Self> - A stream that yields at most n items.
```

--------------------------------

### Get current time in milliseconds

Source: https://docs.rs/writeahead/latest/src/writeahead/logfile.rs.html

Calculates the current Unix time in milliseconds, used for seal footers.

```rust
pub fn now_ms() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .expect("system clock before unix epoch")
        .as_millis() as u64
}
```

--------------------------------

### Writer Launch and Reader Cache Population

Source: https://docs.rs/writeahead/latest/src/writeahead/write_ahead.rs.html

Initializes the WalWriter and populates the reader cache with surviving log files.

```rust
// Launch the writer: creates/initializes the active file and applies
        // startup retention synchronously before returning.
        let (tx, events) =
            WalWriter::<F>::launch(self.options.clone(), active_id, active_path, registry)?;
        self.writer_tx = Some(tx);
        self.writer_events = Some(events);

        // Populate the reader cache from what survived recovery + retention
        let mut cache = BTreeMap::new();
        for entry in std::fs::read_dir(&self.options.log_dir)? {
            let path = entry?.path();
            if file_id_from_path(&path).is_some() {
                let logfile: Logfile<F> = Logfile::open(&path)?;
                cache.insert(logfile.id, Arc::new(logfile));
            }
        }
        *self.readers.write().expect("reader cache poisoned") = cache;

        Ok(())
```

--------------------------------

### WriterCommand and Data Structures

Source: https://docs.rs/writeahead/latest/src/writeahead/writer.rs.html

Defines the commands for writing and trimming, along with structures for tracking trim statistics, write acknowledgments, and file metadata.

```rust
pub(crate) enum WriterCommand {
    Write(flume::Sender<Result<WriteAck>>, Vec<Vec<u8>>),
    Trim(flume::Sender<TrimStats>, u64),
}

/// What an explicit [WriteHandle::trim_before] call deleted.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct TrimStats {
    pub files_deleted: u64,
    pub bytes_reclaimed: u64,
}

#[derive(Debug)]
pub(crate) struct WriteAck {
    /// The file the batch landed in (rotation can happen between submission
    /// and commit, so this is part of the ack).
    pub file_id: u64,
    pub offsets: Vec<u64>,
}

/// Lifecycle notifications for the read side's file cache.
#[derive(Debug, Clone)]
pub(crate) enum WriterEvent {
    Created(u64, PathBuf),
    Sealed(u64),
    Deleted(u64),
}

#[derive(Debug, Clone)]
pub(crate) struct FileMeta {
    pub path: PathBuf,
    pub size: u64,
    pub seal_timestamp_ms: Option<u64>,
}
```

--------------------------------

### Scan records for end offset

Source: https://docs.rs/writeahead/latest/src/writeahead/logfile.rs.html

Iterates through records from the start of the file to find the end offset of the longest valid prefix.

```rust
fn scan_records_end<F: FileIo>(fio: &F, scan_limit: u64) -> u64 {
    let mut offset = FILE_HEADER_SIZE;
    loop {
        match read_record_at(fio, offset, scan_limit) {
            Ok((_, next)) => offset = next,
            Err(_) => return offset,
        }
    }
}
```

--------------------------------

### SimpleFile implementation of FileIo trait

Source: https://docs.rs/writeahead/latest/src/writeahead/fileio/simple_file.rs.html

Defines the SimpleFile struct and implements the FileIo trait with methods for opening files (create or existing), positional read/write, sync (fdatasync), and getting/setting file length. Uses anyhow for error context.

```rust
use std::fs::{File, OpenOptions};
use std::os::unix::fs::FileExt;
use std::path::Path;

use anyhow::{Context, Result};

use super::FileIo;

/// Buffered-by-the-page-cache file IO using positional reads and writes.
///
/// For WAL access patterns (readers mostly reading recently written data)
/// the page cache consistently beats direct-IO approaches like io_uring
/// with O_DIRECT, so this is the default and recommended implementation.
#[derive(Debug)]
pub struct SimpleFile {
    fd: File,
}

impl FileIo for SimpleFile {
    fn open(path: &Path) -> Result<Self> {
        let fd = OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .truncate(false)
            .open(path)
            .with_context(|| format!("Failed to open {}", path.display()))?;
        Ok(Self { fd })
    }

    fn open_existing(path: &Path) -> Result<Self> {
        let fd = OpenOptions::new()
            .read(true)
            .write(true)
            .open(path)
            .with_context(|| format!("Failed to open existing {}", path.display()))?;
        Ok(Self { fd })
    }

    fn read_at(&self, offset: u64, buf: &mut [u8]) -> Result<()> {
        self.fd
            .read_exact_at(buf, offset)
            .with_context(|| format!("Failed to read {} bytes at offset {}", buf.len(), offset))?;
        Ok(())
    }

    fn write_at(&mut self, offset: u64, data: &[u8]) -> Result<()> {
        self.fd.write_all_at(data, offset).with_context(|| {
            format!("Failed to write {} bytes at offset {}", data.len(), offset)
        })?;
        Ok(())
    }

    fn sync(&mut self) -> Result<()> {
        // fdatasync: flushes data and the metadata needed to read it back
        // (including file size), skipping timestamp-only metadata updates.
        self.fd.sync_data().context("Failed to fdatasync")?;
        Ok(())
    }

    fn len(&self) -> Result<u64> {
        Ok(self.fd.metadata().context("Failed to stat file")?.len())
    }

    fn set_len(&mut self, len: u64) -> Result<()> {
        self.fd
            .set_len(len)
            .with_context(|| format!("Failed to resize file to {}", len))?;
        Ok(())
    }
}
```

--------------------------------

### boxed

Source: https://docs.rs/writeahead/latest/writeahead/write_ahead/struct.WriteAheadStream.html

Wrap the stream in a Box, pinning it.

```APIDOC
## boxed

### Description
Wrap the stream in a Box, pinning it.

### Method
boxed

### Signature
fn boxed<'a>(self) -> Pin<Box<dyn Stream<Item = Self::Item> + Send + 'a>>
where Self: Sized + Send + 'a

### Returns
Pin<Box<dyn Stream<Item = Self::Item> + Send + 'a>> - A boxed and pinned stream.
```

--------------------------------

### zip

Source: https://docs.rs/writeahead/latest/writeahead/write_ahead/struct.WriteAheadStream.html

An adapter for zipping two streams together.

```APIDOC
## zip

### Description
An adapter for zipping two streams together.

### Method
zip

### Signature
fn zip<St>(self, other: St) -> Zip<Self, St>
where St: Stream, Self: Sized

### Parameters
- **other** (St) - Required - The stream to zip with.

### Returns
Zip<Self, St> - A stream that yields pairs of items from both streams.
```

--------------------------------

### WriteAhead::with_options

Source: https://docs.rs/writeahead/latest/writeahead/write_ahead/struct.WriteAhead.html

Creates a new WriteAhead instance with the given options.

```APIDOC
## WriteAhead::with_options

### Description
Creates a new WriteAhead instance with the given options.

### Method
Constructor

### Signature
`pub fn with_options(options: WriteAheadOptions) -> Self`

### Parameters
- **options** (WriteAheadOptions) - Required - Configuration options for the write-ahead log.

### Returns
- **Self** - A new WriteAhead instance.

```

--------------------------------

### Log File Recovery and Registry Construction

Source: https://docs.rs/writeahead/latest/src/writeahead/write_ahead.rs.html

Builds the file registry by processing existing log files, healing unsealed files, and determining the active log file for the writer.

```rust
// Recover every file and build the writer's file registry
        let mut registry: BTreeMap<u64, crate::writer::FileMeta> = BTreeMap::new();
        let (active_id, active_path) = match found.last().cloned() {
            None => {
                debug!("creating initial log file");
                (0, log_file_path(&self.options.log_dir, 0))
            }
            Some((last_id, last_path)) => {
                for (id, path) in found.iter().filter(|(id, _)| *id != last_id) {
                    debug!("loading existing logfile {}", id);
                    let logfile: Logfile<F> = Logfile::open(path)?;
                    let seal_ts = if logfile.sealed {
                        logfile.seal_timestamp_ms
                    } else {
                        // Crash between sealing and creating the next file,
                        // or the seal itself was torn: heal it now.
                        warn!("healing unsealed non-active logfile {}", id);
                        drop(logfile);
                        let ts = now_ms();
                        recover_and_seal::<F>(path, ts)?;
                        Some(ts)
                    };
                    registry.insert(
                        *id,
                        crate::writer::FileMeta {
                            path: path.clone(),
                            size: std::fs::metadata(path)?.len(),
                            seal_timestamp_ms: seal_ts,
                        },
                    );
                }

                let last_len = std::fs::metadata(&last_path)
                    .with_context(|| format!("Failed to stat {}", last_path.display()))?
                    .len();
                if last_len <= FILE_HEADER_SIZE {
                    // Crash between file creation and a durable header, or a
                    // valid empty active file.
                    debug!("recovering empty/header-only active logfile {}", last_id);
                    recover_unsealed::<F>(&last_path)?;
                    (last_id, last_path)
                } else {
                    let logfile: Logfile<F> = Logfile::open(&last_path)?;
                    if logfile.sealed {
                        // Crash after seal but before the next file existed
                        debug!("last logfile {} is sealed, rotating", last_id);
                        registry.insert(
                            last_id,
                            crate::writer::FileMeta {
                                path: last_path.clone(),
                                size: last_len,
                                seal_timestamp_ms: logfile.seal_timestamp_ms,
                            },
                        );
                        let next_id = last_id
                            .checked_add(1)
                            .ok_or(WriteAheadError::FileIdExhausted)?;
                        (next_id, log_file_path(&self.options.log_dir, next_id))
                    } else {
                        drop(logfile);
                        recover_unsealed::<F>(&last_path)?;
                        (last_id, last_path)
                    }
                }
            }
        };
```

--------------------------------

### Build and parse file footer

Source: https://docs.rs/writeahead/latest/src/writeahead/logfile.rs.html

Functions to construct a footer with a magic number and timestamp, and to validate an existing footer.

```rust
fn build_footer(records_end: u64, ts_ms: u64) -> [u8; FOOTER_SIZE as usize] {
    let mut f = [0u8; FOOTER_SIZE as usize];
    f[0..8].copy_from_slice(&MAGIC_NUMBER);
    f[8..16].copy_from_slice(&ts_ms.to_le_bytes());
    f[16..24].copy_from_slice(&records_end.to_le_bytes());
    let (h1, h2) = murmur3_128(&f[0..24]);
    f[24..32].copy_from_slice(&h1.to_le_bytes());
    f[32..40].copy_from_slice(&h2.to_le_bytes());
    f
}

fn parse_footer(buf: &[u8; FOOTER_SIZE as usize], file_len: u64) -> Option<u64> {
    if buf[0..8] != MAGIC_NUMBER {
        return None;
    }
    let (h1, h2) = murmur3_128(&buf[0..24]);
    if buf[24..32] != h1.to_le_bytes() || buf[32..40] != h2.to_le_bytes() {
        return None;
    }
    let records_end = u64::from_le_bytes(buf[16..24].try_into().unwrap());
```

--------------------------------

### now_ms()

Source: https://docs.rs/writeahead/latest/writeahead/logfile/fn.now_ms.html

Returns the current Unix time in milliseconds. This is the timestamp format used in seal footers.

```APIDOC
## Function: now_ms

### Description
Returns the current Unix time in milliseconds. This is the timestamp format used in seal footers.

### Signature
```rust
pub fn now_ms() -> u64
```

### Parameters
None.

### Returns
- **u64** - The current Unix time in milliseconds.

### Example
```rust
let timestamp = now_ms();
```
```

--------------------------------

### Test WriteAhead Sync and WriteHandle Clone

Source: https://docs.rs/writeahead/latest/src/writeahead/write_ahead.rs.html

Verifies that WriteAhead is Sync and WriteHandle is Cloneable.

```rust
#[test]
    fn test_write_ahead_is_sync_and_handle_clonable() {
        fn assert_send_sync<T: Send + Sync>() {}
        fn assert_clone<T: Clone>() {}
        assert_send_sync::<WriteAhead<SimpleFile>>();
        assert_send_sync::<WriteHandle>();
        assert_clone::<WriteHandle>();
    }
```

--------------------------------

### boxed_local

Source: https://docs.rs/writeahead/latest/writeahead/write_ahead/struct.WriteAheadStream.html

Wrap the stream in a Box, pinning it.

```APIDOC
## boxed_local

### Description
Wrap the stream in a Box, pinning it.

### Method
boxed_local

### Signature
fn boxed_local<'a>(self) -> Pin<Box<dyn Stream<Item = Self::Item> + 'a>>
where Self: Sized + 'a

### Returns
Pin<Box<dyn Stream<Item = Self::Item> + 'a>> - A boxed and pinned stream.
```

--------------------------------

### Logfile Constants

Source: https://docs.rs/writeahead/latest/src/writeahead/logfile.rs.html

Configuration constants defining the file format and header sizes.

```rust
pub const MAGIC_NUMBER: [u8; 8] = [0xff; 8];
pub const FORMAT_VERSION: u8 = 3;
/// Magic number + version byte, before the committed-end slots.
const FILE_HEADER_PREFIX_SIZE: u64 = 9;
const COMMIT_SLOT_SIZE: u64 = 32;
const COMMIT_SLOT_COUNT: u64 = 2;
const HEADER_BLOCK_SIZE: u64 = 4 * 1024;
/// Current (v3) header: one format block plus one block per commit slot.
pub const FILE_HEADER_SIZE: u64 = (1 + COMMIT_SLOT_COUNT) * HEADER_BLOCK_SIZE;
/// Record hash + length prefix.
pub const RECORD_HEADER_SIZE: u64 = 20;
/// Magic + timestamp + records-end + footer hash.
pub const FOOTER_SIZE: u64 = 40;
```

--------------------------------

### sync_dir function to fsync a directory

Source: https://docs.rs/writeahead/latest/src/writeahead/fileio/mod.rs.html

Fsyncs a directory so newly created or removed entries survive a crash. Opens the directory and calls sync_all, wrapping errors with context that includes the directory path.

```rust
/// fsyncs a directory so newly created/removed entries survive a crash.
pub(crate) fn sync_dir(dir: &Path) -> Result<()> {
    use anyhow::Context;
    std::fs::File::open(dir)
        .and_then(|f| f.sync_all())
        .with_context(|| format!("Failed to fsync directory {}", dir.display()))?;
    Ok(())
}

```

--------------------------------

### Log File Test Helpers

Source: https://docs.rs/writeahead/latest/src/writeahead/logfile.rs.html

Utility functions for writing raw records and sealing log files during testing.

```rust
fn write_raw(path: &Path, records: &[Vec<u8>]) -> Vec<u64> {
        let mut fio = SimpleFile::open(path).unwrap();
        if fio.len().unwrap() == 0 {
            write_header(&mut fio).unwrap();
        }
        let state = read_commit_state(&fio).unwrap();
        let end = state.records_end;
        let mut buf = Vec::new();
        let offsets = encode_records(&mut buf, records, end);
        fio.write_at(end, &buf).unwrap();
        write_committed_end(&mut fio, state, end + buf.len() as u64).unwrap();
        fio.sync().unwrap();
        offsets
    }

    fn seal_raw(path: &Path) {
        let fio = SimpleFile::open(path).unwrap();
        let end = read_commit_state(&fio).unwrap().records_end;
        drop(fio);
        let mut fio = SimpleFile::open(path).unwrap();
        append_footer(&mut fio, end, now_ms()).unwrap();
    }
```

--------------------------------

### CommitState and FileHeader Structs

Source: https://docs.rs/writeahead/latest/src/writeahead/logfile.rs.html

Internal structures for tracking commit state and file headers.

```rust
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct CommitState {
    pub generation: u64,
    pub records_end: u64,
    slot: usize,
}

#[derive(Debug, Clone, Copy)]
struct FileHeader {
    commit: CommitState,
    commit_slots: [Option<CommitState>; COMMIT_SLOT_COUNT as usize],
}
```

--------------------------------

### buffered

Source: https://docs.rs/writeahead/latest/writeahead/write_ahead/struct.WriteAheadStream.html

An adaptor for creating a buffered list of pending futures.

```APIDOC
## buffered

### Description
An adaptor for creating a buffered list of pending futures.

### Method
buffered

### Signature
fn buffered(self, n: usize) -> Buffered<Self>
where Self::Item: Future, Self: Sized

### Parameters
- **n** (usize) - Required - The maximum number of futures to buffer.

### Returns
Buffered<Self> - A stream that yields the results of buffered futures.
```

--------------------------------

### try_ready_chunks

Source: https://docs.rs/writeahead/latest/writeahead/write_ahead/struct.WriteAheadStream.html

An adaptor for chunking up successful, ready items of the stream inside a vector.

```APIDOC
## try_ready_chunks

### Description
An adaptor for chunking up successful, ready items of the stream inside a vector.

### Method
fn try_ready_chunks(self, capacity: usize) -> TryReadyChunks<Self>

### Parameters
- **capacity** (usize) - The maximum size of each chunk.

### Return Value
TryReadyChunks<Self> - A stream that yields chunks of successful, ready items.
```

--------------------------------

### ready_chunks

Source: https://docs.rs/writeahead/latest/writeahead/write_ahead/struct.WriteAheadStream.html

An adaptor for chunking up ready items of the stream inside a vector.

```APIDOC
## ready_chunks

### Description
An adaptor for chunking up ready items of the stream inside a vector.

### Method
ready_chunks

### Signature
fn ready_chunks(self, capacity: usize) -> ReadyChunks<Self>
where Self: Sized

### Parameters
- **capacity** (usize) - Required - The maximum size of each chunk.

### Returns
ReadyChunks<Self> - A stream that yields vectors of ready items.
```

--------------------------------

### StreamExt::inspect

Source: https://docs.rs/writeahead/latest/writeahead/write_ahead/struct.WriteAheadStream.html

Do something with each item of this stream, afterwards passing it on.

```APIDOC
## StreamExt::inspect

### Description
Do something with each item of this stream, afterwards passing it on.

### Method
fn inspect<F>(self, f: F) -> Inspect<Self, F>
where F: FnMut(&Self::Item), Self: Sized

### Parameters
- **f** (F) - Required - A closure that takes a reference to each item and performs an action.

### Returns
- **Inspect<Self, F>** - A stream adapter that applies the closure to each item before passing it on.
```

--------------------------------

### FileIo trait implementation for SimpleFile

Source: https://docs.rs/writeahead/latest/writeahead/fileio/simple_file/struct.SimpleFile.html

The FileIo trait provides file I/O operations for SimpleFile, including opening, reading at a position, writing at a position, synchronizing to disk, and managing file length.

```APIDOC
## Struct SimpleFile

### Description
Buffered-by-the-page-cache file IO using positional reads and writes. For WAL access patterns (readers mostly reading recently written data) the page cache consistently beats direct-IO approaches like io_uring with O_DIRECT.

### Methods

#### open

Opens a file for writing, creating it if missing.

- **Signature**: `fn open(path: &Path) -> Result<Self>`
- **Parameters**:
  - `path` (&Path) - Path to the file to open.
- **Returns**: `Result<Self>` - The opened SimpleFile or an error.

#### open_existing

Opens an existing file; errors if it does not exist. Read and recovery paths use this so they can never create stray files.

- **Signature**: `fn open_existing(path: &Path) -> Result<Self>`
- **Parameters**:
  - `path` (&Path) - Path to an existing file.
- **Returns**: `Result<Self>` - The opened SimpleFile or an error.

#### read_at

Read exactly `buf.len()` bytes at `offset`. Errors if the file is too short.

- **Signature**: `fn read_at(&self, offset: u64, buf: &mut [u8]) -> Result<()>`
- **Parameters**:
  - `offset` (u64) - The byte offset in the file to read from.
  - `buf` (&mut [u8]) - The buffer to read data into.
- **Returns**: `Result<()>` - Success or an error.

#### write_at

Write all of `data` at `offset`. Durability requires a subsequent `sync`.

- **Signature**: `fn write_at(&mut self, offset: u64, data: &[u8]) -> Result<()>`
- **Parameters**:
  - `offset` (u64) - The byte offset in the file to write to.
  - `data` (&[u8]) - The data to write.
- **Returns**: `Result<()>` - Success or an error.

#### sync

Flush written data (and the metadata needed to read it back) to disk.

- **Signature**: `fn sync(&mut self) -> Result<()>`
- **Parameters**: None
- **Returns**: `Result<()>` - Success or an error.

#### len

Returns the current length of the file.

- **Signature**: `fn len(&self) -> Result<u64>`
- **Parameters**: None
- **Returns**: `Result<u64>` - The length of the file or an error.

#### set_len

Changes the physical file length. The writer uses growth as sparse preallocation and shrinkage to remove allocation tails during recovery and sealing.

- **Signature**: `fn set_len(&mut self, len: u64) -> Result<()>`
- **Parameters**:
  - `len` (u64) - The new file length.
- **Returns**: `Result<()>` - Success or an error.
```

--------------------------------

### try_ready_chunks

Source: https://docs.rs/writeahead/latest/writeahead/logfile/struct.LogFileStream.html

Chunks successful, ready items of the stream into vectors, returning an error if any item fails.

```APIDOC
## try_ready_chunks

### Description
An adaptor for chunking up successful, ready items of the stream inside a vector.

### Method
`fn try_ready_chunks(self, capacity: usize) -> TryReadyChunks<Self>`

### Parameters
- `capacity` (usize) - Required - Maximum number of items per chunk.

### Type Constraints
- `Self: Sized`

### Returns
`TryReadyChunks<Self>` - A stream that yields vectors of ready items, or an error if any item fails.
```

--------------------------------

### RecordID::new

Source: https://docs.rs/writeahead/latest/writeahead/record/struct.RecordID.html

Creates a new RecordID instance with the given file_id and file_offset.

```APIDOC
## RecordID::new

### Description
Creates a new RecordID instance with the given file_id and file_offset.

### Method
pub fn new(file_id: u64, file_offset: u64) -> Self

### Parameters
- **file_id** (u64) - Required - The identifier of the log file.
- **file_offset** (u64) - Required - The offset within the log file.

### Returns
- **Self** - A new RecordID instance.
```

--------------------------------

### TryStreamExt::inspect_ok

Source: https://docs.rs/writeahead/latest/writeahead/write_ahead/struct.WriteAheadStream.html

Do something with the success value of this stream, afterwards passing it on.

```APIDOC
## TryStreamExt::inspect_ok

### Description
Do something with the success value of this stream, afterwards passing it on.

### Method
fn inspect_ok<F>(self, f: F) -> InspectOk<Self, F>
where F: FnMut(&Self::Ok), Self: Sized

### Parameters
- **f** (F) - Required - A closure that takes a reference to the success value.

### Returns
- **InspectOk<Self, F>** - A stream that inspects success values.
```

--------------------------------

### chain

Source: https://docs.rs/writeahead/latest/writeahead/write_ahead/struct.WriteAheadStream.html

Adapter for chaining two streams.

```APIDOC
## chain

### Description
Adapter for chaining two streams.

### Method
chain

### Signature
fn chain<St>(self, other: St) -> Chain<Self, St>
where St: Stream<Item = Self::Item>, Self: Sized

### Parameters
- **other** (St) - Required - The stream to chain after this one.

### Returns
Chain<Self, St> - A stream that yields items from the first stream then the second.
```

--------------------------------

### ShrinkRaceFile Test Implementation

Source: https://docs.rs/writeahead/latest/src/writeahead/logfile.rs.html

A mock FileIo implementation used to simulate race conditions where a file might shrink during a read operation.

```rust
#[derive(Debug)]
    struct ShrinkRaceFile {
        fd: std::fs::File,
    }

    impl FileIo for ShrinkRaceFile {
        fn open(path: &Path) -> Result<Self> {
            let fd = std::fs::OpenOptions::new()
                .read(true)
                .write(true)
                .create(true)
                .truncate(false)
                .open(path)?;
            Ok(Self { fd })
        }

        fn open_existing(path: &Path) -> Result<Self> {
            let fd = std::fs::OpenOptions::new()
                .read(true)
                .write(true)
                .open(path)?;
            Ok(Self { fd })
        }

        fn read_at(&self, offset: u64, buf: &mut [u8]) -> Result<()> {
            if offset == SHRINK_OFFSET.load(Ordering::SeqCst)
                && SHRINK_ON_FOOTER_READ.swap(false, Ordering::SeqCst)
            {
                self.fd.set_len(offset)?;
            }
            std::os::unix::fs::FileExt::read_exact_at(&self.fd, buf, offset)?;
            Ok(())
        }

        fn write_at(&mut self, offset: u64, data: &[u8]) -> Result<()> {
            std::os::unix::fs::FileExt::write_all_at(&self.fd, data, offset)?;
            Ok(())
        }

        fn sync(&mut self) -> Result<()> {
            self.fd.sync_data()?;
            Ok(())
        }

        fn len(&self) -> Result<u64> {
            Ok(self.fd.metadata()?.len())
        }

        fn set_len(&mut self, len: u64) -> Result<()> {
            self.fd.set_len(len)?;
            Ok(())
        }
    }
```

--------------------------------

### Commit Slot Utilities

Source: https://docs.rs/writeahead/latest/src/writeahead/logfile.rs.html

Functions for calculating offsets, building, and parsing commit slots used to track file generation and record boundaries.

```rust
fn commit_slot_offset(slot: usize) -> u64 {
    (slot as u64 + 1) * HEADER_BLOCK_SIZE
}

fn build_commit_slot(generation: u64, records_end: u64) -> [u8; COMMIT_SLOT_SIZE as usize] {
    let mut slot = [0u8; COMMIT_SLOT_SIZE as usize];
    slot[0..8].copy_from_slice(&generation.to_le_bytes());
    slot[8..16].copy_from_slice(&records_end.to_le_bytes());
    let (h1, h2) = murmur3_128(&slot[0..16]);
    slot[16..24].copy_from_slice(&h1.to_le_bytes());
    slot[24..32].copy_from_slice(&h2.to_le_bytes());
    slot
}

fn parse_commit_slot(buf: &[u8], slot: usize) -> Option<CommitState> {
    let (h1, h2) = murmur3_128(&buf[0..16]);
    if buf[16..24] != h1.to_le_bytes() || buf[24..32] != h2.to_le_bytes() {
        return None;
    }
    Some(CommitState {
        generation: u64::from_le_bytes(buf[0..8].try_into().unwrap()),
        records_end: u64::from_le_bytes(buf[8..16].try_into().unwrap()),
        slot,
    })
}
```

--------------------------------

### simple_file Module

Source: https://docs.rs/writeahead/latest/src/writeahead/fileio/mod.rs.html

A module containing a concrete implementation of the FileIo trait for standard file operations.

```APIDOC
## Module: simple_file

### Description
Contains a standard implementation of the `FileIo` trait, providing thin wrapper around `std::fs::File` for the actual file I/O operations. Not externally documented here.
```

--------------------------------

### test_recovery_does_not_parse_sparse_zeroes_as_empty_records

Source: https://docs.rs/writeahead/latest/src/writeahead/logfile.rs.html

Ensures that recovery correctly handles sparse zero regions without misinterpreting them as empty records.

```rust
#[test]
fn test_recovery_does_not_parse_sparse_zeroes_as_empty_records() {
    let (_dir, path) = temp_log("0000000017.log");
    let mut fio = SimpleFile::open(&path).unwrap();
    write_header(&mut fio).unwrap();
    let state = read_commit_state(&fio).unwrap();
    let sparse_end = FILE_HEADER_SIZE + 2 * RECORD_HEADER_SIZE;
    fio.set_len(sparse_end).unwrap();
    // Simulate a committed-end slot reaching disk while the corresponding
    // record pages remain unwritten sparse zeroes.
    write_committed_end(&mut fio, state, sparse_end).unwrap();
    fio.sync().unwrap();
    drop(fio);

    assert_eq!(
        recover_unsealed::<SimpleFile>(&path).unwrap(),
        FILE_HEADER_SIZE
    );
    assert_eq!(std::fs::metadata(&path).unwrap().len(), FILE_HEADER_SIZE);
}
```

--------------------------------

### peekable

Source: https://docs.rs/writeahead/latest/writeahead/write_ahead/struct.WriteAheadStream.html

Creates a new stream which exposes a peek method.

```APIDOC
## peekable

### Description
Creates a new stream which exposes a peek method.

### Method
peekable

### Signature
fn peekable(self) -> Peekable<Self>
where Self: Sized

### Returns
Peekable<Self> - A stream that allows peeking at the next item without consuming it.
```

--------------------------------

### WriteAhead Integration Tests

Source: https://docs.rs/writeahead/latest/src/writeahead/write_ahead.rs.html

Integration tests for the WriteAhead system, covering basic write/read operations, log rotation, and streaming across multiple files.

```rust
fn test_wal(dir: &Path) -> WriteAhead<SimpleFile> {
    WriteAhead::with_options(WriteAheadOptions {
        log_dir: dir.to_path_buf(),
        ..Default::default()
    })
}

fn test_wal_rotating(dir: &Path, max_file_size: u64) -> WriteAhead<SimpleFile> {
    WriteAhead::with_options(WriteAheadOptions {
        log_dir: dir.to_path_buf(),
        max_file_size: FILE_HEADER_SIZE + max_file_size,
        ..Default::default()
    })
}

#[tokio::test]
async fn test_write_read() {
    let dir = tempfile::tempdir().unwrap();
    let mut wal = test_wal(dir.path());
    wal.start().unwrap();

    let ids = wal
        .write_batch(vec![b"Hello, world!".to_vec()])
        .await
        .unwrap();
    let record = wal.read(ids[0].file_id, ids[0].file_offset).unwrap();
    assert_eq!(record, b"Hello, world!");
}

#[tokio::test]
async fn test_rotation_and_read_back() {
    let dir = tempfile::tempdir().unwrap();
    let mut wal = test_wal_rotating(dir.path(), 128);
    wal.start().unwrap();

    let mut ids = Vec::new();
    for i in 0..10 {
        let batch = wal
            .write_batch(vec![format!("Hello, world! {}", i).into_bytes()])
            .await
            .unwrap();
        ids.extend(batch);
    }
    // Rotation must have happened
    assert!(ids.last().unwrap().file_id > 0);

    for (i, id) in ids.iter().enumerate() {
        let record = wal.read(id.file_id, id.file_offset).unwrap();
        assert_eq!(record, format!("Hello, world! {}", i).into_bytes());
    }
}

#[tokio::test]
async fn test_stream_across_rotations() {
    let dir = tempfile::tempdir().unwrap();
    let mut wal = test_wal_rotating(dir.path(), 128);
    wal.start().unwrap();

    for i in 0..100 {
        wal.write_batch(vec![format!("Hello, world! {}", i).into_bytes()])
            .await
            .unwrap();
    }

    let mut stream = wal.create_stream().unwrap();
    for i in 0..100 {
        let (id, record) = stream.next().await.unwrap().unwrap();
        assert_eq!(record, format!("Hello, world! {}", i).into_bytes());
        assert_eq!(wal.read(id.file_id, id.file_offset).unwrap(), record);
    }
    assert!(stream.next().await.is_none());
}
```

--------------------------------

### buffer_unordered

Source: https://docs.rs/writeahead/latest/writeahead/write_ahead/struct.WriteAheadStream.html

An adaptor for creating a buffered list of pending futures (unordered).

```APIDOC
## buffer_unordered

### Description
An adaptor for creating a buffered list of pending futures (unordered).

### Method
buffer_unordered

### Signature
fn buffer_unordered(self, n: usize) -> BufferUnordered<Self>
where Self::Item: Future, Self: Sized

### Parameters
- **n** (usize) - Required - The maximum number of futures to buffer.

### Returns
BufferUnordered<Self> - A stream that yields the results of buffered futures in any order.
```

--------------------------------

### TryStreamExt::and_then

Source: https://docs.rs/writeahead/latest/writeahead/write_ahead/struct.WriteAheadStream.html

Chain on a computation for when a value is ready, passing the successful results to the provided closure f.

```APIDOC
## TryStreamExt::and_then

### Description
Chain on a computation for when a value is ready, passing the successful results to the provided closure f.

### Method
fn and_then<Fut, F>(self, f: F) -> AndThen<Self, Fut, F>
where F: FnMut(Self::Ok) -> Fut, Fut: TryFuture<Error = Self::Error>, Self: Sized

### Parameters
- **f** (F) - Required - A closure that takes the success value and returns a TryFuture.

### Returns
- **AndThen<Self, Fut, F>** - A stream that chains computations on success.
```

--------------------------------

### WriteHandle Implementation

Source: https://docs.rs/writeahead/latest/src/writeahead/writer.rs.html

Provides a thread-safe handle for submitting durable writes and trimming log files. Writes are coalesced into group commits for efficiency.

```rust
/// Cheaply cloneable handle for submitting durable writes from any task or
/// thread. A write resolves once its records are fsync'd; writes submitted
/// while a commit is in flight coalesce into the next group commit and
/// share its fsync.
///
/// Handles stay valid across log rotations (each ack names the file the
/// records landed in) and keep the writer alive even if the owning
/// [crate::WriteAhead] is dropped.
#[derive(Debug, Clone)]
pub struct WriteHandle {
    pub(crate) tx: flume::Sender<WriterCommand>,
}

impl WriteHandle {
    /// Writes one record durably and returns its address.
    pub async fn write(&self, record: Vec<u8>) -> Result<RecordID> {
        let mut ids = self.write_batch(vec![record]).await?;
        Ok(ids.remove(0))
    }

    /// Writes a batch of records durably (contiguous, all in one file) and
    /// returns their addresses.
    pub async fn write_batch(&self, records: Vec<Vec<u8>>) -> Result<Vec<RecordID>> {
        let (tx, rx) = flume::bounded(1);
        self.tx
            .send(WriterCommand::Write(tx, records))
            .map_err(|_| WriteAheadError::WriterClosed)?;
        let ack = rx
            .recv_async()
            .await
            .map_err(|_| WriteAheadError::WriterClosed)??;
        Ok(ack
            .offsets
            .iter()
            .map(|offset| RecordID::new(ack.file_id, *offset))
            .collect())
    }

    /// Deletes every sealed log file with id strictly below `file_id`,
    /// reclaiming disk space the consumer no longer needs — e.g. once a
    /// replay high-water mark has moved past a rotated file, or after an
    /// explicit compaction rewrote its live records. The active file and
    /// records at or above `file_id` are never touched.
    ///
    /// Resolves after the files are gone; failures to delete individual
    /// files are logged and skipped (retryable). Streams created before the
    /// trim keep their open file handles and finish undisturbed; new reads
    /// into trimmed files return `LogfileNotFound`.
    pub async fn trim_before(&self, file_id: u64) -> Result<TrimStats> {
        let (tx, rx) = flume::bounded(1);
        self.tx
            .send(WriterCommand::Trim(tx, file_id))
            .map_err(|_| WriteAheadError::WriterClosed)?;
        Ok(rx
            .recv_async()
            .await
            .map_err(|_| WriteAheadError::WriterClosed)?)
    }
}
```

--------------------------------

### TryStreamExt::map_ok

Source: https://docs.rs/writeahead/latest/writeahead/write_ahead/struct.WriteAheadStream.html

Wraps the current stream in a new stream which maps the success value using the provided closure.

```APIDOC
## TryStreamExt::map_ok

### Description
Wraps the current stream in a new stream which maps the success value using the provided closure.

### Method
fn map_ok<T, F>(self, f: F) -> MapOk<Self, F>
where Self: Sized, F: FnMut(Self::Ok) -> T

### Parameters
- **f** (F) - Required - A closure that maps the success value to a new type T.

### Returns
- **MapOk<Self, F>** - A stream that maps success values.
```

--------------------------------

### Test Bulk Writes

Source: https://docs.rs/writeahead/latest/src/writeahead/logfile.rs.html

Verifies that multiple records written in bulk can be retrieved individually.

```rust
#[test]
fn test_bulk_writes() {
    let (_dir, path) = temp_log("0000000011.log");
    let records = vec![b"first".to_vec(), b"second".to_vec(), b"third".to_vec()];
    let offsets = write_raw(&path, &records);

    let logfile: Logfile<SimpleFile> = Logfile::open(&path).unwrap();
    assert_eq!(offsets.len(), 3);
    assert_eq!(logfile.read_record(offsets[0]).unwrap(), b"first");
    assert_eq!(logfile.read_record(offsets[1]).unwrap(), b"second");
    assert_eq!(logfile.read_record(offsets[2]).unwrap(), b"third");
}
```

--------------------------------

### by_ref

Source: https://docs.rs/writeahead/latest/writeahead/write_ahead/struct.WriteAheadStream.html

Borrows a stream, rather than consuming it.

```APIDOC
## by_ref

### Description
Borrows a stream, rather than consuming it.

### Method
by_ref

### Signature
fn by_ref(&mut self) -> &mut Self

### Returns
&mut Self - A mutable reference to the stream.
```

--------------------------------

### Test Logfile Sealing

Source: https://docs.rs/writeahead/latest/src/writeahead/logfile.rs.html

Verifies that a logfile can be correctly sealed and that metadata like the seal timestamp is preserved.

```rust
#[test]
fn test_write_with_sealing() {
    let (_dir, path) = temp_log("0000000002.log");
    let offsets = write_raw(&path, &[b"hello".to_vec()]);
    seal_raw(&path);

    let logfile: Logfile<SimpleFile> = Logfile::open(&path).unwrap();
    assert_eq!(logfile.id, 2);
    assert!(logfile.sealed);
    assert!(logfile.seal_timestamp_ms.unwrap() > 0);
    assert_eq!(logfile.read_record(offsets[0]).unwrap(), b"hello");
}
```

--------------------------------

### TryStreamExt::into_stream

Source: https://docs.rs/writeahead/latest/writeahead/write_ahead/struct.WriteAheadStream.html

Wraps a TryStream into a type that implements Stream.

```APIDOC
## TryStreamExt::into_stream

### Description
Wraps a TryStream into a type that implements Stream.

### Method
fn into_stream(self) -> IntoStream<Self>
where Self: Sized

### Returns
- **IntoStream<Self>** - A Stream wrapper around the TryStream.
```