Struct Journal 

pub struct Journal<E: Storage + Metrics, V: Codec> { /* private fields */ }

A position-based journal for variable-length items.

This journal manages section assignment automatically, allowing callers to append items sequentially without manually tracking section numbers.

Invariants

1. Section Fullness

All non-final sections are full (items_per_section items) and persisted. This ensures that on init(), we only need to replay the last section to determine the exact size.

2. Data Journal is Source of Truth

The data journal is always the source of truth. The offsets journal is an index that may temporarily diverge during crashes. Divergences are automatically aligned during init():

• If offsets.size() < data.size(): Rebuild missing offsets by replaying data. (This can happen if we crash after writing data journal but before writing offsets journal)
• If offsets.size() > data.size(): Rewind offsets to match data size. (This can happen if we crash after rewinding data journal but before rewinding offsets journal)
• If offsets.oldest_retained_pos() < data.oldest_retained_pos(): Prune offsets to match (This can happen if we crash after pruning data journal but before pruning offsets journal)

Note that we don’t recover from the case where offsets.oldest_retained_pos() > data.oldest_retained_pos(). This should never occur because we always prune the data journal before the offsets journal.

Implementations

impl<E: Storage + Metrics, V: Codec> Journal<E, V>

pub async fn init(context: E, cfg: Config<V::Cfg>) -> Result<Self, Error>

Initialize a contiguous variable journal.

Crash Recovery

The data journal is the source of truth. If the offsets journal is inconsistent it will be updated to match the data journal.

pub async fn init_at_size( context: E, cfg: Config<V::Cfg>, size: u64, ) -> Result<Self, Error>

Initialize a Journal in a fully pruned state at a specific logical size.

This creates a journal that reports size() as size but contains no data. The oldest_retained_pos() will return None, indicating all positions before size have been pruned. This is useful for state sync when starting from a non-zero position without historical data.

Arguments

• size - The logical size to initialize at.

Post-conditions

• size() returns size
• oldest_retained_pos() returns None (fully pruned)
• Next append receives position size

pub async fn rewind(&mut self, size: u64) -> Result<(), Error>

Rewind the journal to the given size, discarding items from the end.

After rewinding to size N, the journal will contain exactly N items, and the next append will receive position N.

Errors

Returns Error::InvalidRewind if size is invalid (too large or points to pruned data).

Warning

• This operation is not guaranteed to survive restarts until commit or sync is called.

pub async fn append(&mut self, item: V) -> Result<u64, Error>

Append a new item to the journal, returning its position.

The position returned is a stable, consecutively increasing value starting from 0. This position remains constant after pruning.

When a section becomes full, both the data journal and offsets journal are persisted to maintain the invariant that all non-final sections are full and consistent.

Errors

Returns an error if the underlying storage operation fails or if the item cannot be encoded.

Errors may leave the journal in an inconsistent state. The journal should be closed and reopened to trigger alignment in Journal::init.

pub const fn size(&self) -> u64

Return the total number of items that have been appended to the journal.

This count is NOT affected by pruning. The next appended item will receive this position as its value.

pub const fn oldest_retained_pos(&self) -> Option<u64>

Return the position of the oldest item still retained in the journal.

Returns None if the journal is empty or if all items have been pruned.

pub fn pruning_boundary(&self) -> u64

Returns the location before which all items have been pruned.

pub async fn prune(&mut self, min_position: u64) -> Result<bool, Error>

Prune items at positions strictly less than min_position.

Returns true if any data was pruned, false otherwise.

Errors

Returns an error if the underlying storage operation fails.

Errors may leave the journal in an inconsistent state. The journal should be closed and reopened to trigger alignment in Journal::init.

pub async fn replay( &self, start_pos: u64, buffer_size: NonZeroUsize, ) -> Result<impl Stream<Item = Result<(u64, V), Error>> + '_, Error>

Return a stream of all items in the journal starting from start_pos.

Each item is yielded as a tuple (position, item) where position is the item’s position in the journal.

Errors

Returns an error if start_pos exceeds the journal size or if any storage/decoding errors occur during replay.

pub async fn read(&self, position: u64) -> Result<V, Error>

Read the item at the given position.

Errors

• Returns Error::ItemPruned if the item at position has been pruned.
• Returns Error::ItemOutOfRange if position is beyond the journal size.
• Returns other errors if storage or decoding fails.

pub async fn commit(&mut self) -> Result<(), Error>

Durably persist the journal.

This is faster than sync() but recovery will be required on startup if a crash occurs before the next call to sync().

pub async fn sync(&mut self) -> Result<(), Error>

Durably persist the journal and ensure recovery is not required on startup.

This is slower than commit() but ensures the journal doesn’t require recovery on startup.

pub async fn close(self) -> Result<(), Error>

Close the journal, persisting all pending writes.

This closes both the data journal and the offsets journal.

pub async fn destroy(self) -> Result<(), Error>

Remove any underlying blobs created by the journal.

This destroys both the data journal and the offsets journal.

Trait Implementations

impl<E: Storage + Metrics, V: Codec> Contiguous for Journal<E, V>

type Item = V

The type of items stored in the journal.

fn size(&self) -> u64

Return the total number of items that have been appended to the journal.

fn oldest_retained_pos(&self) -> Option<u64>

Return the position of the oldest item still retained in the journal.

fn pruning_boundary(&self) -> u64

Return the location before which all items have been pruned.

async fn replay( &self, start_pos: u64, buffer: NonZeroUsize, ) -> Result<impl Stream<Item = Result<(u64, Self::Item), Error>> + '_, Error>

Return a stream of all items in the journal starting from start_pos.

async fn read(&self, position: u64) -> Result<Self::Item, Error>

Read the item at the given position.

impl<E: Storage + Metrics, V: Codec> MutableContiguous for Journal<E, V>

async fn append(&mut self, item: Self::Item) -> Result<u64, Error>

Append a new item to the journal, returning its position.

async fn prune(&mut self, min_position: u64) -> Result<bool, Error>

Prune items at positions strictly less than min_position.

async fn rewind(&mut self, size: u64) -> Result<(), Error>

Rewind the journal to the given size, discarding items from the end.

fn rewind_to<'a, P>( &'a mut self, predicate: P, ) -> impl Future<Output = Result<u64, Error>> + 'a

where P: FnMut(&Self::Item) -> bool + 'a,

Rewinds the journal to the last item matching predicate. If no item matches, the journal is rewound to the pruning boundary, discarding all unpruned items.

impl<E: Storage + Metrics, V: Codec> PersistableContiguous for Journal<E, V>

async fn commit(&mut self) -> Result<(), Error>

Durably persist the journal but does not write all data, potentially leaving recovery required on startup.

async fn sync(&mut self) -> Result<(), Error>

Durably persist the journal and write all data, guaranteeing no recovery will be required on startup.

async fn close(self) -> Result<(), Error>

Close the journal, syncing all pending writes and releasing resources.

async fn destroy(self) -> Result<(), Error>

Destroy the journal, removing all associated storage.