change: storage traits return io::Error

Issue #1457: Simplify storage trait error API.

Benefits: Simpler trait implementation - Storage code doesn't need to
know about `StorageError<C>` or generic type parameters in error handling.

## Changes

Storage trait methods now return `io::Error` instead of `StorageError<C>`:
- `RaftLogStorage`: `append()`, `truncate()`, `purge()`, `save_vote()`, `save_committed()`
- `RaftLogReader`: `try_get_log_entries()`, `limited_get_log_entries()`, `read_vote()`, `get_key_log_ids()`
- `RaftStateMachine`: `applied_state()`, `apply()`, `get_current_snapshot()`, `begin_receiving_snapshot()`, `install_snapshot()`
- `RaftSnapshotBuilder`: `build_snapshot()`

Added internal `StorageIOResult` trait for error conversion:
- Provides 14 methods like `.sto_read_logs()`, `.sto_write_snapshot()`, etc.
- Converts `io::Error` to `StorageError<C>` with proper context at Openraft boundaries

- Fix: #1457

Upgrade tip:

Storage implementors should update trait methods to return `io::Error`:

```rust
// Before
async fn read_vote(&mut self) -> Result<Option<Vote>, StorageError<C>> {
    self.vote.read().map_err(|e| StorageError::read_vote(&e))
}

// After
async fn read_vote(&mut self) -> Result<Option<Vote>, io::Error> {
    self.vote.read()
}
```

Return `io::Error` directly - Openraft adds context automatically.
The method body may leave unchanged: `io::Error` can be converted to
`StorageError` with a simple `?`.

Author: drmingdrmer

cluster_benchmark/tests/benchmark/store.rs

@@ -3,6 +3,7 @@ mod test;
 use std::collections::BTreeMap;
 use std::fmt;
 use std::fmt::Debug;
+use std::io;
 use std::io::Cursor;
 use std::ops::RangeBounds;
 use std::sync::atomic::AtomicU64;
@@ -24,7 +25,6 @@ use openraft::Entry;
 use openraft::EntryPayload;
 use openraft::OptionalSend;
 use openraft::SnapshotMeta;
-use openraft::StorageError;
 use openraft::StoredMembership;
 use openraft::Vote;
 use serde::Deserialize;
