OpenRaft Storage for Replication: Implementing the RaftLogStorage and RaftLogReader Traits

OpenRaft delegates all persistent state management to pluggable storage traits—RaftLogStorage for writes and RaftLogReader for reads—allowing you to back the replication engine with anything from in-memory maps to production databases.

OpenRaft, the high-performance Raft consensus implementation from databendlabs/openraft, enforces a strict separation between the replication engine and persistent storage. To integrate custom persistence for log replication, you implement two async traits that manage log entries, votes, and snapshots while guaranteeing serialized writes and no log holes.

Core Storage Traits for OpenRaft Replication

The storage API lives in openraft/src/storage/v2/ and defines the contract between the Raft engine and your persistence layer.

RaftLogStorage – The Write Interface

The RaftLogStorage trait in openraft/src/storage/v2/raft_log_storage.rs handles durable writes of log entries, votes, and truncation operations. Every method must respect the no log holes and serialized writes guarantees to prevent split-brain scenarios.

#[add_async_trait]
pub trait RaftLogStorage<C>: OptionalSend + OptionalSync + 'static
where C: RaftTypeConfig
{
    type LogReader: RaftLogReader<C>;

    async fn get_log_state(&mut self) -> Result<LogState<C>, io::Error>;

    async fn get_log_reader(&mut self) -> Self::LogReader;

    async fn save_vote(&mut self, vote: &VoteOf<C>) -> Result<(), io::Error>;

    async fn save_committed(&mut self, _committed: Option<LogIdOf<C>>) -> Result<(), io::Error> { Ok(()) }

    async fn read_committed(&mut self) -> Result<Option<LogIdOf<C>>, io::Error> { Ok(None) }

    async fn append<I>(&mut self, entries: I, callback: IOFlushed<C>) -> Result<(), io::Error>
    where I: IntoIterator<Item = C::Entry> + OptionalSend,
          I::IntoIter: OptionalSend;

    async fn truncate_after(&mut self, last_log_id: Option<LogIdOf<C>>) -> Result<(), io::Error>;

    async fn purge(&mut self, log_id: LogIdOf<C>) -> Result<(), io::Error>;
}

Key methods include append for inserting entries with a completion callback, save_vote for persisting the current term and voted-for node, and purge for log compaction. The append method must invoke callback.io_completed() only after entries are durably stored.

RaftLogReader – The Read Interface for Replication

The RaftLogReader trait in openraft/src/storage/v2/raft_log_reader.rs provides the streaming interface used by the leader to replicate logs to followers.

#[add_async_trait]
pub trait RaftLogReader<C>: OptionalSend + OptionalSync + 'static
where C: RaftTypeConfig
{
    async fn leader_bounded_stream<RB>(&mut self, leader: LeaderIdOf<C>, range: RB)
        -> impl Stream<Item = LeaderBoundedStreamResult<C>> + OptionalSend
    where RB: RangeBounds<u64> + Clone + Debug + OptionalSend;

    async fn entries_stream<RB>(&mut self, range: RB)
        -> impl Stream<Item = EntriesStreamResult<C>> + OptionalSend
    where RB: RangeBounds<u64> + Clone + Debug + OptionalSend;

    async fn try_get_log_entries<RB>(&mut self, range: RB) -> Result<Vec<C::Entry>, io::Error>
    where RB: RangeBounds<u64> + Clone + Debug;
}

The leader_bounded_stream method is essential for safety: it monitors the current leader ID and automatically terminates the stream if the leader changes, preventing followers from receiving stale entries during a leadership transition.

Reference Implementation: Inside MemLogStore

The MemLogStore in stores/memstore/src/lib.rs provides a canonical implementation using an async RwLock-protected BTreeMap<u64, String>.

Method	Location	Description
get_log_state	#L77-L100	Computes the last log ID and last purged ID by scanning the entry map.
save_vote	#L107-L113	Persists the Vote struct inside the async lock.
append	#L138-L147	Serializes entries to JSON, inserts into the map, and triggers the IOFlushed callback.
truncate_after	#L152-L160	Removes all entries with indices greater than last_log_id.
purge	#L162-L171	Updates the purge marker and drops entries up to the specified log_id.

The implementation starts at line 74 with impl RaftLogStorage<TypeConfig> for Arc<MemLogStore>. It also implements RaftSnapshotBuilder and RaftStateMachine (lines 332-371) to support snapshotting and state-machine interaction.

