Struct Journal 

pub struct Journal<E: Context, V: Codec>(/* private fields */);

A contiguous journal with variable-size entries.

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

Repair

Like sqlite and rocksdb, the first invalid data read above the recovery watermark’s acknowledged prefix is considered the new end of the journal (and the underlying blob is truncated to the last valid item). Data at or below that prefix is never repaired: a blob that no longer backs its acknowledged items fails init, and other damage there surfaces as a read error. Repair is performed during init. Incomplete trailing frames are repaired as torn writes. Complete frames whose payloads fail to decode are treated as corruption.

Invariants

1. Data Blobs are the Source of Truth

The data blobs are 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 are behind data after the recovery watermark: rebuild missing offsets by replaying data from the recovery anchor.
• If offsets are ahead of the retained data prefix but the data still reaches the recovery watermark, truncate offsets to match the data-backed size. Retained data ending before the watermark is corruption because acknowledged data is missing.
• If offsets.bounds().start < the oldest data blob’s start: prune offsets to match (this can happen if we crash after pruning the data blobs but before pruning the offsets journal).

Offsets may start after the data’s blob-aligned start when both are in the same blob, as in a mid-blob init_at_size. Offsets starting in a later blob imply corruption because we always prune the data blobs before the offsets journal.

2. Offsets Recovery Watermark

The offsets journal’s recovery watermark records a durable lower bound on the journal size and a preferred point for replaying data to rebuild offset entries after a crash. Fixed-journal recovery rejects watermarks beyond the recovered offsets size as corruption. A watermark below the recovered offsets start is stale after a prune, so init falls back to the offsets start. If retained data exists but ends before the watermark, init returns corruption because acknowledged data is missing. If no retained data exists, init reconciles both sides to an empty journal. Replay after a valid anchor stops at the first short data blob and truncates newer blobs so the recovered journal remains a contiguous prefix.

Mutating functions consume the journal and return it only on success: an error (or a dropped future) destroys the handle.

Implementations

impl<E: Context, V: CodecShared> 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 blobs are the source of truth. If the offsets journal is inconsistent it will be updated to match the data blobs.

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

Open at most max_size items, including pruned positions.

The discarded suffix is durably removed before success. Subsequent appends may exceed the cap. A cap below the retained start returns Error::ItemPruned.

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

Initialize an empty Journal at the given logical size.

This discards any existing data and offsets. The offsets reset intent is staged before the data partition is cleared so recovery can complete the requested reset if a crash interrupts the operation.

Returns a journal with journal.bounds() == Range{start: size, end: size} and next append at position size.

Errors

Returns Error::SizeOverflow if size is u64::MAX.

