# OpenRaft Storage for Replication: Implementing the RaftLogStorage and RaftLogReader Traits

> Learn how OpenRaft uses `RaftLogStorage` and `RaftLogReader` traits for flexible replication state management. Back your engine with any storage, from memory to production databases.

- Repository: [Databend Labs/openraft](https://github.com/databendlabs/openraft)
- Tags: deep-dive
- Published: 2026-02-28

---

**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`](https://github.com/databendlabs/openraft/blob/main/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.

```rust
#[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`](https://github.com/databendlabs/openraft/blob/main/openraft/src/storage/v2/raft_log_reader.rs) provides the streaming interface used by the leader to replicate logs to followers.

```rust
#[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`](https://github.com/databendlabs/openraft/blob/main/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.

```rust
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:

```rust
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`](https://github.com/databendlabs/openraft/blob/main/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`.