Understanding the OpenRaft Storage Module: Persistence Architecture and Implementation

The OpenRaft storage module defines the persistence contracts through four core traits—RaftLogStorage, RaftStateMachine, RaftLogReader, and RaftSnapshotBuilder—that separate log persistence from state-machine application and snapshot management.

The openraft::storage module in the databendlabs/openraft repository provides the essential interfaces that applications must implement to integrate custom persistence layers. This architecture cleanly separates log storage (WAL entries and votes) from state-machine storage (application state and snapshots), enabling flexible backend choices ranging from in-memory implementations to production-grade disk or distributed storage.

Core Storage Traits in the OpenRaft Storage Module

The storage module is organized around four primary traits that define the contract between the Raft engine and the underlying storage layer.

RaftLogStorage: Log Persistence Contract

Defined in openraft/src/storage/v2/raft_log_storage.rs, the RaftLogStorage trait handles all persistent storage of Raft log entries, votes, and log metadata. Implementations must guarantee log continuity (no gaps in log indices) and serialization of writes (votes and log entries must be persisted in order).

#[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 – Persists new log entries and invokes the IOFlushed callback once entries are durable, allowing the Raft core to advance the commit index without blocking.
• save_vote – Persists the node's current vote (term and candidate).
• truncate_after – Removes log entries after a specific index during log reconciliation.
• purge – Removes obsolete log entries up to a specific index after snapshotting.

RaftStateMachine: Application State Management

Defined in openraft/src/storage/v2/raft_state_machine.rs, the RaftStateMachine trait manages the application-specific state machine, applying committed entries and handling snapshot creation and installation.

#[add_async_trait]
pub trait RaftStateMachine<C>: OptionalSend + OptionalSync + 'static
where C: RaftTypeConfig
{
    type SnapshotBuilder: RaftSnapshotBuilder<C>;

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

    async fn apply<Strm>(&mut self, entries: Strm) -> Result<(), io::Error>
    where Strm: Stream<Item = Result<EntryResponder<C>, io::Error>> + Unpin + OptionalSend;

    async fn try_create_snapshot_builder(&mut self, force: bool) -> Option<Self::SnapshotBuilder> { ... }
    async fn get_snapshot_builder(&mut self) -> Self::SnapshotBuilder; // deprecated
    async fn begin_receiving_snapshot(&mut self) -> Result<C::SnapshotData, io::Error>;
    async fn install_snapshot(&mut self, meta: &SnapshotMeta<C>, snapshot: C::SnapshotData) -> Result<(), io::Error>;
    async fn get_current_snapshot(&mut self) -> Result<Option<Snapshot<C>>, io::Error>;
}

Critical responsibilities include:

• apply – Consumes a stream of EntryResponder items, persisting the last applied log ID, executing business logic, and calling EntryResponder::send to return results to the Raft core.
• applied_state – Returns the last applied log ID and current membership configuration.
• Snapshot lifecycle – try_create_snapshot_builder initiates snapshot creation, while begin_receiving_snapshot and install_snapshot handle incoming snapshots from the leader.

RaftLogReader: Replication Read Interface

Located in openraft/src/storage/v2/raft_log_reader.rs, the RaftLogReader trait provides asynchronous, range-based reads of log entries for replication tasks. Implementations are typically returned by RaftLogStorage::get_log_reader and must support concurrent read access.

RaftSnapshotBuilder: Point-in-Time Snapshots

Defined in openraft/src/storage/v2/raft_snapshot_builder.rs, the RaftSnapshotBuilder trait allows the application to produce a read-only snapshot of the state machine. The builder is invoked when OpenRaft requires a snapshot for log compaction or for streaming to slow followers.

Supporting Types in the OpenRaft Storage Module

The storage module defines several auxiliary types that facilitate communication between the Raft core and storage implementations:

Type	Location	Purpose
LogState<C>	openraft/src/storage/log_state.rs	Holds last_purged_log_id and last_log_id boundaries.
Snapshot<C>	openraft/src/storage/snapshot.rs	Wraps snapshot data (C::SnapshotData) and its SnapshotMeta.
SnapshotMeta<C>	openraft/src/storage/snapshot_meta.rs	Stores the log ID and membership configuration of a snapshot.
IOFlushed<C>	openraft/src/storage/callback.rs	Callback invoked after log entries are safely persisted.
EntryResponder<C>	openraft/src/storage/v2/entry_responder.rs	Provides send(Result<ApplyResponse, io::Error>) for each applied entry.

All types are publicly re-exported from openraft::storage via openraft/src/storage/mod.rs.

