OpenRaft Storage Implementation: A Complete Guide to Async Traits and Custom Backends

OpenRaft storage implementation relies on four async traits—RaftLogStorage, RaftLogReader, RaftStateMachine, and RaftSnapshotBuilder—that separate log persistence, state machine application, and snapshot handling to allow any backend from in-memory stores to RocksDB.

The databendlabs/openraft crate decouples the Raft consensus algorithm from persistence details through a generic storage layer. By implementing the async storage traits defined in openraft::storage::v2, you can wire OpenRaft to any durable backend while maintaining strict guarantees about log continuity and write ordering.

Core Storage Architecture in OpenRaft

OpenRaft’s storage layer is partitioned into three responsibilities: log storage, state machine persistence, and snapshot management. Each responsibility maps to a specific trait generic over a type configuration C: RaftTypeConfig that defines your application-specific request, response, and node ID types.

RaftLogStorage and RaftLogReader

The RaftLogStorage<C> trait, defined in openraft/src/storage/v2/raft_log_storage.rs, handles persistent storage for log entries and Raft state. Key methods include:

• append – Writes new entries to the log and invokes the IOFlushed callback when persisted.
• save_vote – Persists the current Raft vote to disk before returning.
• purge – Removes entries up to a specific log ID to reclaim space.
• get_log_reader – Returns a RaftLogReader instance for reading entries.

The RaftLogReader<C> trait, defined in openraft/src/storage/v2/raft_log_reader.rs, provides a read-only interface used by replication workers:

• try_get_log_entries – Fetches entries within a given index range.
• read_vote – Retrieves the persisted vote state.

RaftStateMachine and RaftSnapshotBuilder

The RaftStateMachine<C> trait, defined in openraft/src/storage/v2/raft_state_machine.rs, represents your application state that consumes committed log entries:

• apply – Processes a stream of committed entries and sends responses via EntryResponder.
• install_snapshot – Replaces the current state with data from a snapshot streamed by the leader.
• begin_receiving_snapshot – Allocates a writable handle for incoming snapshot data.
• get_current_snapshot – Returns the latest persisted snapshot for replication or recovery.

The RaftSnapshotBuilder<C> trait, defined in openraft/src/storage/v2/raft_snapshot_builder.rs, creates read-only snapshots:

• build_snapshot – Serializes the current state machine state and returns a Snapshot with metadata.

High-Level Data Flow

Understanding the interaction between these traits clarifies how OpenRaft persists and replicates data:

1. Leader appends entries – The Raft engine calls RaftLogStorage::append to persist new entries to the log storage backend.

2. Followers read entries – Replication workers obtain a RaftLogReader via get_log_reader and call try_get_log_entries to fetch entries for replication.

3. Committed entries apply – Once entries are committed, the engine delivers them to RaftStateMachine::apply, where your business logic processes the requests.

4. Snapshot creation – When snapshot policy triggers or a follower lags too far behind, RaftSnapshotBuilder::build_snapshot creates a point-in-time snapshot of the state machine.

5. Snapshot installation – On the follower side, RaftStateMachine::install_snapshot replaces the local state with the streamed snapshot data, allowing the node to catch up without replaying the entire log.

OpenRaft guarantees that log operations are serialized and never leave a "hole," ensuring that the append method writes entries consecutively and that purge or truncate_after operations maintain consistency.

Implementing a Custom Storage Backend

To integrate OpenRaft with your persistence layer, you must implement the four core traits. Below is a structured skeleton demonstrating the required methods for each component.

Log Storage Implementation

Implement RaftLogStorage for your log backend and RaftLogReader for read operations:

use openraft::storage::{RaftLogStorage, RaftLogReader, IOFlushed};
use openraft::{Entry, LogId, Vote, RaftTypeConfig, LogState};
use std::io;

pub struct MyConfig;
impl RaftTypeConfig for MyConfig {
    type D = String;
    type R = String;
    type NodeId = u64;
    type Entry = openraft::Entry<MyConfig>;
    type SnapshotData = std::io::Cursor<Vec<u8>>;
}