@@ -121,7 +121,7 @@ impl RaftLogReader<TypeConfig> for Arc<LogStore> {
     async fn try_get_log_entries<RB: RangeBounds<u64> + Clone + Debug + OptionalSend>(
         &mut self,
         range: RB,
-    ) -> Result<Vec<Entry<TypeConfig>>, StorageError<TypeConfig>> {
+    ) -> Result<Vec<Entry<TypeConfig>>, io::Error> {
         let mut entries = vec![];
         {
             let log = self.log.read().await;
@@ -133,22 +133,22 @@ impl RaftLogReader<TypeConfig> for Arc<LogStore> {
         Ok(entries)
     }
 
-    async fn read_vote(&mut self) -> Result<Option<Vote<TypeConfig>>, StorageError<TypeConfig>> {
+    async fn read_vote(&mut self) -> Result<Option<Vote<TypeConfig>>, io::Error> {
         Ok(self.vote.read().await.clone())
     }
 }
 
 impl RaftSnapshotBuilder<TypeConfig> for Arc<StateMachineStore> {
     #[tracing::instrument(level = "trace", skip(self))]
-    async fn build_snapshot(&mut self) -> Result<Snapshot<TypeConfig>, StorageError<TypeConfig>> {
+    async fn build_snapshot(&mut self) -> Result<Snapshot<TypeConfig>, io::Error> {
         let data;
         let last_applied_log;
         let last_membership;
 
         {
             // Serialize the data of the state machine.
             let sm = self.sm.read().await;
-            data = serde_json::to_vec(&*sm).map_err(|e| StorageError::read_state_machine(&e))?;
+            data = serde_json::to_vec(&*sm).map_err(|e| io::Error::new(io::ErrorKind::Other, e))?;
 
             last_applied_log = sm.last_applied_log;
             last_membership = sm.last_membership.clone();
@@ -190,7 +190,7 @@ impl RaftSnapshotBuilder<TypeConfig> for Arc<StateMachineStore> {
 }
 
 impl RaftLogStorage<TypeConfig> for Arc<LogStore> {
-    async fn get_log_state(&mut self) -> Result<LogState<TypeConfig>, StorageError<TypeConfig>> {
+    async fn get_log_state(&mut self) -> Result<LogState<TypeConfig>, io::Error> {
         let log = self.log.read().await;
         let last_serialized = log.iter().rev().next().map(|(_, ent)| ent);
 
@@ -213,22 +213,22 @@ impl RaftLogStorage<TypeConfig> for Arc<LogStore> {
     }
 
     #[tracing::instrument(level = "trace", skip(self))]
-    async fn save_vote(&mut self, vote: &Vote<TypeConfig>) -> Result<(), StorageError<TypeConfig>> {
+    async fn save_vote(&mut self, vote: &Vote<TypeConfig>) -> Result<(), io::Error> {
         let mut v = self.vote.write().await;
         *v = Some(*vote);
         Ok(())
     }
 
     #[tracing::instrument(level = "debug", skip(self))]
-    async fn truncate(&mut self, log_id: LogIdOf<TypeConfig>) -> Result<(), StorageError<TypeConfig>> {
+    async fn truncate(&mut self, log_id: LogIdOf<TypeConfig>) -> Result<(), io::Error> {
         let mut log = self.log.write().await;
         log.split_off(&log_id.index());
 
         Ok(())
     }
 
     #[tracing::instrument(level = "debug", skip_all)]
-    async fn purge(&mut self, log_id: LogIdOf<TypeConfig>) -> Result<(), StorageError<TypeConfig>> {
+    async fn purge(&mut self, log_id: LogIdOf<TypeConfig>) -> Result<(), io::Error> {
         {
             let mut p = self.last_purged_log_id.write().await;
             *p = Some(log_id);
@@ -241,7 +241,7 @@ impl RaftLogStorage<TypeConfig> for Arc<LogStore> {
     }
 
     #[tracing::instrument(level = "trace", skip_all)]
-    async fn append<I>(&mut self, entries: I, callback: IOFlushed<TypeConfig>) -> Result<(), StorageError<TypeConfig>>
+    async fn append<I>(&mut self, entries: I, callback: IOFlushed<TypeConfig>) -> Result<(), io::Error>
     where I: IntoIterator<Item = Entry<TypeConfig>> + Send {
         {
             let mut log = self.log.write().await;
@@ -261,12 +261,12 @@ impl RaftLogStorage<TypeConfig> for Arc<LogStore> {
 impl RaftStateMachine<TypeConfig> for Arc<StateMachineStore> {
     async fn applied_state(
         &mut self,
-    ) -> Result<(Option<LogIdOf<TypeConfig>>, StoredMembership<TypeConfig>), StorageError<TypeConfig>> {
+    ) -> Result<(Option<LogIdOf<TypeConfig>>, StoredMembership<TypeConfig>), io::Error> {
         let sm = self.sm.read().await;
         Ok((sm.last_applied_log, sm.last_membership.clone()))
     }
 
-    async fn apply<I>(&mut self, entries: I) -> Result<(), StorageError<TypeConfig>>
+    async fn apply<I>(&mut self, entries: I) -> Result<(), io::Error>
     where
         I: IntoIterator<Item = EntryResponder<TypeConfig>> + Send,
         I::IntoIter: Send,
@@ -291,7 +291,7 @@ impl RaftStateMachine<TypeConfig> for Arc<StateMachineStore> {
     }
 
     #[tracing::instrument(level = "trace", skip(self))]
-    async fn begin_receiving_snapshot(&mut self) -> Result<SnapshotDataOf<TypeConfig>, StorageError<TypeConfig>> {
+    async fn begin_receiving_snapshot(&mut self) -> Result<SnapshotDataOf<TypeConfig>, io::Error> {
         Ok(Cursor::new(Vec::new()))
     }
 
@@ -300,16 +300,16 @@ impl RaftStateMachine<TypeConfig> for Arc<StateMachineStore> {
         &mut self,
         meta: &SnapshotMeta<TypeConfig>,
         snapshot: SnapshotDataOf<TypeConfig>,
-    ) -> Result<(), StorageError<TypeConfig>> {
+    ) -> Result<(), io::Error> {
         let new_snapshot = StoredSnapshot {
             meta: meta.clone(),
             data: snapshot.into_inner(),
         };
 
         // Update the state machine.
         {
-            let new_sm: StateMachine = serde_json::from_slice(&new_snapshot.data)
-                .map_err(|e| StorageError::read_snapshot(Some(new_snapshot.meta.signature()), &e))?;
+            let new_sm: StateMachine =
+                serde_json::from_slice(&new_snapshot.data).map_err(|e| io::Error::new(io::ErrorKind::Other, e))?;
             let mut sm = self.sm.write().await;
             *sm = new_sm;
         }
@@ -321,7 +321,7 @@ impl RaftStateMachine<TypeConfig> for Arc<StateMachineStore> {
     }
 
     #[tracing::instrument(level = "trace", skip(self))]
-    async fn get_current_snapshot(&mut self) -> Result<Option<Snapshot<TypeConfig>>, StorageError<TypeConfig>> {
+    async fn get_current_snapshot(&mut self) -> Result<Option<Snapshot<TypeConfig>>, io::Error> {
         match &*self.current_snapshot.read().await {
             Some(snapshot) => {
                 let data = snapshot.data.clone();

examples/mem-log/src/log_store.rs

@@ -3,6 +3,7 @@
 
 use std::collections::BTreeMap;
 use std::fmt::Debug;
+use std::io;
 use std::ops::RangeBounds;
 use std::sync::Arc;
 
@@ -12,7 +13,6 @@ use openraft::entry::RaftEntry;
 use openraft::storage::IOFlushed;
 use openraft::LogState;
 use openraft::RaftTypeConfig;
-use openraft::StorageError;
 use tokio::sync::Mutex;
 
 /// RaftLogStore implementation with a in-memory storage
@@ -51,15 +51,15 @@ impl<C: RaftTypeConfig> LogStoreInner<C> {
     async fn try_get_log_entries<RB: RangeBounds<u64> + Clone + Debug>(
         &mut self,
         range: RB,
-    ) -> Result<Vec<C::Entry>, StorageError<C>>
+    ) -> Result<Vec<C::Entry>, io::Error>
     where
         C::Entry: Clone,
     {
         let response = self.log.range(range.clone()).map(|(_, val)| val.clone()).collect::<Vec<_>>();
         Ok(response)
     }
 
-    async fn get_log_state(&mut self) -> Result<LogState<C>, StorageError<C>> {
+    async fn get_log_state(&mut self) -> Result<LogState<C>, io::Error> {
         let last = self.log.iter().next_back().map(|(_, ent)| ent.log_id());
 
         let last_purged = self.last_purged_log_id.clone();
@@ -75,25 +75,25 @@ impl<C: RaftTypeConfig> LogStoreInner<C> {
         })
     }
 
-    async fn save_committed(&mut self, committed: Option<LogIdOf<C>>) -> Result<(), StorageError<C>> {
+    async fn save_committed(&mut self, committed: Option<LogIdOf<C>>) -> Result<(), io::Error> {
         self.committed = committed;
         Ok(())
     }
 
-    async fn read_committed(&mut self) -> Result<Option<LogIdOf<C>>, StorageError<C>> {
+    async fn read_committed(&mut self) -> Result<Option<LogIdOf<C>>, io::Error> {
         Ok(self.committed.clone())
     }
 
-    async fn save_vote(&mut self, vote: &VoteOf<C>) -> Result<(), StorageError<C>> {
+    async fn save_vote(&mut self, vote: &VoteOf<C>) -> Result<(), io::Error> {
         self.vote = Some(vote.clone());
         Ok(())
     }
 
-    async fn read_vote(&mut self) -> Result<Option<VoteOf<C>>, StorageError<C>> {
+    async fn read_vote(&mut self) -> Result<Option<VoteOf<C>>, io::Error> {
         Ok(self.vote.clone())
     }
 
-    async fn append<I>(&mut self, entries: I, callback: IOFlushed<C>) -> Result<(), StorageError<C>>
+    async fn append<I>(&mut self, entries: I, callback: IOFlushed<C>) -> Result<(), io::Error>
     where I: IntoIterator<Item = C::Entry> {
         // Simple implementation that calls the flush-before-return `append_to_log`.
         for entry in entries {
@@ -104,7 +104,7 @@ impl<C: RaftTypeConfig> LogStoreInner<C> {
         Ok(())
     }
 
-    async fn truncate(&mut self, log_id: LogIdOf<C>) -> Result<(), StorageError<C>> {
+    async fn truncate(&mut self, log_id: LogIdOf<C>) -> Result<(), io::Error> {
         let keys = self.log.range(log_id.index()..).map(|(k, _v)| *k).collect::<Vec<_>>();
         for key in keys {
             self.log.remove(&key);
@@ -113,7 +113,7 @@ impl<C: RaftTypeConfig> LogStoreInner<C> {
         Ok(())
     }
 
-    async fn purge(&mut self, log_id: LogIdOf<C>) -> Result<(), StorageError<C>> {
+    async fn purge(&mut self, log_id: LogIdOf<C>) -> Result<(), io::Error> {
         {
             let ld = &mut self.last_purged_log_id;
             assert!(ld.as_ref() <= Some(&log_id));
@@ -133,6 +133,7 @@ impl<C: RaftTypeConfig> LogStoreInner<C> {
 
 mod impl_log_store {
     use std::fmt::Debug;
+    use std::io;
     use std::ops::RangeBounds;
 
     use openraft::alias::LogIdOf;
@@ -142,7 +143,6 @@ mod impl_log_store {
     use openraft::LogState;
     use openraft::RaftLogReader;
     use openraft::RaftTypeConfig;
-    use openraft::StorageError;
 
     use crate::log_store::LogStore;
 
@@ -152,12 +152,12 @@ mod impl_log_store {
         async fn try_get_log_entries<RB: RangeBounds<u64> + Clone + Debug>(
             &mut self,
             range: RB,
-        ) -> Result<Vec<C::Entry>, StorageError<C>> {
+        ) -> Result<Vec<C::Entry>, io::Error> {
             let mut inner = self.inner.lock().await;
             inner.try_get_log_entries(range).await
         }
 
-        async fn read_vote(&mut self) -> Result<Option<VoteOf<C>>, StorageError<C>> {
+        async fn read_vote(&mut self) -> Result<Option<VoteOf<C>>, io::Error> {
             let mut inner = self.inner.lock().await;
             inner.read_vote().await
         }
@@ -168,38 +168,38 @@ mod impl_log_store {
     {
         type LogReader = Self;
 
-        async fn get_log_state(&mut self) -> Result<LogState<C>, StorageError<C>> {
+        async fn get_log_state(&mut self) -> Result<LogState<C>, io::Error> {
             let mut inner = self.inner.lock().await;
             inner.get_log_state().await
         }
 
-        async fn save_committed(&mut self, committed: Option<LogIdOf<C>>) -> Result<(), StorageError<C>> {
+        async fn save_committed(&mut self, committed: Option<LogIdOf<C>>) -> Result<(), io::Error> {
             let mut inner = self.inner.lock().await;
             inner.save_committed(committed).await
         }
 
-        async fn read_committed(&mut self) -> Result<Option<LogIdOf<C>>, StorageError<C>> {
+        async fn read_committed(&mut self) -> Result<Option<LogIdOf<C>>, io::Error> {
             let mut inner = self.inner.lock().await;
             inner.read_committed().await
         }
 
-        async fn save_vote(&mut self, vote: &VoteOf<C>) -> Result<(), StorageError<C>> {
+        async fn save_vote(&mut self, vote: &VoteOf<C>) -> Result<(), io::Error> {
             let mut inner = self.inner.lock().await;
             inner.save_vote(vote).await
         }
 
-        async fn append<I>(&mut self, entries: I, callback: IOFlushed<C>) -> Result<(), StorageError<C>>
+        async fn append<I>(&mut self, entries: I, callback: IOFlushed<C>) -> Result<(), io::Error>
         where I: IntoIterator<Item = C::Entry> {
             let mut inner = self.inner.lock().await;
             inner.append(entries, callback).await
         }
 
-        async fn truncate(&mut self, log_id: LogIdOf<C>) -> Result<(), StorageError<C>> {
+        async fn truncate(&mut self, log_id: LogIdOf<C>) -> Result<(), io::Error> {
             let mut inner = self.inner.lock().await;
             inner.truncate(log_id).await
         }
 
-        async fn purge(&mut self, log_id: LogIdOf<C>) -> Result<(), StorageError<C>> {
+        async fn purge(&mut self, log_id: LogIdOf<C>) -> Result<(), io::Error> {
             let mut inner = self.inner.lock().await;
             inner.purge(log_id).await
         }

examples/raft-kv-memstore-grpc/src/store/mod.rs

@@ -1,4 +1,5 @@
 use std::fmt::Debug;
+use std::io;
 use std::sync::Arc;
 use std::sync::Mutex;
 
@@ -48,7 +49,7 @@ impl StateMachineStore {
 
 impl RaftSnapshotBuilder<TypeConfig> for Arc<StateMachineStore> {
     #[tracing::instrument(level = "trace", skip(self))]
-    async fn build_snapshot(&mut self) -> Result<Snapshot, StorageError> {
+    async fn build_snapshot(&mut self) -> Result<Snapshot, io::Error> {
         let data;
         let last_applied: Option<LogId>;
         let last_membership;
@@ -97,7 +98,7 @@ impl RaftSnapshotBuilder<TypeConfig> for Arc<StateMachineStore> {
 impl RaftStateMachine<TypeConfig> for Arc<StateMachineStore> {
     type SnapshotBuilder = Self;
 
-    async fn applied_state(&mut self) -> Result<(Option<LogId>, StoredMembership), StorageError> {
+    async fn applied_state(&mut self) -> Result<(Option<LogId>, StoredMembership), io::Error> {
         let sm = self.state_machine.lock().unwrap();
 
         let last_applied = sm.last_applied.map(|x| x.into());
@@ -111,7 +112,7 @@ impl RaftStateMachine<TypeConfig> for Arc<StateMachineStore> {
     }
 
     #[tracing::instrument(level = "trace", skip(self, entries))]
-    async fn apply<I>(&mut self, entries: I) -> Result<(), StorageError>
+    async fn apply<I>(&mut self, entries: I) -> Result<(), io::Error>
     where
         I: IntoIterator<Item = EntryResponder<TypeConfig>>,
         I::IntoIter: Send,
@@ -144,12 +145,12 @@ impl RaftStateMachine<TypeConfig> for Arc<StateMachineStore> {
     }
 
     #[tracing::instrument(level = "trace", skip(self))]
-    async fn begin_receiving_snapshot(&mut self) -> Result<SnapshotData, StorageError> {
+    async fn begin_receiving_snapshot(&mut self) -> Result<SnapshotData, io::Error> {
         Ok(Default::default())
     }
 
     #[tracing::instrument(level = "trace", skip(self, snapshot))]
-    async fn install_snapshot(&mut self, meta: &SnapshotMeta, snapshot: SnapshotData) -> Result<(), StorageError> {
+    async fn install_snapshot(&mut self, meta: &SnapshotMeta, snapshot: SnapshotData) -> Result<(), io::Error> {
         tracing::info!("install snapshot");
 
         let new_snapshot = StoredSnapshot {
@@ -160,7 +161,7 @@ impl RaftStateMachine<TypeConfig> for Arc<StateMachineStore> {
         // Update the state machine.
         {
             let d: pb::StateMachineData = prost::Message::decode(new_snapshot.data.as_ref())
-                .map_err(|e| StorageError::read_snapshot(None, &e))?;
+                .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
 
             let mut state_machine = self.state_machine.lock().unwrap();
             *state_machine = d;
@@ -173,7 +174,7 @@ impl RaftStateMachine<TypeConfig> for Arc<StateMachineStore> {
     }
 
     #[tracing::instrument(level = "trace", skip(self))]
-    async fn get_current_snapshot(&mut self) -> Result<Option<Snapshot>, StorageError> {
+    async fn get_current_snapshot(&mut self) -> Result<Option<Snapshot>, io::Error> {
         match &*self.current_snapshot.lock().unwrap() {
             Some(snapshot) => {
                 let data = snapshot.data.clone();