Practical Implementation: In-Memory Storage Example

Below is a minimal in-memory implementation suitable for testing or prototyping. It stores logs, votes, and a simple key-value map; snapshots are produced via bincode serialization.

use openraft::{
    storage::{
        EntryResponder, IOFlushed, LogState, RaftLogReader, RaftLogStorage,
        RaftSnapshotBuilder, RaftStateMachine, Snapshot, SnapshotMeta,
    },
    RaftTypeConfig, StoredMembership,
};
use async_trait::async_trait;
use std::{
    collections::BTreeMap,
    io,
    sync::{Arc, Mutex},
};

#[derive(Clone, Default)]
struct MyConfig;
impl RaftTypeConfig for MyConfig {
    type D = String;
    type R = String;
    type NodeId = u64;
    type SnapshotData = Vec<u8>;
}

#[derive(Clone)]
struct LogEntry {
    log_id: openraft::LogId<u64>,
    entry: String,
}

#[derive(Default)]
struct Shared {
    log: Vec<LogEntry>,
    vote: Option<openraft::Vote<u64>>,
    committed: Option<openraft::LogId<u64>>,
    kv: BTreeMap<String, String>,
}

type SharedArc = Arc<Mutex<Shared>>;

struct InMemLogReader {
    shared: SharedArc,
}
#[async_trait]
impl RaftLogReader<MyConfig> for InMemLogReader {
    async fn get_log_entries(&mut self, start: u64, stop: u64) -> Result<Vec<openraft::Entry<MyConfig>>, io::Error> {
        let guard = self.shared.lock().unwrap();
        let slice = guard
            .log
            .iter()
            .filter(|e| e.log_id.index >= start && e.log_id.index < stop)
            .map(|e| openraft::Entry {
                log_id: e.log_id.clone(),
                payload: openraft::EntryPayload::Normal(openraft::EntryNormal { data: e.entry.clone() }),
                membership: None,
            })
            .collect();
        Ok(slice)
    }
}

struct InMemLogStorage {
    shared: SharedArc,
}
#[async_trait]
impl RaftLogStorage<MyConfig> for InMemLogStorage {
    type LogReader = InMemLogReader;

    async fn get_log_state(&mut self) -> Result<LogState<MyConfig>, io::Error> {
        let guard = self.shared.lock().unwrap();
        let last_log_id = guard.log.last().map(|e| e.log_id.clone());
        Ok(LogState {
            last_purged_log_id: None,
            last_log_id,
        })
    }

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

    async fn save_vote(&mut self, vote: &openraft::Vote<u64>) -> Result<(), io::Error> {
        let mut guard = self.shared.lock().unwrap();
        guard.vote = Some(vote.clone());
        Ok(())
    }

    async fn append<I>(&mut self, entries: I, callback: IOFlushed<MyConfig>) -> Result<(), io::Error>
    where I: IntoIterator<Item = openraft::Entry<MyConfig>> + openraft::OptionalSend,
          I::IntoIter: openraft::OptionalSend,
    {
        let mut guard = self.shared.lock().unwrap();
        for e in entries {
            if let openraft::EntryPayload::Normal(p) = e.payload {
                guard.log.push(LogEntry {
                    log_id: e.log_id,
                    entry: p.data,
                });
            }
        }
        (callback)(Ok(()));
        Ok(())
    }

    async fn truncate_after(&mut self, last_log_id: Option<openraft::LogId<u64>>) -> Result<(), io::Error> {
        let mut guard = self.shared.lock().unwrap();
        if let Some(id) = last_log_id {
            guard.log.retain(|e| e.log_id.index <= id.index);
        } else {
            guard.log.clear();
        }
        Ok(())
    }

    async fn purge(&mut self, log_id: openraft::LogId<u64>) -> Result<(), io::Error> {
        let mut guard = self.shared.lock().unwrap();
        guard.log.retain(|e| e.log_id.index > log_id.index);
        Ok(())
    }
}

struct InMemSnapshotBuilder {
    shared: SharedArc,
}
#[async_trait]
impl RaftSnapshotBuilder<MyConfig> for InMemSnapshotBuilder {
    async fn build(&mut self) -> Result<Snapshot<MyConfig>, io::Error> {
        let guard = self.shared.lock().unwrap();
        let data = bincode::serialize(&guard.kv).unwrap();
        let meta = SnapshotMeta {
            last_log_id: guard.log.last().map(|e| e.log_id.clone()),
            last_membership: StoredMembership::new_default(),
        };
        Ok(Snapshot { meta, snapshot: data })
    }
}

