Struct openraft::StoreExt

pub struct StoreExt<C: RaftTypeConfig, T: RaftStorage<C>> { /* private fields */ }

Extended store backed by another impl.

It provides defensive check against input and the state of underlying store. And it provides more APIs.

Implementations

impl<C: RaftTypeConfig, T: RaftStorage<C>> StoreExt<C, T>

pub fn new(inner: T) -> Self

Create a StoreExt backed by another store.

pub fn set_delay_log_read(&self, ms: u64)

Trait Implementations

impl<C: RaftTypeConfig, T: RaftStorage<C> + Clone> Clone for StoreExt<C, T>

fn clone(&self) -> Self

Returns a copy of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<C, T> DefensiveCheck<C, T> for StoreExt<C, T>where C: RaftTypeConfig, T: RaftStorage<C>,

fn defensive_no_dirty_log<'life0, 'async_trait>( &'life0 mut self ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<C::NodeId>>> + Send + 'async_trait>>where Self: Send + 'async_trait, 'life0: 'async_trait,

Ensure that logs that have greater index than last_applied should have greater log_id. Invariant must hold: log.log_id.index > last_applied.index implies log.log_id > last_applied.

fn defensive_incremental_vote<'life0, 'life1, 'async_trait>( &'life0 mut self, vote: &'life1 Vote<C::NodeId> ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<C::NodeId>>> + Send + 'async_trait>>where Self: Send + 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Ensure that current_term must increment for every update, and for every term there could be only one value for voted_for.

fn defensive_consecutive_input<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, entries: &'life1 [&'life2 Entry<C>] ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<C::NodeId>>> + Send + 'async_trait>>where Self: Sync + 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

The log entries fed into a store must be consecutive otherwise it is a bug.

fn defensive_nonempty_input<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, entries: &'life1 [&'life2 Entry<C>] ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<C::NodeId>>> + Send + 'async_trait>>where Self: Sync + 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Trying to feed in empty entries slice is an inappropriate action.