pub struct MyLogStore {
    // Your fields: e.g., RocksDB handle, PostgreSQL pool, etc.
}

#[async_trait::async_trait]
impl RaftLogStorage<MyConfig> for MyLogStore {
    type LogReader = Self;

    async fn get_log_state(&mut self) -> Result<LogState<MyConfig>, io::Error> {
        // Return LogState { last_purged_log_id, last_log_id }
        unimplemented!()
    }

    async fn get_log_reader(&mut self) -> Self::LogReader {
        self.clone()
    }

    async fn save_vote(&mut self, vote: &Vote<MyConfig>) -> Result<(), io::Error> {
        // Persist vote atomically before returning
        unimplemented!()
    }

    async fn append<I>(
        &mut self,
        entries: I,
        callback: IOFlushed<MyConfig>,
    ) -> Result<(), io::Error>
    where
        I: IntoIterator<Item = Entry<MyConfig>> + Send,
        I::IntoIter: Send,
    {
        // Serialize and persist entries, then invoke callback
        unimplemented!()
    }

    async fn truncate_after(&mut self, log_id: LogId<MyConfig>) -> Result<(), io::Error> {
        // Remove entries after log_id
        unimplemented!()
    }

    async fn purge(&mut self, log_id: LogId<MyConfig>) -> Result<(), io::Error> {
        // Remove entries up to log_id
        unimplemented!()
    }
}

#[async_trait::async_trait]
impl RaftLogReader<MyConfig> for MyLogStore {
    async fn try_get_log_entries<RB>(
        &mut self,
        range: RB,
    ) -> Result<Vec<Entry<MyConfig>>, io::Error>
    where
        RB: std::ops::RangeBounds<u64> + Clone + Send,
    {
        // Fetch entries from storage
        unimplemented!()
    }

    async fn read_vote(&mut self) -> Result<Option<Vote<MyConfig>>, io::Error> {
        // Read persisted vote
        unimplemented!()
    }
}

State Machine Implementation

Implement RaftStateMachine to apply committed entries and handle snapshots:

use openraft::storage::{RaftStateMachine, EntryResponder, Snapshot};
use openraft::{Entry, LogId, SnapshotMeta, RaftTypeConfig, StoredMembership};
use futures::StreamExt;
use std::io;

pub struct MyStateMachine {
    // Your application state
}

#[async_trait::async_trait]
impl RaftStateMachine<MyConfig> for MyStateMachine {
    type SnapshotBuilder = MySnapshotBuilder;

    async fn applied_state(
        &mut self,
    ) -> Result<(Option<LogId<MyConfig>>, StoredMembership<MyConfig>), io::Error> {
        // Return last applied log id and current membership
        unimplemented!()
    }

    async fn apply<Strm>(
        &mut self,
        mut entries: Strm,
    ) -> Result<(), io::Error>
    where
        Strm: futures::Stream<Item = Result<EntryResponder<MyConfig>, io::Error>> + Unpin + Send,
    {
        while let Some(entry) = entries.next().await {
            let entry = entry?;
            // Apply business logic here
            let response = format!("Applied: {:?}", entry);
            entry.send_response(response).await?;
        }
        Ok(())
    }

    async fn try_create_snapshot_builder(
        &mut self,
        force: bool,
    ) -> Option<Self::SnapshotBuilder> {
        if force || self.should_snapshot() {
            Some(MySnapshotBuilder::new(self))
        } else {
            None
        }
    }

    async fn get_current_snapshot(
        &mut self,
    ) -> Result<Option<Snapshot<MyConfig>>, io::Error> {
        // Return latest persisted snapshot
        unimplemented!()
    }

    async fn install_snapshot(
        &mut self,
        meta: &SnapshotMeta<MyConfig>,
        snapshot: <MyConfig as RaftTypeConfig>::SnapshotData,
    ) -> Result<(), io::Error> {
        // Replace internal state with snapshot data
        unimplemented!()
    }

    async fn begin_receiving_snapshot(
        &mut self,
    ) -> Result<<MyConfig as RaftTypeConfig>::SnapshotData, io::Error> {
        // Allocate buffer for incoming snapshot
        unimplemented!()
    }
}