pub async fn append(self, item: &V) -> Result<(Self, 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.

Errors

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

pub async fn append_many(self, items: Many<'_, V>) -> Result<(Self, u64), Error>

Append items to the journal, returning the position of the last item appended.

Returns Error::EmptyAppend if items is empty.

pub fn prepare_append( &self, items: Many<'_, V>, ) -> Result<PreparedAppend<V>, Error>

Encode items into a buffer that can be appended later with Self::append_prepared.

This lets callers serialize borrowed items synchronously, release those borrows, and perform the append without holding unrelated locks across journal I/O.

pub async fn append_prepared( self, prepared: PreparedAppend<V>, ) -> Result<(Self, u64), Error>

Append items encoded by Self::prepare_append, returning the position of the last item appended.

Returns Error::EmptyAppend if prepared contains no items. Returns Error::InvalidConfiguration if prepared was encoded with different compression settings than this journal uses.

pub async fn snapshot(self) -> Result<(Self, Reader<'static, E, V>), Error>

Capture an owned snapshot (Reader) over the current journal. Bounds are frozen at creation, and the snapshot stays readable across concurrent appends and prunes. It keeps the journal’s blobs open, so reopening a partition that still holds one of them fails while the snapshot is alive.

pub fn size(&self) -> u64

Return the total number of items in the journal, irrespective of pruning. The next value appended to the journal will be at this position.

pub async fn prune(self, min_position: u64) -> Result<(Self, 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.

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

Persist data blobs so committed data survives a crash.

Does not advance the recovery watermark, so reopen may replay entries above it.

pub async fn start_sync(self) -> Result<(Self, Handle<()>), Error>

Begin durably persisting the current state of the journal.

Awaiting the returned Handle guarantees state appended before this call survives a crash. Also tries to advance the recovery watermark to the previous proven durable size, bounding startup recovery. Only sync() guarantees a current watermark.

At most one data sync and one watermark sync are in flight at a time: this call waits for the prior call’s syncs before starting new ones. It does not wait for a pending rollover fsync: the returned handle joins it, so an earlier call’s handle may still be pending when this call returns. Reads proceed while the returned handle is pending, and appends proceed while they fit in the write buffer.

Flush errors are returned directly. Dropping the handle does not cancel the sync or lose its failure: a later operation reports it, no later than the next Self::sync.

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

Persist data blobs and all metadata for both the data and offsets journals.

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

Remove any underlying blobs created by the journal.

This destroys both the data blobs and the offsets journal. A started sync that failed is reported before anything is removed.

Crash Safety

This operation is intended for final teardown and is not crash-safe. If interrupted, reopening the same partitions may observe partially removed state. Use Self::init_at_size for a recoverable reset.

Trait Implementations

impl<E: Context, V: CodecShared> Backing<E> for Journal<E, V>

Available on neither commonware_stability_BETA nor commonware_stability_DELTA nor commonware_stability_EPSILON nor commonware_stability_GAMMA nor commonware_stability_RESERVED.

type Recovery = Recovery<E, V>

Initialization-owned storage used to select and validate the retained prefix.

type Config = Config<<V as Read>::Cfg>

The configuration needed to initialize this journal.

async fn recover( context: E, cfg: Self::Config, max_size: Option<u64>, ) -> Result<Self::Recovery, Error>

Open recovery storage for an optional exclusive item end. Implementations may inspect later storage to validate recovery boundaries. Returns JournalError::ItemPruned when max_size lies below the retained start.

async fn span(context: E, cfg: &Self::Config) -> Result<Range<u64>, Error>

Positions stored items may occupy, determined without reading them.

async fn clear( context: E, cfg: Self::Config, size: u64, ) -> Result<Self::Recovery, Error>

Open recovery storage reset to an empty journal at size, discarding stored items without reading them. Returns JournalError::SizeOverflow for u64::MAX.

impl<E: Context, V: CodecShared> Contiguous for Journal<E, V>

type Item = V

The type of items stored in the journal.

fn bounds(&self) -> Range<u64>

Returns [start, end) with a guaranteed stable pruning boundary.

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

Read the item at the given position.

async fn read_many(&self, positions: &[u64]) -> Result<Vec<V>, Error>

Read multiple items at the given positions, which must be strictly increasing.

fn try_read_sync(&self, position: u64) -> Option<V>

Read an item if it can be done synchronously (e.g. without I/O), returning None otherwise. Decode failures surface as None and the async read path reports the error.

fn try_read_many_sync(&self, positions: &[u64]) -> Vec<Option<V>>

Probe multiple strictly increasing positions, serving those that can be read synchronously (e.g. from a page cache) and returning one slot per position. Positions that require I/O, fail to decode, or fall outside bounds() decline to None. The async read paths are the sole error authority for declined positions.

async fn replay( &self, range: Range<u64>, buffer: NonZeroUsize, read_options: ReadOptions, ) -> Result<impl Stream<Item = Result<Batch<V>, Error>> + Send, Error>

Return a stream of the items in range as Batches of consecutive items, in position order. range must fall within bounds(). An empty range within bounds() yields an empty stream.

impl<E: Context, V: CodecShared> Debug for Journal<E, V>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<E: Context, V: CodecShared> Mutable for Journal<E, V>

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

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

async fn append_many( self, items: Many<'_, Self::Item>, ) -> Result<(Self, u64), Error>

Append items to the journal, returning the position of the last item appended.

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

Prune items at positions strictly less than min_position.

async fn start_sync(self) -> Result<(Self, Handle<()>), Error>

Begin durably persisting the current state of the journal.

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

Durably persist the journal, guaranteeing the current state will survive a crash.

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

Durably persist the journal, guaranteeing the current state will survive a crash, and that no recovery will be needed on startup.

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

Destroy the journal, removing all associated storage.