Integrating Custom Storage with OpenRaft

To wire a storage implementation into a Raft node, wrap it in an Arc and pass it to Raft::new. The storage must be Send + Sync because OpenRaft clones it for background replication tasks.

use std::sync::Arc;
use openraft::{Raft, Config};
use openraft::storage::{RaftLogStorage, RaftStateMachine};
use memstore::{MemLogStore, MemStateMachine, TypeConfig, BlockConfig};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let block_cfg = BlockConfig::default();
    let log_store = Arc::new(MemLogStore::new(block_cfg));
    let sm_store = Arc::new(MemStateMachine::default());
    
    let cfg = Arc::new(Config::default().validate()?);
    
    let raft = Raft::new(cfg, log_store, sm_store).await?;
    
    raft.add_learner(2, "127.0.0.1:5002".to_string()).await?;
    
    Ok(())
}

Skeleton for a Persistent Store

When implementing a production backend (e.g., RocksDB or PostgreSQL), follow this structure:

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

#[derive(Debug, Clone)]
pub struct PersistentLogStore {
    // Database connection pool
}

#[async_trait]
impl<C: openraft::RaftTypeConfig> RaftLogReader<C> for PersistentLogStore {
    async fn try_get_log_entries<RB>(&mut self, range: RB) -> Result<Vec<C::Entry>, io::Error>
    where RB: std::ops::RangeBounds<u64> + Clone + std::fmt::Debug {
        // Query database for entries in range
        todo!()
    }
}

#[async_trait]
impl<C: openraft::RaftTypeConfig> RaftLogStorage<C> for PersistentLogStore {
    type LogReader = Self;

    async fn get_log_state(&mut self) -> Result<LogState<C>, io::Error> {
        // Return highest log ID and last purged ID
        todo!()
    }

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

    async fn save_vote(&mut self, vote: &Vote<C>) -> Result<(), io::Error> {
        // Atomic write of vote
        todo!()
    }

    async fn append<I>(&mut self, entries: I, cb: IOFlushed<C>) -> Result<(), io::Error>
    where I: IntoIterator<Item = C::Entry> + openraft::OptionalSend {
        // Transactional insert, then call cb.io_completed()
        todo!()
    }

    async fn truncate_after(&mut self, last_log_id: Option<LogId<C>>) -> Result<(), io::Error> {
        todo!()
    }

    async fn purge(&mut self, log_id: LogId<C>) -> Result<(), io::Error> {
        todo!()
    }
}

Your implementation must ensure that append operations are visible immediately to subsequent reads (no holes) and that concurrent writes complete in the order they were issued (serialization).

Summary

• OpenRaft storage for replication requires implementing RaftLogStorage for durable writes and RaftLogReader for log streaming.
• Implementations must guarantee no log holes and serialized writes to maintain consensus safety and prevent split-brain scenarios.
• Use Arc<YourStore> when constructing the Raft instance, as the engine clones storage handles for background tasks.
• Reference MemLogStore in stores/memstore/src/lib.rs for a complete, thread-safe implementation pattern that includes snapshot support.

Frequently Asked Questions

What are the correctness requirements for OpenRaft storage?

Your storage must guarantee no log holes, meaning every appended entry is immediately visible to readers, and serialized writes, ensuring that a later write cannot complete before an earlier one. These properties prevent followers from detecting false gaps and protect against split-brain scenarios during leader transitions.

How does OpenRaft handle leader changes during log streaming?

The RaftLogReader trait provides leader_bounded_stream, which accepts a LeaderId and automatically terminates the stream if the leader changes. This ensures followers stop reading from a deposed leader immediately, preventing replication of stale entries.

Can I use a synchronous database driver with OpenRaft storage?

No. All storage trait methods are async and must not block the executor. If using a synchronous database driver, offload blocking operations to a dedicated thread pool (e.g., tokio::task::spawn_blocking) before returning to the async context.

What is the difference between RaftLogStorage and RaftLogReader?

RaftLogStorage is the write-side interface used by the Raft state machine to append entries, save votes, and truncate logs. RaftLogReader is the read-side interface used exclusively by the replication task to stream entries to followers. A single struct can implement both traits, as demonstrated by MemLogStore.

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