struct InMemStateMachine {
    shared: SharedArc,
}
#[async_trait]
impl RaftStateMachine<MyConfig> for InMemStateMachine {
    type SnapshotBuilder = InMemSnapshotBuilder;

    async fn applied_state(&mut self) -> Result<(Option<openraft::LogId<u64>>, StoredMembership<u64>), io::Error> {
        let guard = self.shared.lock().unwrap();
        Ok((guard.log.last().map(|e| e.log_id.clone()), StoredMembership::new_default()))
    }

    async fn apply<Strm>(&mut self, mut entries: Strm) -> Result<(), io::Error>
    where Strm: futures_util::Stream<Item = Result<EntryResponder<MyConfig>, io::Error>> + Unpin,
    {
        use futures_util::TryStreamExt;
        while let Some(entry_responder) = entries.try_next().await? {
            if let openraft::EntryPayload::Normal(p) = entry_responder.entry.payload {
                if let Some((k, v)) = p.data.split_once('=') {
                    let mut guard = self.shared.lock().unwrap();
                    guard.kv.insert(k.to_string(), v.to_string());
                }
            }
            entry_responder.send(Ok(()));
        }
        Ok(())
    }

    async fn try_create_snapshot_builder(&mut self, _force: bool) -> Option<Self::SnapshotBuilder> {
        Some(InMemSnapshotBuilder { shared: self.shared.clone() })
    }

    async fn begin_receiving_snapshot(&mut self) -> Result<Vec<u8>, io::Error> {
        Ok(Vec::new())
    }

    async fn install_snapshot(&mut self, _meta: &SnapshotMeta<MyConfig>, snapshot: Vec<u8>) -> Result<(), io::Error> {
        let mut guard = self.shared.lock().unwrap();
        guard.kv = bincode::deserialize(&snapshot).unwrap();
        Ok(())
    }

    async fn get_current_snapshot(&mut self) -> Result<Option<Snapshot<MyConfig>>, io::Error> {
        let mut builder = InMemSnapshotBuilder { shared: self.shared.clone() };
        builder.build().map(Some)
    }
}

This implementation demonstrates converting OpenRaft's generic Entry types into a simple key-value protocol, using bincode for snapshot serialization, and properly signaling durability through the IOFlushed callback.

Summary

• The OpenRaft storage module splits persistence into RaftLogStorage for WAL entries and RaftStateMachine for application state, enabling independent scaling and optimization.
• RaftLogStorage requires implementations to maintain log continuity and serialize writes through methods like append, truncate_after, and save_vote defined in openraft/src/storage/v2/raft_log_storage.rs.
• RaftStateMachine handles entry application via the apply method and snapshot lifecycle through try_create_snapshot_builder and install_snapshot, as specified in openraft/src/storage/v2/raft_state_machine.rs.
• Supporting types like IOFlushed, EntryResponder, SnapshotMeta, and LogState facilitate asynchronous communication between the Raft core and storage implementations.
• The in-memory implementation pattern demonstrates proper handling of IOFlushed callbacks, EntryResponder signaling, and snapshot serialization using bincode.

Frequently Asked Questions

What is the difference between RaftLogStorage and RaftStateMachine in OpenRaft?

RaftLogStorage persists the Raft log entries, votes, and committed indices to durable storage, acting as a write-ahead log. RaftStateMachine manages the application-level state, applying committed entries to your business logic and handling snapshot creation and installation. This separation allows the log to be stored on fast SSDs while the state machine might use a specialized database.

How does the IOFlushed callback work in RaftLogStorage::append?

The IOFlushed callback is invoked by the storage implementation once appended entries are guaranteed to be durable (fsynced to disk or replicated in a distributed store). This asynchronous notification allows the OpenRaft core to advance the commit index without blocking on I/O, improving throughput while maintaining safety guarantees.

What is the purpose of EntryResponder in the state machine apply method?

EntryResponder provides a channel to return the result of applying a log entry back to the Raft core, which then forwards it to the client. For each entry in the stream, the implementation must call entry_responder.send(Ok(())) (or an error) after persisting the state change, ensuring linearizable consistency between storage and client responses.

How do I implement log compaction and snapshotting?

Log compaction is triggered when OpenRaft calls RaftStateMachine::try_create_snapshot_builder, returning a RaftSnapshotBuilder that creates a point-in-time snapshot of your state. Once built, the snapshot metadata is stored, and you can purge old log entries via RaftLogStorage::purge. For receiving snapshots from leaders, implement begin_receiving_snapshot to allocate a buffer and install_snapshot to replace your state with the received data.

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