Snapshot Builder Implementation

Implement RaftSnapshotBuilder to create point-in-time snapshots:

use openraft::storage::RaftSnapshotBuilder;
use openraft::{Snapshot, SnapshotMeta, RaftTypeConfig};
use std::io;

pub struct MySnapshotBuilder {
    state_machine: MyStateMachine,
}

impl MySnapshotBuilder {
    fn new(sm: &MyStateMachine) -> Self {
        Self { state_machine: sm.clone() }
    }
}

#[async_trait::async_trait]
impl RaftSnapshotBuilder<MyConfig> for MySnapshotBuilder {
    async fn build_snapshot(&mut self) -> Result<Snapshot<MyConfig>, io::Error> {
        // Serialize state machine state
        let data = serde_json::to_vec(&self.state_machine)?;
        let meta = SnapshotMeta {
            last_log_id: self.state_machine.last_log_id,
            last_membership: self.state_machine.membership.clone(),
            snapshot_id: format!("snapshot-{}", uuid::Uuid::new_v4()),
        };
        
        Ok(Snapshot {
            meta,
            snapshot: std::io::Cursor::new(data),
        })
    }
}

Reference Implementation: memstore

The memstore crate in stores/memstore/src/lib.rs provides a complete, production-ready reference implementation. It demonstrates:

• Log storage: MemLogStore implements RaftLogStorage using a BTreeMap for entries and tracking last_purged_log_id and last_log_id in LogState.
• State machine: MemStateMachine implements RaftStateMachine with an in-memory key-value map, handling apply for committed entries and install_snapshot for recovery.
• Snapshot building: MemSnapshotBuilder creates snapshots by serializing the entire state machine into a byte vector wrapped in a Cursor.

Reviewing the impl RaftLogStorage<TypeConfig> for Arc<MemLogStore> block starting at line 374 and the impl RaftStateMachine<TypeConfig> for Arc<MemStateMachine> block provides a concrete pattern for error handling, callback invocation, and state management.

Summary

• OpenRaft storage implementation centers on four async traits: RaftLogStorage, RaftLogReader, RaftStateMachine, and RaftSnapshotBuilder.
• Log storage must guarantee consecutive entries without gaps, serialized writes, and atomic vote persistence via save_vote.
• State machines consume committed entries through apply, create snapshots via RaftSnapshotBuilder::build_snapshot, and recover via install_snapshot.
• The memstore crate provides a complete reference implementation demonstrating correct trait implementation patterns.
• Custom backends can replace the in-memory store by implementing the same traits with RocksDB, PostgreSQL, S3, or other durable storage.

Frequently Asked Questions

What is the difference between RaftLogStorage and RaftStateMachine?

RaftLogStorage manages the append-only log entries and Raft vote state, ensuring durability of the consensus log, while RaftStateMachine represents your application logic that consumes committed entries via the apply method and handles snapshot creation and installation. The log storage is concerned with what to replicate, whereas the state machine handles how to apply that replication to your specific domain.

How does OpenRaft handle snapshot streaming?

When a follower lags too far behind or a new node joins, the leader invokes RaftSnapshotBuilder::build_snapshot to create a point-in-time snapshot, then streams it to the follower. The follower receives the snapshot through RaftStateMachine::begin_receiving_snapshot to allocate a buffer, followed by install_snapshot to replace its current state with the received data, allowing it to catch up without processing the entire log history.

Can I use a non-async storage backend with OpenRaft?

No, OpenRaft requires all storage traits to be async, but you can wrap synchronous I/O operations such as RocksDB or PostgreSQL calls within async blocks using tokio::task::spawn_blocking or equivalent runtime utilities. This approach satisfies the async trait requirements while allowing you to leverage existing synchronous database drivers or file system operations.

What guarantees must a RaftLogStorage implementation provide?

Your implementation must ensure that log entries are stored consecutively without gaps, that all write operations (appends, vote saves, and purges) are serialized in order, and that save_vote persists the vote to durable storage before returning. These guarantees are essential for OpenRaft to maintain consensus safety and recover correctly after crashes.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}