diff --git a/cluster_benchmark/tests/benchmark/store.rs b/cluster_benchmark/tests/benchmark/store.rs
index f6d5360cc..2b3b9bf73 100644
--- a/cluster_benchmark/tests/benchmark/store.rs
+++ b/cluster_benchmark/tests/benchmark/store.rs
@@ -19,7 +19,6 @@ use openraft::Entry;
 use openraft::EntryPayload;
 use openraft::LogId;
 use openraft::OptionalSend;
-use openraft::OptionalSync;
 use openraft::RaftLogId;
 use openraft::RaftTypeConfig;
 use openraft::SnapshotMeta;
@@ -225,8 +224,14 @@ impl RaftLogStorage<TypeConfig> for Arc<LogStore> {
     }
 
     #[tracing::instrument(level = "trace", skip_all)]
-    async fn append<I>(&mut self, entries: I, callback: LogFlushed<NodeId>) -> Result<(), StorageError<NodeId>>
-    where I: IntoIterator<Item = Entry<TypeConfig>> + Send {
+    async fn append<I>(
+        &mut self,
+        entries: I,
+        callback: LogFlushed<TypeConfig>,
+    ) -> Result<(), StorageError<NodeId>>
+    where
+        I: IntoIterator<Item = Entry<TypeConfig>> + Send,
+    {
         {
             let mut log = self.log.write().await;
             log.extend(entries.into_iter().map(|entry| (entry.get_log_id().index, entry)));
diff --git a/examples/memstore/src/log_store.rs b/examples/memstore/src/log_store.rs
index 25715e781..867465ac6 100644
--- a/examples/memstore/src/log_store.rs
+++ b/examples/memstore/src/log_store.rs
@@ -93,7 +93,7 @@ impl<C: RaftTypeConfig> LogStoreInner<C> {
         Ok(self.vote)
     }
 
-    async fn append<I>(&mut self, entries: I, callback: LogFlushed<C::NodeId>) -> Result<(), StorageError<C::NodeId>>
+    async fn append<I>(&mut self, entries: I, callback: LogFlushed<C>) -> Result<(), StorageError<C::NodeId>>
     where I: IntoIterator<Item = C::Entry> {
         // Simple implementation that calls the flush-before-return `append_to_log`.
         for entry in entries {
@@ -188,14 +188,8 @@ mod impl_log_store {
             inner.read_vote().await
         }
 
-        async fn append<I>(
-            &mut self,
-            entries: I,
-            callback: LogFlushed<C::NodeId>,
-        ) -> Result<(), StorageError<C::NodeId>>
-        where
-            I: IntoIterator<Item = C::Entry>,
-        {
+        async fn append<I>(&mut self, entries: I, callback: LogFlushed<C>) -> Result<(), StorageError<C::NodeId>>
+        where I: IntoIterator<Item = C::Entry> {
             let mut inner = self.inner.lock().await;
             inner.append(entries, callback).await
         }
diff --git a/examples/raft-kv-memstore-singlethreaded/src/store.rs b/examples/raft-kv-memstore-singlethreaded/src/store.rs
index 425130e19..227fba770 100644
--- a/examples/raft-kv-memstore-singlethreaded/src/store.rs
+++ b/examples/raft-kv-memstore-singlethreaded/src/store.rs
@@ -321,7 +321,7 @@ impl RaftLogStorage<TypeConfig> for Rc<LogStore> {
     }
 
     #[tracing::instrument(level = "trace", skip(self, entries, callback))]
-    async fn append<I>(&mut self, entries: I, callback: LogFlushed<NodeId>) -> Result<(), StorageError<NodeId>>
+    async fn append<I>(&mut self, entries: I, callback: LogFlushed<TypeConfig>) -> Result<(), StorageError<NodeId>>
     where I: IntoIterator<Item = Entry<TypeConfig>> {
         // Simple implementation that calls the flush-before-return `append_to_log`.
         let mut log = self.log.borrow_mut();
diff --git a/examples/raft-kv-rocksdb/src/store.rs b/examples/raft-kv-rocksdb/src/store.rs
index bd678d04a..4c1bac8b9 100644
--- a/examples/raft-kv-rocksdb/src/store.rs
+++ b/examples/raft-kv-rocksdb/src/store.rs
@@ -436,7 +436,7 @@ impl RaftLogStorage<TypeConfig> for LogStore {
     }
 
     #[tracing::instrument(level = "trace", skip_all)]
-    async fn append<I>(&mut self, entries: I, callback: LogFlushed<NodeId>) -> StorageResult<()>
+    async fn append<I>(&mut self, entries: I, callback: LogFlushed<TypeConfig>) -> StorageResult<()>
     where
         I: IntoIterator<Item = Entry<TypeConfig>> + Send,
         I::IntoIter: Send,
diff --git a/openraft/src/async_runtime.rs b/openraft/src/async_runtime.rs
index 5e9c73e2a..c602d48a7 100644
--- a/openraft/src/async_runtime.rs
+++ b/openraft/src/async_runtime.rs
@@ -18,7 +18,7 @@ use crate::TokioInstant;
 /// ## Note
 ///
 /// The default asynchronous runtime is `tokio`.
-pub trait AsyncRuntime: Debug + Default + OptionalSend + OptionalSync + 'static {
+pub trait AsyncRuntime: Debug + Default + PartialEq + Eq + OptionalSend + OptionalSync + 'static {
     /// The error type of [`Self::JoinHandle`].
     type JoinError: Debug + Display + OptionalSend;
 
@@ -44,6 +44,18 @@ pub trait AsyncRuntime: Debug + Default + OptionalSend + OptionalSync + 'static
     /// Type of a thread-local random number generator.
     type ThreadLocalRng: rand::Rng;
 
+    /// Type of a `oneshot` sender.
+    type OneshotSender<T: OptionalSend>: AsyncOneshotSendExt<T> + OptionalSend + OptionalSync + Debug + Sized;
+
+    /// Type of a `oneshot` receiver error.
+    type OneshotReceiverError: std::error::Error + OptionalSend;
+
+    /// Type of a `oneshot` receiver.
+    type OneshotReceiver<T: OptionalSend>: OptionalSend
+        + OptionalSync
+        + Future<Output = Result<T, Self::OneshotReceiverError>>
+        + Unpin;
+
     /// Spawn a new task.
     fn spawn<T>(future: T) -> Self::JoinHandle<T::Output>
     where
@@ -72,12 +84,24 @@ pub trait AsyncRuntime: Debug + Default + OptionalSend + OptionalSync + 'static
     /// This is a per-thread instance, which cannot be shared across threads or
     /// sent to another thread.
     fn thread_rng() -> Self::ThreadLocalRng;
+
+    /// Creates a new one-shot channel for sending single values.
+    ///
+    /// The function returns separate "send" and "receive" handles. The `Sender`
+    /// handle is used by the producer to send the value. The `Receiver` handle is
+    /// used by the consumer to receive the value.
+    ///
+    /// Each handle can be used on separate tasks.
+    fn oneshot<T>() -> (Self::OneshotSender<T>, Self::OneshotReceiver<T>)
+    where T: OptionalSend;
 }
 
 /// `Tokio` is the default asynchronous executor.
-#[derive(Debug, Default)]
+#[derive(Debug, Default, PartialEq, Eq)]
 pub struct TokioRuntime;
 
+pub struct TokioOneShotSender<T: OptionalSend>(pub tokio::sync::oneshot::Sender<T>);
+
 impl AsyncRuntime for TokioRuntime {
     type JoinError = tokio::task::JoinError;
     type JoinHandle<T: OptionalSend + 'static> = tokio::task::JoinHandle<T>;
@@ -86,6 +110,9 @@ impl AsyncRuntime for TokioRuntime {
     type TimeoutError = tokio::time::error::Elapsed;
     type Timeout<R, T: Future<Output = R> + OptionalSend> = tokio::time::Timeout<T>;
     type ThreadLocalRng = rand::rngs::ThreadRng;
+    type OneshotSender<T: OptionalSend> = TokioOneShotSender<T>;
+    type OneshotReceiver<T: OptionalSend> = tokio::sync::oneshot::Receiver<T>;
+    type OneshotReceiverError = tokio::sync::oneshot::error::RecvError;
 
     #[inline]
     fn spawn<T>(future: T) -> Self::JoinHandle<T::Output>
@@ -132,4 +159,36 @@ impl AsyncRuntime for TokioRuntime {
     fn thread_rng() -> Self::ThreadLocalRng {
         rand::thread_rng()
     }
+
+    #[inline]
+    fn oneshot<T>() -> (Self::OneshotSender<T>, Self::OneshotReceiver<T>)
+    where T: OptionalSend {
+        let (tx, rx) = tokio::sync::oneshot::channel();
+        (TokioOneShotSender(tx), rx)
+    }
+}
+
+pub trait AsyncOneshotSendExt<T>: Unpin {
+    /// Attempts to send a value on this channel, returning it back if it could
+    /// not be sent.
+    ///
+    /// This method consumes `self` as only one value may ever be sent on a `oneshot`
+    /// channel. It is not marked async because sending a message to an `oneshot`
+    /// channel never requires any form of waiting.  Because of this, the `send`
+    /// method can be used in both synchronous and asynchronous code without
+    /// problems.
+    fn send(self, t: T) -> Result<(), T>;
+}
+
+impl<T: OptionalSend> AsyncOneshotSendExt<T> for TokioOneShotSender<T> {
+    #[inline]
+    fn send(self, t: T) -> Result<(), T> {
+        self.0.send(t)
+    }
+}
+
+impl<T: OptionalSend> Debug for TokioOneShotSender<T> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_tuple("TokioSendWrapper").finish()
+    }
 }
diff --git a/openraft/src/core/raft_core.rs b/openraft/src/core/raft_core.rs
index 9ce86f542..921a08fb7 100644
--- a/openraft/src/core/raft_core.rs
+++ b/openraft/src/core/raft_core.rs
@@ -15,12 +15,12 @@ use futures::TryFutureExt;
 use maplit::btreeset;
 use tokio::select;
 use tokio::sync::mpsc;
-use tokio::sync::oneshot;
 use tokio::sync::watch;
 use tracing::Instrument;
 use tracing::Level;
 use tracing::Span;
 
+use crate::async_runtime::AsyncOneshotSendExt;
 use crate::config::Config;
 use crate::config::RuntimeConfig;
 use crate::core::balancer::Balancer;
@@ -215,7 +215,10 @@ where
     SM: RaftStateMachine<C>,
 {
     /// The main loop of the Raft protocol.
-    pub(crate) async fn main(mut self, rx_shutdown: oneshot::Receiver<()>) -> Result<(), Fatal<C::NodeId>> {
+    pub(crate) async fn main(
+        mut self,
+        rx_shutdown: <C::AsyncRuntime as AsyncRuntime>::OneshotReceiver<()>,
+    ) -> Result<(), Fatal<C::NodeId>> {
         let span = tracing::span!(parent: &self.span, Level::DEBUG, "main");
         let res = self.do_main(rx_shutdown).instrument(span).await;
 
@@ -239,7 +242,10 @@ where
     }
 
     #[tracing::instrument(level="trace", skip_all, fields(id=display(self.id), cluster=%self.config.cluster_name))]
-    async fn do_main(&mut self, rx_shutdown: oneshot::Receiver<()>) -> Result<(), Fatal<C::NodeId>> {
+    async fn do_main(
+        &mut self,
+        rx_shutdown: <C::AsyncRuntime as AsyncRuntime>::OneshotReceiver<()>,
+    ) -> Result<(), Fatal<C::NodeId>> {
         tracing::debug!("raft node is initializing");
 
         self.engine.startup();
@@ -432,7 +438,7 @@ where
         &mut self,
         changes: ChangeMembers<C::NodeId, C::Node>,
         retain: bool,
-        tx: ResultSender<ClientWriteResponse<C>, ClientWriteError<C::NodeId, C::Node>>,
+        tx: ResultSender<C, ClientWriteResponse<C>, ClientWriteError<C::NodeId, C::Node>>,
     ) {
         let res = self.engine.state.membership_state.change_handler().apply(changes, retain);
         let new_membership = match res {
@@ -593,7 +599,7 @@ where
     pub(crate) fn handle_initialize(
         &mut self,
         member_nodes: BTreeMap<C::NodeId, C::Node>,
-        tx: ResultSender<(), InitializeError<C::NodeId, C::Node>>,
+        tx: ResultSender<C, (), InitializeError<C::NodeId, C::Node>>,
     ) {
         tracing::debug!(member_nodes = debug(&member_nodes), "{}", func_name!());
 
@@ -616,7 +622,7 @@ where
 
     /// Reject a request due to the Raft node being in a state which prohibits the request.
     #[tracing::instrument(level = "trace", skip(self, tx))]
-    pub(crate) fn reject_with_forward_to_leader<T, E>(&self, tx: ResultSender<T, E>)
+    pub(crate) fn reject_with_forward_to_leader<T: OptionalSend, E: OptionalSend>(&self, tx: ResultSender<C, T, E>)
     where E: From<ForwardToLeader<C::NodeId, C::Node>> {
         let mut leader_id = self.current_leader();
         let leader_node = self.get_leader_node(leader_id);
@@ -680,7 +686,7 @@ where
     {
         tracing::debug!("append_to_log");
 
-        let (tx, rx) = oneshot::channel();
+        let (tx, rx) = C::AsyncRuntime::oneshot();
         let callback = LogFlushed::new(Some(last_log_id), tx);
         self.log_store.append(entries, callback).await?;
         rx.await
@@ -865,7 +871,10 @@ where
 
     /// Run an event handling loop
     #[tracing::instrument(level="debug", skip_all, fields(id=display(self.id)))]
-    async fn runtime_loop(&mut self, mut rx_shutdown: oneshot::Receiver<()>) -> Result<(), Fatal<C::NodeId>> {
+    async fn runtime_loop(
+        &mut self,
+        mut rx_shutdown: <C::AsyncRuntime as AsyncRuntime>::OneshotReceiver<()>,
+    ) -> Result<(), Fatal<C::NodeId>> {
         // Ratio control the ratio of number of RaftMsg to process to number of Notify to process.
         let mut balancer = Balancer::new(10_000);
 
@@ -1067,7 +1076,7 @@ where
     }
 
     #[tracing::instrument(level = "debug", skip_all)]
-    pub(super) fn handle_vote_request(&mut self, req: VoteRequest<C::NodeId>, tx: VoteTx<C::NodeId>) {
+    pub(super) fn handle_vote_request(&mut self, req: VoteRequest<C::NodeId>, tx: VoteTx<C>) {
         tracing::info!(req = display(req.summary()), func = func_name!());
 
         let resp = self.engine.handle_vote_req(req);
@@ -1078,11 +1087,7 @@ where
     }
 
     #[tracing::instrument(level = "debug", skip_all)]
-    pub(super) fn handle_append_entries_request(
-        &mut self,
-        req: AppendEntriesRequest<C>,
-        tx: AppendEntriesTx<C::NodeId>,
-    ) {
+    pub(super) fn handle_append_entries_request(&mut self, req: AppendEntriesRequest<C>, tx: AppendEntriesTx<C>) {
         tracing::debug!(req = display(req.summary()), func = func_name!());
 
         let is_ok = self.engine.handle_append_entries(&req.vote, req.prev_log_id, req.entries, Some(tx));
@@ -1657,7 +1662,7 @@ where
 
                             // Create a channel to let state machine worker to send the snapshot and the replication
                             // worker to receive it.
-                            let (tx, rx) = oneshot::channel();
+                            let (tx, rx) = C::AsyncRuntime::oneshot();
 
                             let cmd = sm::Command::get_snapshot(tx);
                             self.sm_handle
diff --git a/openraft/src/core/raft_msg/external_command.rs b/openraft/src/core/raft_msg/external_command.rs
index 17c983b3c..5714df39c 100644
--- a/openraft/src/core/raft_msg/external_command.rs
+++ b/openraft/src/core/raft_msg/external_command.rs
@@ -23,7 +23,7 @@ pub(crate) enum ExternalCommand<C: RaftTypeConfig> {
     Snapshot,
 
     /// Get a snapshot from the state machine, send back via a oneshot::Sender.
-    GetSnapshot { tx: ResultSender<Option<Snapshot<C>>> },
+    GetSnapshot { tx: ResultSender<C, Option<Snapshot<C>>> },
 
     /// Purge logs covered by a snapshot up to a specified index.
     ///
diff --git a/openraft/src/core/raft_msg/mod.rs b/openraft/src/core/raft_msg/mod.rs
index b6679d43a..f4ea72cdc 100644
--- a/openraft/src/core/raft_msg/mod.rs
+++ b/openraft/src/core/raft_msg/mod.rs
@@ -1,7 +1,5 @@
 use std::collections::BTreeMap;
 
-use tokio::sync::oneshot;
-
 use crate::core::raft_msg::external_command::ExternalCommand;
 use crate::error::CheckIsLeaderError;
 use crate::error::ClientWriteError;
@@ -15,10 +13,13 @@ use crate::raft::ClientWriteResponse;
 use crate::raft::SnapshotResponse;
 use crate::raft::VoteRequest;
 use crate::raft::VoteResponse;
+use crate::type_config::alias::AsyncRuntimeOf;
 use crate::type_config::alias::LogIdOf;
 use crate::type_config::alias::NodeIdOf;
 use crate::type_config::alias::NodeOf;
+use crate::type_config::alias::OneshotSenderOf;
 use crate::type_config::alias::SnapshotDataOf;
+use crate::AsyncRuntime;
 use crate::ChangeMembers;
 use crate::MessageSummary;
 use crate::RaftTypeConfig;
@@ -28,22 +29,23 @@ use crate::Vote;
 pub(crate) mod external_command;
 
 /// A oneshot TX to send result from `RaftCore` to external caller, e.g. `Raft::append_entries`.
-pub(crate) type ResultSender<T, E = Infallible> = oneshot::Sender<Result<T, E>>;
+pub(crate) type ResultSender<C, T, E = Infallible> = OneshotSenderOf<C, Result<T, E>>;
 
-pub(crate) type ResultReceiver<T, E = Infallible> = oneshot::Receiver<Result<T, E>>;
+pub(crate) type ResultReceiver<C, T, E = Infallible> =
+    <AsyncRuntimeOf<C> as AsyncRuntime>::OneshotReceiver<Result<T, E>>;
 
 /// TX for Vote Response
-pub(crate) type VoteTx<NID> = ResultSender<VoteResponse<NID>>;
+pub(crate) type VoteTx<C> = ResultSender<C, VoteResponse<NodeIdOf<C>>>;
 
 /// TX for Append Entries Response
-pub(crate) type AppendEntriesTx<NID> = ResultSender<AppendEntriesResponse<NID>>;
+pub(crate) type AppendEntriesTx<C> = ResultSender<C, AppendEntriesResponse<NodeIdOf<C>>>;
 
 /// TX for Client Write Response
-pub(crate) type ClientWriteTx<C> = ResultSender<ClientWriteResponse<C>, ClientWriteError<NodeIdOf<C>, NodeOf<C>>>;
+pub(crate) type ClientWriteTx<C> = ResultSender<C, ClientWriteResponse<C>, ClientWriteError<NodeIdOf<C>, NodeOf<C>>>;
 
 /// TX for Linearizable Read Response
 pub(crate) type ClientReadTx<C> =
-    ResultSender<(Option<LogIdOf<C>>, Option<LogIdOf<C>>), CheckIsLeaderError<NodeIdOf<C>, NodeOf<C>>>;
+    ResultSender<C, (Option<LogIdOf<C>>, Option<LogIdOf<C>>), CheckIsLeaderError<NodeIdOf<C>, NodeOf<C>>>;
 
 /// A message sent by application to the [`RaftCore`].
 ///
@@ -53,18 +55,18 @@ where C: RaftTypeConfig
 {
     AppendEntries {
         rpc: AppendEntriesRequest<C>,
-        tx: AppendEntriesTx<C::NodeId>,
+        tx: AppendEntriesTx<C>,
     },
 
     RequestVote {
         rpc: VoteRequest<C::NodeId>,
-        tx: VoteTx<C::NodeId>,
+        tx: VoteTx<C>,
     },
 
     InstallFullSnapshot {
         vote: Vote<C::NodeId>,
         snapshot: Snapshot<C>,
-        tx: ResultSender<SnapshotResponse<C::NodeId>>,
+        tx: ResultSender<C, SnapshotResponse<C::NodeId>>,
     },
 
     /// Begin receiving a snapshot from the leader.
@@ -74,7 +76,7 @@ where C: RaftTypeConfig
     /// will be returned in a Err
     BeginReceivingSnapshot {
         vote: Vote<C::NodeId>,
-        tx: ResultSender<Box<SnapshotDataOf<C>>, HigherVote<C::NodeId>>,
+        tx: ResultSender<C, Box<SnapshotDataOf<C>>, HigherVote<C::NodeId>>,
     },
 
     ClientWriteRequest {
@@ -88,7 +90,7 @@ where C: RaftTypeConfig
 
     Initialize {
         members: BTreeMap<C::NodeId, C::Node>,
-        tx: ResultSender<(), InitializeError<C::NodeId, C::Node>>,
+        tx: ResultSender<C, (), InitializeError<C::NodeId, C::Node>>,
     },
 
     ChangeMembership {
@@ -98,7 +100,7 @@ where C: RaftTypeConfig
         /// config will be converted into learners, otherwise they will be removed.
         retain: bool,
 
-        tx: ResultSender<ClientWriteResponse<C>, ClientWriteError<C::NodeId, C::Node>>,
+        tx: ResultSender<C, ClientWriteResponse<C>, ClientWriteError<C::NodeId, C::Node>>,
     },
 
     ExternalCoreRequest {
diff --git a/openraft/src/core/sm/command.rs b/openraft/src/core/sm/command.rs
index 2c49c2597..11ddcf8bc 100644
--- a/openraft/src/core/sm/command.rs
+++ b/openraft/src/core/sm/command.rs
@@ -54,12 +54,12 @@ where C: RaftTypeConfig
         Command::new(payload)
     }
 
-    pub(crate) fn get_snapshot(tx: ResultSender<Option<Snapshot<C>>>) -> Self {
+    pub(crate) fn get_snapshot(tx: ResultSender<C, Option<Snapshot<C>>>) -> Self {
         let payload = CommandPayload::GetSnapshot { tx };
         Command::new(payload)
     }
 
-    pub(crate) fn begin_receiving_snapshot(tx: ResultSender<Box<SnapshotDataOf<C>>, HigherVote<C::NodeId>>) -> Self {
+    pub(crate) fn begin_receiving_snapshot(tx: ResultSender<C, Box<SnapshotDataOf<C>>, HigherVote<C::NodeId>>) -> Self {
         let payload = CommandPayload::BeginReceivingSnapshot { tx };
         Command::new(payload)
     }
@@ -91,11 +91,11 @@ where C: RaftTypeConfig
 
     /// Get the latest built snapshot.
     GetSnapshot {
-        tx: ResultSender<Option<Snapshot<C>>>,
+        tx: ResultSender<C, Option<Snapshot<C>>>,
     },
 
     BeginReceivingSnapshot {
-        tx: ResultSender<Box<SnapshotDataOf<C>>, HigherVote<C::NodeId>>,
+        tx: ResultSender<C, Box<SnapshotDataOf<C>>, HigherVote<C::NodeId>>,
     },
 
     InstallFullSnapshot {
diff --git a/openraft/src/core/sm/mod.rs b/openraft/src/core/sm/mod.rs
index 05a9d6987..af810f6c0 100644
--- a/openraft/src/core/sm/mod.rs
+++ b/openraft/src/core/sm/mod.rs
@@ -6,6 +6,7 @@
 
 use tokio::sync::mpsc;
 
+use crate::async_runtime::AsyncOneshotSendExt;
 use crate::core::ApplyResult;
 use crate::core::ApplyingEntry;
 use crate::entry::RaftPayload;
@@ -219,7 +220,7 @@ where
     }
 
     #[tracing::instrument(level = "info", skip_all)]
-    async fn get_snapshot(&mut self, tx: ResultSender<Option<Snapshot<C>>>) -> Result<(), StorageError<C::NodeId>> {
+    async fn get_snapshot(&mut self, tx: ResultSender<C, Option<Snapshot<C>>>) -> Result<(), StorageError<C::NodeId>> {
         tracing::info!("{}", func_name!());
 
         let snapshot = self.state_machine.get_current_snapshot().await?;
diff --git a/openraft/src/engine/command.rs b/openraft/src/engine/command.rs
index a61ff97f4..6673ebc75 100644
--- a/openraft/src/engine/command.rs
+++ b/openraft/src/engine/command.rs
@@ -1,7 +1,6 @@
 use std::fmt::Debug;
 
-use tokio::sync::oneshot;
-
+use crate::async_runtime::AsyncOneshotSendExt;
 use crate::core::sm;
 use crate::engine::CommandKind;
 use crate::error::Infallible;
@@ -14,10 +13,11 @@ use crate::raft::InstallSnapshotResponse;
 use crate::raft::SnapshotResponse;
 use crate::raft::VoteRequest;
 use crate::raft::VoteResponse;
+use crate::type_config::alias::OneshotSenderOf;
 use crate::LeaderId;
 use crate::LogId;
-use crate::Node;
 use crate::NodeId;
+use crate::OptionalSend;
 use crate::RaftTypeConfig;
 use crate::Vote;
 
@@ -98,7 +98,7 @@ where C: RaftTypeConfig
     /// Send result to caller
     Respond {
         when: Option<Condition<C::NodeId>>,
-        resp: Respond<C::NodeId, C::Node>,
+        resp: Respond<C>,
     },
 }
 
@@ -218,31 +218,26 @@ where NID: NodeId
 }
 
 /// A command to send return value to the caller via a `oneshot::Sender`.
-#[derive(Debug)]
-#[derive(PartialEq, Eq)]
+#[derive(Debug, PartialEq, Eq)]
 #[derive(derive_more::From)]
-pub(crate) enum Respond<NID, N>
-where
-    NID: NodeId,
-    N: Node,
+pub(crate) enum Respond<C>
+where C: RaftTypeConfig
 {
-    Vote(ValueSender<Result<VoteResponse<NID>, Infallible>>),
-    AppendEntries(ValueSender<Result<AppendEntriesResponse<NID>, Infallible>>),
-    ReceiveSnapshotChunk(ValueSender<Result<(), InstallSnapshotError>>),
-    InstallSnapshot(ValueSender<Result<InstallSnapshotResponse<NID>, InstallSnapshotError>>),
-    InstallFullSnapshot(ValueSender<Result<SnapshotResponse<NID>, Infallible>>),
-    Initialize(ValueSender<Result<(), InitializeError<NID, N>>>),
+    Vote(ValueSender<C, Result<VoteResponse<C::NodeId>, Infallible>>),
+    AppendEntries(ValueSender<C, Result<AppendEntriesResponse<C::NodeId>, Infallible>>),
+    ReceiveSnapshotChunk(ValueSender<C, Result<(), InstallSnapshotError>>),
+    InstallSnapshot(ValueSender<C, Result<InstallSnapshotResponse<C::NodeId>, InstallSnapshotError>>),
+    InstallFullSnapshot(ValueSender<C, Result<SnapshotResponse<C::NodeId>, Infallible>>),
+    Initialize(ValueSender<C, Result<(), InitializeError<C::NodeId, C::Node>>>),
 }
 
-impl<NID, N> Respond<NID, N>
-where
-    NID: NodeId,
-    N: Node,
+impl<C> Respond<C>
+where C: RaftTypeConfig
 {
-    pub(crate) fn new<T>(res: T, tx: oneshot::Sender<T>) -> Self
+    pub(crate) fn new<T>(res: T, tx: OneshotSenderOf<C, T>) -> Self
     where
-        T: Debug + PartialEq + Eq,
-        Self: From<ValueSender<T>>,
+        T: Debug + PartialEq + Eq + OptionalSend,
+        Self: From<ValueSender<C, T>>,
     {
         Respond::from(ValueSender::new(res, tx))
     }
@@ -260,27 +255,38 @@ where
 }
 
 #[derive(Debug)]
-pub(crate) struct ValueSender<T>
-where T: Debug + PartialEq + Eq
+pub(crate) struct ValueSender<C, T>
+where
+    T: Debug + PartialEq + Eq + OptionalSend,
+    C: RaftTypeConfig,
 {
     value: T,
-    tx: oneshot::Sender<T>,
+    tx: OneshotSenderOf<C, T>,
 }
 
-impl<T> PartialEq for ValueSender<T>
-where T: Debug + PartialEq + Eq
+impl<C, T> PartialEq for ValueSender<C, T>
+where
+    T: Debug + PartialEq + Eq + OptionalSend,
+    C: RaftTypeConfig,
 {
     fn eq(&self, other: &Self) -> bool {
         self.value == other.value
     }
 }
 
-impl<T> Eq for ValueSender<T> where T: Debug + PartialEq + Eq {}
+impl<C, T> Eq for ValueSender<C, T>
+where
+    T: Debug + PartialEq + Eq + OptionalSend,
+    C: RaftTypeConfig,
+{
+}
 
-impl<T> ValueSender<T>
-where T: Debug + PartialEq + Eq
+impl<C, T> ValueSender<C, T>
+where
+    T: Debug + PartialEq + Eq + OptionalSend,
+    C: RaftTypeConfig,
 {
-    pub(crate) fn new(res: T, tx: oneshot::Sender<T>) -> Self {
+    pub(crate) fn new(res: T, tx: OneshotSenderOf<C, T>) -> Self {
         Self { value: res, tx }
     }
 
diff --git a/openraft/src/engine/engine_impl.rs b/openraft/src/engine/engine_impl.rs
index 8fbe07c0a..2cf2074b1 100644
--- a/openraft/src/engine/engine_impl.rs
+++ b/openraft/src/engine/engine_impl.rs
@@ -2,6 +2,7 @@ use std::time::Duration;
 
 use validit::Valid;
 
+use crate::async_runtime::AsyncOneshotSendExt;
 use crate::core::raft_msg::AppendEntriesTx;
 use crate::core::raft_msg::ResultSender;
 use crate::core::sm;
@@ -44,6 +45,7 @@ use crate::Instant;
 use crate::LogId;
 use crate::LogIdOptionExt;
 use crate::Membership;
+use crate::OptionalSend;
 use crate::RaftLogId;
 use crate::RaftTypeConfig;
 use crate::Snapshot;
@@ -222,9 +224,11 @@ where C: RaftTypeConfig
     #[tracing::instrument(level = "debug", skip_all)]
     pub(crate) fn get_leader_handler_or_reject<T, E>(
         &mut self,
-        tx: Option<ResultSender<T, E>>,
-    ) -> Option<(LeaderHandler<C>, Option<ResultSender<T, E>>)>
+        tx: Option<ResultSender<C, T, E>>,
+    ) -> Option<(LeaderHandler<C>, Option<ResultSender<C, T, E>>)>
     where
+        T: OptionalSend,
+        E: OptionalSend,
         E: From<ForwardToLeader<C::NodeId, C::Node>>,
     {
         let res = self.leader_handler();
@@ -391,7 +395,7 @@ where C: RaftTypeConfig
         vote: &Vote<C::NodeId>,
         prev_log_id: Option<LogId<C::NodeId>>,
         entries: Vec<C::Entry>,
-        tx: Option<AppendEntriesTx<C::NodeId>>,
+        tx: Option<AppendEntriesTx<C>>,
     ) -> bool {
         tracing::debug!(
             vote = display(vote),
@@ -454,7 +458,7 @@ where C: RaftTypeConfig
         &mut self,
         vote: Vote<C::NodeId>,
         snapshot: Snapshot<C>,
-        tx: ResultSender<SnapshotResponse<C::NodeId>>,
+        tx: ResultSender<C, SnapshotResponse<C::NodeId>>,
     ) {
         tracing::info!(vote = display(vote), snapshot = display(&snapshot), "{}", func_name!());
 
@@ -487,7 +491,7 @@ where C: RaftTypeConfig
     pub(crate) fn handle_begin_receiving_snapshot(
         &mut self,
         vote: Vote<C::NodeId>,
-        tx: ResultSender<Box<SnapshotDataOf<C>>, HigherVote<C::NodeId>>,
+        tx: ResultSender<C, Box<SnapshotDataOf<C>>, HigherVote<C::NodeId>>,
     ) {
         tracing::info!(vote = display(vote), "{}", func_name!());
 
diff --git a/openraft/src/engine/handler/vote_handler/accept_vote_test.rs b/openraft/src/engine/handler/vote_handler/accept_vote_test.rs
index 6e71a591d..7de499e7a 100644
--- a/openraft/src/engine/handler/vote_handler/accept_vote_test.rs
+++ b/openraft/src/engine/handler/vote_handler/accept_vote_test.rs
@@ -2,7 +2,6 @@ use std::sync::Arc;
 
 use maplit::btreeset;
 use pretty_assertions::assert_eq;
-use tokio::sync::oneshot;
 
 use crate::core::ServerState;
 use crate::engine::testing::UTConfig;
@@ -12,7 +11,9 @@ use crate::engine::Respond;
 use crate::error::Infallible;
 use crate::raft::VoteResponse;
 use crate::testing::log_id;
+use crate::type_config::alias::AsyncRuntimeOf;
 use crate::utime::UTime;
+use crate::AsyncRuntime;
 use crate::EffectiveMembership;
 use crate::Membership;
 use crate::TokioInstant;
@@ -51,12 +52,12 @@ fn test_accept_vote_reject_smaller_vote() -> anyhow::Result<()> {
     // When a vote is reject, it generate SendResultCommand and return an error.
     let mut eng = eng();
 
-    let (tx, _rx) = oneshot::channel();
+    let (tx, _rx) = AsyncRuntimeOf::<UTConfig>::oneshot();
     let resp = eng.vote_handler().accept_vote(&Vote::new(1, 2), tx, |_state, _err| mk_res());
 
     assert!(resp.is_none());
 
-    let (tx, _rx) = oneshot::channel();
+    let (tx, _rx) = AsyncRuntimeOf::<UTConfig>::oneshot();
     assert_eq!(
         vec![
             //
@@ -76,7 +77,7 @@ fn test_accept_vote_granted_greater_vote() -> anyhow::Result<()> {
     // When a vote is accepted, it generate SaveVote command and return an Ok.
     let mut eng = eng();
 
-    let (tx, _rx) = oneshot::channel();
+    let (tx, _rx) = AsyncRuntimeOf::<UTConfig>::oneshot();
     let resp = eng.vote_handler().accept_vote(&Vote::new(3, 3), tx, |_state, _err| mk_res());
 
     assert!(resp.is_some());
diff --git a/openraft/src/engine/handler/vote_handler/mod.rs b/openraft/src/engine/handler/vote_handler/mod.rs
index c6b6bdd1b..92f4eb647 100644
--- a/openraft/src/engine/handler/vote_handler/mod.rs
+++ b/openraft/src/engine/handler/vote_handler/mod.rs
@@ -11,9 +11,11 @@ use crate::error::RejectVoteRequest;
 use crate::internal_server_state::InternalServerState;
 use crate::leader::Leading;
 use crate::raft_state::LogStateReader;
+use crate::type_config::alias::InstantOf;
 use crate::utime::UTime;
 use crate::AsyncRuntime;
 use crate::Instant;
+use crate::OptionalSend;
 use crate::RaftState;
 use crate::RaftTypeConfig;
 use crate::Vote;
@@ -50,17 +52,14 @@ where C: RaftTypeConfig
     pub(crate) fn accept_vote<T, E, F>(
         &mut self,
         vote: &Vote<C::NodeId>,
-        tx: ResultSender<T, E>,
+        tx: ResultSender<C, T, E>,
         f: F,
-    ) -> Option<ResultSender<T, E>>
+    ) -> Option<ResultSender<C, T, E>>
     where
-        T: Debug + Eq,
-        E: Debug + Eq,
-        Respond<C::NodeId, C::Node>: From<ValueSender<Result<T, E>>>,
-        F: Fn(
-            &RaftState<C::NodeId, C::Node, <C::AsyncRuntime as AsyncRuntime>::Instant>,
-            RejectVoteRequest<C::NodeId>,
-        ) -> Result<T, E>,
+        T: Debug + Eq + OptionalSend,
+        E: Debug + Eq + OptionalSend,
+        Respond<C>: From<ValueSender<C, Result<T, E>>>,
+        F: Fn(&RaftState<C::NodeId, C::Node, InstantOf<C>>, RejectVoteRequest<C::NodeId>) -> Result<T, E>,
     {
         let vote_res = self.update_vote(vote);
 
diff --git a/openraft/src/raft/external_request.rs b/openraft/src/raft/external_request.rs
index c9b9b425e..a7309c38d 100644
--- a/openraft/src/raft/external_request.rs
+++ b/openraft/src/raft/external_request.rs
@@ -3,7 +3,19 @@
 use crate::type_config::alias::InstantOf;
 use crate::type_config::alias::NodeIdOf;
 use crate::type_config::alias::NodeOf;
+use crate::OptionalSend;
 use crate::RaftState;
+use crate::RaftTypeConfig;
+
+pub trait BoxCoreFnInternal<C>: FnOnce(&RaftState<NodeIdOf<C>, NodeOf<C>, InstantOf<C>>) + OptionalSend
+where C: RaftTypeConfig
+{
+}
+
+impl<C: RaftTypeConfig, T: FnOnce(&RaftState<NodeIdOf<C>, NodeOf<C>, InstantOf<C>>) + OptionalSend> BoxCoreFnInternal<C>
+    for T
+{
+}
 
 /// Boxed trait object for external request function run in `RaftCore` task.
-pub(crate) type BoxCoreFn<C> = Box<dyn FnOnce(&RaftState<NodeIdOf<C>, NodeOf<C>, InstantOf<C>>) + Send + 'static>;
+pub(crate) type BoxCoreFn<C> = Box<dyn BoxCoreFnInternal<C> + 'static>;
diff --git a/openraft/src/raft/mod.rs b/openraft/src/raft/mod.rs
index 5ae484b05..3fece00cf 100644
--- a/openraft/src/raft/mod.rs
+++ b/openraft/src/raft/mod.rs
@@ -28,13 +28,13 @@ pub use message::SnapshotResponse;
 pub use message::VoteRequest;
 pub use message::VoteResponse;
 use tokio::sync::mpsc;
-use tokio::sync::oneshot;
 use tokio::sync::watch;
 use tokio::sync::Mutex;
 use tracing::trace_span;
 use tracing::Instrument;
 use tracing::Level;
 
+use crate::async_runtime::AsyncOneshotSendExt;
 use crate::config::Config;
 use crate::config::RuntimeConfig;
 use crate::core::command_state::CommandState;
@@ -70,6 +70,7 @@ use crate::ChangeMembers;
 use crate::LogId;
 use crate::LogIdOptionExt;
 use crate::MessageSummary;
+use crate::OptionalSend;
 use crate::RaftState;
 pub use crate::RaftTypeConfig;
 use crate::Snapshot;
@@ -180,7 +181,7 @@ where C: RaftTypeConfig
         let (tx_metrics, rx_metrics) = watch::channel(RaftMetrics::new_initial(id));
         let (tx_data_metrics, rx_data_metrics) = watch::channel(RaftDataMetrics::default());
         let (tx_server_metrics, rx_server_metrics) = watch::channel(RaftServerMetrics::default());
-        let (tx_shutdown, rx_shutdown) = oneshot::channel();
+        let (tx_shutdown, rx_shutdown) = C::AsyncRuntime::oneshot();
 
         let tick_handle = Tick::spawn(
             Duration::from_millis(config.heartbeat_interval * 3 / 2),
@@ -335,7 +336,7 @@ where C: RaftTypeConfig
     ) -> Result<AppendEntriesResponse<C::NodeId>, RaftError<C::NodeId>> {
         tracing::debug!(rpc = display(rpc.summary()), "Raft::append_entries");
 
-        let (tx, rx) = oneshot::channel();
+        let (tx, rx) = C::AsyncRuntime::oneshot();
         self.inner.call_core(RaftMsg::AppendEntries { rpc, tx }, rx).await
     }
 
@@ -347,7 +348,7 @@ where C: RaftTypeConfig
     pub async fn vote(&self, rpc: VoteRequest<C::NodeId>) -> Result<VoteResponse<C::NodeId>, RaftError<C::NodeId>> {
         tracing::info!(rpc = display(rpc.summary()), "Raft::vote()");
 
-        let (tx, rx) = oneshot::channel();
+        let (tx, rx) = C::AsyncRuntime::oneshot();
         self.inner.call_core(RaftMsg::RequestVote { rpc, tx }, rx).await
     }
 
@@ -359,7 +360,7 @@ where C: RaftTypeConfig
     pub async fn get_snapshot(&self) -> Result<Option<Snapshot<C>>, RaftError<C::NodeId>> {
         tracing::debug!("Raft::get_snapshot()");
 
-        let (tx, rx) = oneshot::channel();
+        let (tx, rx) = C::AsyncRuntime::oneshot();
         let cmd = ExternalCommand::GetSnapshot { tx };
         self.inner.call_core(RaftMsg::ExternalCommand { cmd }, rx).await
     }
@@ -372,7 +373,7 @@ where C: RaftTypeConfig
     ) -> Result<Box<SnapshotDataOf<C>>, RaftError<C::NodeId, HigherVote<C::NodeId>>> {
         tracing::info!("Raft::begin_receiving_snapshot()");
 
-        let (tx, rx) = oneshot::channel();
+        let (tx, rx) = C::AsyncRuntime::oneshot();
         let resp = self.inner.call_core(RaftMsg::BeginReceivingSnapshot { vote, tx }, rx).await?;
         Ok(resp)
     }
@@ -390,7 +391,7 @@ where C: RaftTypeConfig
     ) -> Result<SnapshotResponse<C::NodeId>, Fatal<C::NodeId>> {
         tracing::info!("Raft::install_full_snapshot()");
 
-        let (tx, rx) = oneshot::channel();
+        let (tx, rx) = C::AsyncRuntime::oneshot();
         let res = self.inner.call_core(RaftMsg::InstallFullSnapshot { vote, snapshot, tx }, rx).await;
         match res {
             Ok(x) => Ok(x),
@@ -491,7 +492,7 @@ where C: RaftTypeConfig
     #[deprecated(since = "0.9.0", note = "use `Raft::ensure_linearizable()` instead")]
     #[tracing::instrument(level = "debug", skip(self))]
     pub async fn is_leader(&self) -> Result<(), RaftError<C::NodeId, CheckIsLeaderError<C::NodeId, C::Node>>> {
-        let (tx, rx) = oneshot::channel();
+        let (tx, rx) = C::AsyncRuntime::oneshot();
         let _ = self.inner.call_core(RaftMsg::CheckIsLeaderRequest { tx }, rx).await?;
         Ok(())
     }
@@ -575,7 +576,7 @@ where C: RaftTypeConfig
         (Option<LogId<C::NodeId>>, Option<LogId<C::NodeId>>),
         RaftError<C::NodeId, CheckIsLeaderError<C::NodeId, C::Node>>,
     > {
-        let (tx, rx) = oneshot::channel();
+        let (tx, rx) = C::AsyncRuntime::oneshot();
         let (read_log_id, applied) = self.inner.call_core(RaftMsg::CheckIsLeaderRequest { tx }, rx).await?;
         Ok((read_log_id, applied))
     }
@@ -603,7 +604,7 @@ where C: RaftTypeConfig
         &self,
         app_data: C::D,
     ) -> Result<ClientWriteResponse<C>, RaftError<C::NodeId, ClientWriteError<C::NodeId, C::Node>>> {
-        let (tx, rx) = oneshot::channel();
+        let (tx, rx) = C::AsyncRuntime::oneshot();
         self.inner.call_core(RaftMsg::ClientWriteRequest { app_data, tx }, rx).await
     }
 
@@ -636,7 +637,7 @@ where C: RaftTypeConfig
     where
         T: IntoNodes<C::NodeId, C::Node> + Debug,
     {
-        let (tx, rx) = oneshot::channel();
+        let (tx, rx) = C::AsyncRuntime::oneshot();
         self.inner
             .call_core(
                 RaftMsg::Initialize {
@@ -671,7 +672,7 @@ where C: RaftTypeConfig
         node: C::Node,
         blocking: bool,
     ) -> Result<ClientWriteResponse<C>, RaftError<C::NodeId, ClientWriteError<C::NodeId, C::Node>>> {
-        let (tx, rx) = oneshot::channel();
+        let (tx, rx) = C::AsyncRuntime::oneshot();
         let resp = self
             .inner
             .call_core(
@@ -801,7 +802,7 @@ where C: RaftTypeConfig
             "change_membership: start to commit joint config"
         );
 
-        let (tx, rx) = oneshot::channel();
+        let (tx, rx) = C::AsyncRuntime::oneshot();
         // res is error if membership can not be changed.
         // If no error, it will enter a joint state
         let res = self
@@ -832,7 +833,7 @@ where C: RaftTypeConfig
         tracing::debug!("committed a joint config: {} {:?}", log_id, joint);
         tracing::debug!("the second step is to change to uniform config: {:?}", changes);
 
-        let (tx, rx) = oneshot::channel();
+        let (tx, rx) = C::AsyncRuntime::oneshot();
         let res = self.inner.call_core(RaftMsg::ChangeMembership { changes, retain, tx }, rx).await;
 
         if let Err(e) = &res {
@@ -862,10 +863,12 @@ where C: RaftTypeConfig
     /// ```
     pub async fn with_raft_state<F, V>(&self, func: F) -> Result<V, Fatal<C::NodeId>>
     where
-        F: FnOnce(&RaftState<C::NodeId, C::Node, <C::AsyncRuntime as AsyncRuntime>::Instant>) -> V + Send + 'static,
-        V: Send + 'static,
+        F: FnOnce(&RaftState<C::NodeId, C::Node, <C::AsyncRuntime as AsyncRuntime>::Instant>) -> V
+            + OptionalSend
+            + 'static,
+        V: OptionalSend + 'static,
     {
-        let (tx, rx) = oneshot::channel();
+        let (tx, rx) = C::AsyncRuntime::oneshot();
 
         self.external_request(|st| {
             let result = func(st);
@@ -899,7 +902,8 @@ where C: RaftTypeConfig
     /// If the API channel is already closed (Raft is in shutdown), then the request functor is
     /// destroyed right away and not called at all.
     pub fn external_request<F>(&self, req: F)
-    where F: FnOnce(&RaftState<C::NodeId, C::Node, <C::AsyncRuntime as AsyncRuntime>::Instant>) + Send + 'static {
+    where F: FnOnce(&RaftState<C::NodeId, C::Node, <C::AsyncRuntime as AsyncRuntime>::Instant>) + OptionalSend + 'static
+    {
         let req: BoxCoreFn<C> = Box::new(req);
         let _ignore_error = self.inner.tx_api.send(RaftMsg::ExternalCoreRequest { req });
     }
diff --git a/openraft/src/raft/raft_inner.rs b/openraft/src/raft/raft_inner.rs
index ceecf7f4c..d13fdfeec 100644
--- a/openraft/src/raft/raft_inner.rs
+++ b/openraft/src/raft/raft_inner.rs
@@ -3,7 +3,6 @@ use std::fmt::Debug;
 use std::sync::Arc;
 
 use tokio::sync::mpsc;
-use tokio::sync::oneshot;
 use tokio::sync::watch;
 use tokio::sync::Mutex;
 use tracing::Level;
@@ -17,9 +16,11 @@ use crate::error::RaftError;
 use crate::metrics::RaftDataMetrics;
 use crate::metrics::RaftServerMetrics;
 use crate::raft::core_state::CoreState;
+use crate::type_config::alias::OneshotSenderOf;
 use crate::AsyncRuntime;
 use crate::Config;
 use crate::MessageSummary;
+use crate::OptionalSend;
 use crate::RaftMetrics;
 use crate::RaftTypeConfig;
 
@@ -39,7 +40,7 @@ where C: RaftTypeConfig
 
     // TODO(xp): it does not need to be a async mutex.
     #[allow(clippy::type_complexity)]
-    pub(in crate::raft) tx_shutdown: Mutex<Option<oneshot::Sender<()>>>,
+    pub(in crate::raft) tx_shutdown: Mutex<Option<OneshotSenderOf<C, ()>>>,
     pub(in crate::raft) core_state: Mutex<CoreState<C::NodeId, C::AsyncRuntime>>,
 
     /// The ongoing snapshot transmission.
@@ -55,10 +56,11 @@ where C: RaftTypeConfig
     pub(crate) async fn call_core<T, E>(
         &self,
         mes: RaftMsg<C>,
-        rx: oneshot::Receiver<Result<T, E>>,
+        rx: <C::AsyncRuntime as AsyncRuntime>::OneshotReceiver<Result<T, E>>,
     ) -> Result<T, RaftError<C::NodeId, E>>
     where
-        E: Debug,
+        E: Debug + OptionalSend,
+        T: OptionalSend,
     {
         let sum = if tracing::enabled!(Level::DEBUG) {
             Some(mes.summary())
diff --git a/openraft/src/replication/mod.rs b/openraft/src/replication/mod.rs
index d0927629a..d716b5c93 100644
--- a/openraft/src/replication/mod.rs
+++ b/openraft/src/replication/mod.rs
@@ -733,7 +733,7 @@ where
     #[tracing::instrument(level = "info", skip_all)]
     async fn stream_snapshot(
         &mut self,
-        snapshot_rx: DataWithId<ResultReceiver<Option<Snapshot<C>>>>,
+        snapshot_rx: DataWithId<ResultReceiver<C, Option<Snapshot<C>>>>,
     ) -> Result<Option<Data<C>>, ReplicationError<C::NodeId, C::Node>> {
         let request_id = snapshot_rx.request_id();
         let rx = snapshot_rx.into_data();
diff --git a/openraft/src/replication/request.rs b/openraft/src/replication/request.rs
index bbcda719c..ee2ad6d07 100644
--- a/openraft/src/replication/request.rs
+++ b/openraft/src/replication/request.rs
@@ -22,7 +22,7 @@ where C: RaftTypeConfig
         Self::Data(Data::new_logs(id, log_id_range))
     }
 
-    pub(crate) fn snapshot(id: Option<u64>, snapshot_rx: ResultReceiver<Option<Snapshot<C>>>) -> Self {
+    pub(crate) fn snapshot(id: Option<u64>, snapshot_rx: ResultReceiver<C, Option<Snapshot<C>>>) -> Self {
         Self::Data(Data::new_snapshot(id, snapshot_rx))
     }
 
@@ -73,7 +73,7 @@ where C: RaftTypeConfig
 {
     Heartbeat,
     Logs(DataWithId<LogIdRange<C::NodeId>>),
-    Snapshot(DataWithId<ResultReceiver<Option<Snapshot<C>>>>),
+    Snapshot(DataWithId<ResultReceiver<C, Option<Snapshot<C>>>>),
     SnapshotCallback(DataWithId<SnapshotCallback<C>>),
 }
 
@@ -148,7 +148,7 @@ where C: RaftTypeConfig
         Self::Logs(DataWithId::new(request_id, log_id_range))
     }
 
-    pub(crate) fn new_snapshot(request_id: Option<u64>, snapshot_rx: ResultReceiver<Option<Snapshot<C>>>) -> Self {
+    pub(crate) fn new_snapshot(request_id: Option<u64>, snapshot_rx: ResultReceiver<C, Option<Snapshot<C>>>) -> Self {
         Self::Snapshot(DataWithId::new(request_id, snapshot_rx))
     }
 
diff --git a/openraft/src/storage/adapter.rs b/openraft/src/storage/adapter.rs
index 145123938..fd6526e77 100644
--- a/openraft/src/storage/adapter.rs
+++ b/openraft/src/storage/adapter.rs
@@ -147,7 +147,7 @@ where
         S::get_log_reader(self.storage_mut().await.deref_mut()).await
     }
 
-    async fn append<I>(&mut self, entries: I, callback: LogFlushed<C::NodeId>) -> Result<(), StorageError<C::NodeId>>
+    async fn append<I>(&mut self, entries: I, callback: LogFlushed<C>) -> Result<(), StorageError<C::NodeId>>
     where I: IntoIterator<Item = C::Entry> + OptionalSend {
         // Default implementation that calls the flush-before-return `append_to_log`.
 
diff --git a/openraft/src/storage/callback.rs b/openraft/src/storage/callback.rs
index a0bda0895..b4f666ae2 100644
--- a/openraft/src/storage/callback.rs
+++ b/openraft/src/storage/callback.rs
@@ -4,26 +4,27 @@ use std::io;
 
 use tokio::sync::oneshot;
 
+use crate::async_runtime::AsyncOneshotSendExt;
 use crate::display_ext::DisplayOption;
+use crate::type_config::alias::OneshotSenderOf;
 use crate::LogId;
-use crate::NodeId;
 use crate::RaftTypeConfig;
 use crate::StorageIOError;
 
 /// A oneshot callback for completion of log io operation.
-pub struct LogFlushed<NID>
-where NID: NodeId
+pub struct LogFlushed<C>
+where C: RaftTypeConfig
 {
-    last_log_id: Option<LogId<NID>>,
-    tx: oneshot::Sender<Result<Option<LogId<NID>>, io::Error>>,
+    last_log_id: Option<LogId<C::NodeId>>,
+    tx: OneshotSenderOf<C, Result<Option<LogId<C::NodeId>>, io::Error>>,
 }
 
-impl<NID> LogFlushed<NID>
-where NID: NodeId
+impl<C> LogFlushed<C>
+where C: RaftTypeConfig
 {
     pub(crate) fn new(
-        last_log_id: Option<LogId<NID>>,
-        tx: oneshot::Sender<Result<Option<LogId<NID>>, io::Error>>,
+        last_log_id: Option<LogId<C::NodeId>>,
+        tx: OneshotSenderOf<C, Result<Option<LogId<C::NodeId>>, io::Error>>,
     ) -> Self {
         Self { last_log_id, tx }
     }
diff --git a/openraft/src/storage/v2.rs b/openraft/src/storage/v2.rs
index 77cbf28fb..2c28b7f2a 100644
--- a/openraft/src/storage/v2.rs
+++ b/openraft/src/storage/v2.rs
@@ -120,7 +120,7 @@ where C: RaftTypeConfig
     ///
     /// - There must not be a **hole** in logs. Because Raft only examine the last log id to ensure
     ///   correctness.
-    async fn append<I>(&mut self, entries: I, callback: LogFlushed<C::NodeId>) -> Result<(), StorageError<C::NodeId>>
+    async fn append<I>(&mut self, entries: I, callback: LogFlushed<C>) -> Result<(), StorageError<C::NodeId>>
     where
         I: IntoIterator<Item = C::Entry> + OptionalSend,
         I::IntoIter: OptionalSend;
diff --git a/openraft/src/testing/mod.rs b/openraft/src/testing/mod.rs
index 09530dd52..13a216abc 100644
--- a/openraft/src/testing/mod.rs
+++ b/openraft/src/testing/mod.rs
@@ -6,12 +6,12 @@ use std::collections::BTreeSet;
 use anyerror::AnyError;
 pub use store_builder::StoreBuilder;
 pub use suite::Suite;
-use tokio::sync::oneshot;
 
 use crate::entry::RaftEntry;
 use crate::log_id::RaftLogId;
 use crate::storage::LogFlushed;
 use crate::storage::RaftLogStorage;
+use crate::AsyncRuntime;
 use crate::CommittedLeaderId;
 use crate::LogId;
 use crate::RaftTypeConfig;
@@ -55,7 +55,7 @@ where
     let entries = entries.into_iter().collect::<Vec<_>>();
     let last_log_id = entries.last().map(|e| *e.get_log_id()).unwrap();
 
-    let (tx, rx) = oneshot::channel();
+    let (tx, rx) = <C::AsyncRuntime as AsyncRuntime>::oneshot();
     let cb = LogFlushed::new(Some(last_log_id), tx);
     log_store.append(entries, cb).await?;
     rx.await.unwrap().map_err(|e| StorageIOError::write_logs(AnyError::error(e)))?;
diff --git a/openraft/src/testing/suite.rs b/openraft/src/testing/suite.rs
index 0e4a20a34..49eb2a895 100644
--- a/openraft/src/testing/suite.rs
+++ b/openraft/src/testing/suite.rs
@@ -6,7 +6,6 @@ use std::time::Duration;
 
 use anyerror::AnyError;
 use maplit::btreeset;
-use tokio::sync::oneshot;
 
 use crate::entry::RaftEntry;
 use crate::log_id::RaftLogId;
@@ -1171,7 +1170,7 @@ where
     let entries = entries.into_iter().collect::<Vec<_>>();
     let last_log_id = *entries.last().unwrap().get_log_id();
 
-    let (tx, rx) = oneshot::channel();
+    let (tx, rx) = <C::AsyncRuntime as AsyncRuntime>::oneshot();
 
     let cb = LogFlushed::new(Some(last_log_id), tx);
 
diff --git a/openraft/src/timer/timeout_test.rs b/openraft/src/timer/timeout_test.rs
index 98de77cc9..445ebd51a 100644
--- a/openraft/src/timer/timeout_test.rs
+++ b/openraft/src/timer/timeout_test.rs
@@ -1,11 +1,12 @@
 use std::time::Duration;
 
-use tokio::sync::oneshot;
 use tokio::time::sleep;
 use tokio::time::Instant;
 
+use crate::async_runtime::AsyncOneshotSendExt;
 use crate::timer::timeout::RaftTimer;
 use crate::timer::Timeout;
+use crate::AsyncRuntime;
 use crate::TokioRuntime;
 
 #[cfg(not(feature = "singlethreaded"))]
@@ -24,7 +25,7 @@ fn test_timeout() -> anyhow::Result<()> {
 async fn test_timeout_inner() -> anyhow::Result<()> {
     tracing::info!("--- set timeout, recv result");
     {
-        let (tx, rx) = oneshot::channel();
+        let (tx, rx) = <TokioRuntime as AsyncRuntime>::oneshot();
         let now = Instant::now();
         let _t = Timeout::<TokioRuntime>::new(
             || {
@@ -43,7 +44,7 @@ async fn test_timeout_inner() -> anyhow::Result<()> {
 
     tracing::info!("--- update timeout");
     {
-        let (tx, rx) = oneshot::channel();
+        let (tx, rx) = <TokioRuntime as AsyncRuntime>::oneshot();
         let now = Instant::now();
         let t = Timeout::<TokioRuntime>::new(
             || {
@@ -65,7 +66,7 @@ async fn test_timeout_inner() -> anyhow::Result<()> {
 
     tracing::info!("--- update timeout to a lower value wont take effect");
     {
-        let (tx, rx) = oneshot::channel();
+        let (tx, rx) = <TokioRuntime as AsyncRuntime>::oneshot();
         let now = Instant::now();
         let t = Timeout::<TokioRuntime>::new(
             || {
@@ -87,7 +88,7 @@ async fn test_timeout_inner() -> anyhow::Result<()> {
 
     tracing::info!("--- drop the `Timeout` will cancel the callback");
     {
-        let (tx, rx) = oneshot::channel();
+        let (tx, rx) = <TokioRuntime as AsyncRuntime>::oneshot();
         let now = Instant::now();
         let t = Timeout::<TokioRuntime>::new(
             || {
diff --git a/openraft/src/type_config.rs b/openraft/src/type_config.rs
index 023c37cad..9fa647d1c 100644
--- a/openraft/src/type_config.rs
+++ b/openraft/src/type_config.rs
@@ -91,6 +91,9 @@ pub(crate) mod alias {
     pub(crate) type InstantOf<C> = <AsyncRuntimeOf<C> as crate::AsyncRuntime>::Instant;
     pub(crate) type TimeoutErrorOf<C> = <AsyncRuntimeOf<C> as crate::AsyncRuntime>::TimeoutError;
     pub(crate) type TimeoutOf<C, R, F> = <AsyncRuntimeOf<C> as crate::AsyncRuntime>::Timeout<R, F>;
+    pub(crate) type OneshotSenderOf<C, T> = <AsyncRuntimeOf<C> as crate::AsyncRuntime>::OneshotSender<T>;
+    pub(crate) type OneshotReceiverErrorOf<C> = <AsyncRuntimeOf<C> as crate::AsyncRuntime>::OneshotReceiverError;
+    pub(crate) type OneshotReceiverOf<C, T> = <AsyncRuntimeOf<C> as crate::AsyncRuntime>::OneshotReceiver<T>;
 
     // Usually used types
     pub(crate) type LogIdOf<C> = crate::LogId<NodeIdOf<C>>;
diff --git a/stores/rocksstore-v2/src/lib.rs b/stores/rocksstore-v2/src/lib.rs
index 918c52f4f..4428ac1b0 100644
--- a/stores/rocksstore-v2/src/lib.rs
+++ b/stores/rocksstore-v2/src/lib.rs
@@ -353,7 +353,7 @@ impl RaftLogStorage<TypeConfig> for RocksLogStore {
     async fn append<I>(
         &mut self,
         entries: I,
-        callback: LogFlushed<RocksNodeId>,
+        callback: LogFlushed<TypeConfig>,
     ) -> Result<(), StorageError<RocksNodeId>>
     where
         I: IntoIterator<Item = Entry<TypeConfig>> + Send,