fn defensive_append_log_index_is_last_plus_one<'life0, 'life1, 'life2, 'async_trait>( &'life0 mut self, entries: &'life1 [&'life2 Entry<C>] ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<C::NodeId>>> + Send + 'async_trait>>where Self: Send + 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

The entries to append has to be last_log_id.index + 1

fn defensive_append_log_id_gt_last<'life0, 'life1, 'life2, 'async_trait>( &'life0 mut self, entries: &'life1 [&'life2 Entry<C>] ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<C::NodeId>>> + Send + 'async_trait>>where Self: Send + 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

The entries to append has to be greater than any known log ids

fn defensive_purge_applied_le_last_applied<'life0, 'async_trait>( &'life0 mut self, upto: LogId<C::NodeId> ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<C::NodeId>>> + Send + 'async_trait>>where Self: Send + 'async_trait, 'life0: 'async_trait,

fn defensive_delete_conflict_gt_last_applied<'life0, 'async_trait>( &'life0 mut self, since: LogId<C::NodeId> ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<C::NodeId>>> + Send + 'async_trait>>where Self: Send + 'async_trait, 'life0: 'async_trait,

fn defensive_apply_index_is_last_applied_plus_one<'life0, 'life1, 'life2, 'async_trait>( &'life0 mut self, entries: &'life1 [&'life2 Entry<C>] ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<C::NodeId>>> + Send + 'async_trait>>where Self: Send + 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

The entries to apply to state machine has to be last_applied_log_id.index + 1

fn defensive_half_open_range<'life0, 'async_trait, RB>( &'life0 self, range: RB ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<C::NodeId>>> + Send + 'async_trait>>where RB: 'async_trait + RangeBounds<u64> + Clone + Debug + Send, Self: Sync + 'async_trait, 'life0: 'async_trait,

Requires a range must be at least half open: (-oo, n] or [n, +oo); In order to keep logs continuity.

fn defensive_range_hits_logs<'life0, 'life1, 'async_trait, RB>( &'life0 self, range: RB, logs: &'life1 [Entry<C>] ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<C::NodeId>>> + Send + 'async_trait>>where RB: 'async_trait + RangeBounds<u64> + Debug + Send, Self: Sync + 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

An range operation such as get or delete has to actually covers some log entries in store.

fn defensive_apply_log_id_gt_last<'life0, 'life1, 'life2, 'async_trait>( &'life0 mut self, entries: &'life1 [&'life2 Entry<C>] ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<C::NodeId>>> + Send + 'async_trait>>where Self: Send + 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

The log id of the entries to apply has to be greater than the last known one.

impl<C, T> DefensiveCheckBase<C> for StoreExt<C, T>where C: RaftTypeConfig, T: RaftStorage<C>,

fn set_defensive(&self, d: bool)

Enable or disable defensive check when calling storage APIs.

fn is_defensive(&self) -> bool

fn defensive_nonempty_range<RB: RangeBounds<u64> + Clone + Debug + Send>( &self, range: RB ) -> Result<(), StorageError<C::NodeId>>

The range must not be empty otherwise it is an inappropriate action.

impl<C: RaftTypeConfig, T: RaftStorage<C>> Deref for StoreExt<C, T>

type Target = T

The resulting type after dereferencing.

fn deref(&self) -> &Self::Target

Dereferences the value.

impl<C: RaftTypeConfig, T: RaftStorage<C>> RaftLogReader<C> for StoreExt<C, T>

fn try_get_log_entries<'life0, 'async_trait, RB>( &'life0 mut self, range: RB ) -> Pin<Box<dyn Future<Output = Result<Vec<Entry<C>>, StorageError<C::NodeId>>> + Send + 'async_trait>>where RB: 'async_trait + RangeBounds<u64> + Clone + Debug + Send + Sync, Self: 'async_trait, 'life0: 'async_trait,

Get a series of log entries from storage.

fn get_log_state<'life0, 'async_trait>( &'life0 mut self ) -> Pin<Box<dyn Future<Output = Result<LogState<C>, StorageError<C::NodeId>>> + Send + 'async_trait>>where Self: 'async_trait, 'life0: 'async_trait,

Returns the last deleted log id and the last log id.

impl<C, T> RaftStorage<C> for StoreExt<C, T>where T: RaftStorage<C>, C: RaftTypeConfig,

type SnapshotData = <T as RaftStorage<C>>::SnapshotData

The storage engine’s associated type used for exposing a snapshot for reading & writing.

type LogReader = LogReaderExt<C, T>

Log reader type.

type SnapshotBuilder = SnapshotBuilderExt<C, T>

Snapshot builder type.

fn save_vote<'life0, 'life1, 'async_trait>( &'life0 mut self, vote: &'life1 Vote<C::NodeId> ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<C::NodeId>>> + Send + 'async_trait>>where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

To ensure correctness: the vote must be persisted on disk before returning.

fn read_vote<'life0, 'async_trait>( &'life0 mut self ) -> Pin<Box<dyn Future<Output = Result<Option<Vote<C::NodeId>>, StorageError<C::NodeId>>> + Send + 'async_trait>>where Self: 'async_trait, 'life0: 'async_trait,

fn last_applied_state<'life0, 'async_trait>( &'life0 mut self ) -> Pin<Box<dyn Future<Output = Result<(Option<LogId<C::NodeId>>, StoredMembership<C::NodeId, C::Node>), StorageError<C::NodeId>>> + Send + 'async_trait>>where Self: 'async_trait, 'life0: 'async_trait,

Returns the last applied log id which is recorded in state machine, and the last applied membership config.

fn delete_conflict_logs_since<'life0, 'async_trait>( &'life0 mut self, log_id: LogId<C::NodeId> ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<C::NodeId>>> + Send + 'async_trait>>where Self: 'async_trait, 'life0: 'async_trait,

Delete conflict log entries since log_id, inclusive.

fn purge_logs_upto<'life0, 'async_trait>( &'life0 mut self, log_id: LogId<C::NodeId> ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<C::NodeId>>> + Send + 'async_trait>>where Self: 'async_trait, 'life0: 'async_trait,

Delete applied log entries upto log_id, inclusive.

fn append_to_log<'life0, 'life1, 'life2, 'async_trait>( &'life0 mut self, entries: &'life1 [&'life2 Entry<C>] ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<C::NodeId>>> + Send + 'async_trait>>where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Append a payload of entries to the log.

fn apply_to_state_machine<'life0, 'life1, 'life2, 'async_trait>( &'life0 mut self, entries: &'life1 [&'life2 Entry<C>] ) -> Pin<Box<dyn Future<Output = Result<Vec<C::R>, StorageError<C::NodeId>>> + Send + 'async_trait>>where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Apply the given payload of entries to the state machine.

fn begin_receiving_snapshot<'life0, 'async_trait>( &'life0 mut self ) -> Pin<Box<dyn Future<Output = Result<Box<Self::SnapshotData>, StorageError<C::NodeId>>> + Send + 'async_trait>>where Self: 'async_trait, 'life0: 'async_trait,

Create a new blank snapshot, returning a writable handle to the snapshot object.

fn install_snapshot<'life0, 'life1, 'async_trait>( &'life0 mut self, meta: &'life1 SnapshotMeta<C::NodeId, C::Node>, snapshot: Box<Self::SnapshotData> ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<C::NodeId>>> + Send + 'async_trait>>where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Install a snapshot which has finished streaming from the leader.

fn get_current_snapshot<'life0, 'async_trait>( &'life0 mut self ) -> Pin<Box<dyn Future<Output = Result<Option<Snapshot<C::NodeId, C::Node, Self::SnapshotData>>, StorageError<C::NodeId>>> + Send + 'async_trait>>where Self: 'async_trait, 'life0: 'async_trait,

Get a readable handle to the current snapshot, along with its metadata.

fn get_log_reader<'life0, 'async_trait>( &'life0 mut self ) -> Pin<Box<dyn Future<Output = Self::LogReader> + Send + 'async_trait>>where Self: 'async_trait, 'life0: 'async_trait,

Get the log reader.

fn get_snapshot_builder<'life0, 'async_trait>( &'life0 mut self ) -> Pin<Box<dyn Future<Output = Self::SnapshotBuilder> + Send + 'async_trait>>where Self: 'async_trait, 'life0: 'async_trait,

Get the snapshot builder for the state machine.

impl<C, T, SM> RaftStorageDebug<SM> for StoreExt<C, T>where T: RaftStorage<C> + RaftStorageDebug<SM>, C: RaftTypeConfig,

fn get_state_machine<'life0, 'async_trait>( &'life0 mut self ) -> Pin<Box<dyn Future<Output = SM> + Send + 'async_trait>>where Self: 'async_trait, 'life0: 'async_trait,

Get a handle to the state machine for testing purposes.

impl<C, BaseStore, BaseBuilder> StoreBuilder<C, StoreExt<C, BaseStore>> for DefensiveStoreBuilder<C, BaseStore, BaseBuilder>where C: RaftTypeConfig, C::D: AppData + Debug, C::R: AppDataResponse + Debug, BaseStore: RaftStorage<C>, BaseBuilder: StoreBuilder<C, BaseStore>,

fn run_test<'life0, 'async_trait, Fun, Ret, Res>( &'life0 self, t: Fun ) -> Pin<Box<dyn Future<Output = Result<Ret, StorageError<C::NodeId>>> + Send + 'async_trait>>where Res: Future<Output = Result<Ret, StorageError<C::NodeId>>> + Send + 'async_trait, Fun: Fn(StoreExt<C, BaseStore>) -> Res + Sync + Send + 'async_trait, Ret: 'async_trait, Self: 'async_trait, 'life0: 'async_trait,

impl<C, T> Wrapper<C, T> for StoreExt<C, T>where C: RaftTypeConfig, T: RaftStorage<C>,

fn inner(&mut self) -> &mut T