mocra_cluster/
raft_http.rs

1//! Node-to-node control plane RPC: HTTP + msgpack.
2//!
3//! - **Server**: [`raft_router`] exposes openraft's append_entries / vote / install_snapshot as
4//!   HTTP endpoints.
5//! - **Client**: [`HttpNetwork`] / [`HttpConnection`] implement `RaftNetworkFactory` /
6//!   `RaftNetwork`, issuing RPCs with reqwest to the `addr` of the peer [`BasicNode`].
7//!
8//! Control plane traffic is small (membership / locks / configuration / heartbeats), so
9//! HTTP/msgpack is sufficient and self-contained (no protobuf codegen required).
10
11use axum::Router;
12use axum::body::Bytes;
13use axum::extract::State;
14use axum::routing::post;
15use openraft::BasicNode;
16use openraft::error::{InstallSnapshotError, NetworkError, RPCError, RaftError, RemoteError};
17use openraft::network::{RPCOption, RaftNetwork, RaftNetworkFactory};
18use openraft::raft::{
19    AppendEntriesRequest, AppendEntriesResponse, InstallSnapshotRequest, InstallSnapshotResponse,
20    VoteRequest, VoteResponse,
21};
22use serde::{Deserialize, Serialize};
23
24use crate::cmd::{Cmd, CmdResult};
25use crate::raft::{MocraRaft, Node, NodeId, TypeConfig};
26
27/// The join request body: a new node carries its own id and HTTP address.
28#[derive(Debug, Clone, Serialize, Deserialize)]
29pub struct JoinRequest {
30    pub node_id: NodeId,
31    pub addr: String,
32}
33
34// ============ Server ============
35
36/// Exposes a Raft instance's RPC + cluster management as HTTP routes. Mounted on the node's own
37/// HTTP server.
38pub fn raft_router(raft: MocraRaft) -> Router {
39    Router::new()
40        .route("/raft/append", post(append))
41        .route("/raft/vote", post(vote))
42        .route("/raft/snapshot", post(snapshot))
43        .route("/cluster/join", post(join))
44        .route("/cluster/write", post(cluster_write))
45        .with_state(raft)
46}
47
48/// The client write-forwarding endpoint: when any node receives a write request, its
49/// `client_write` fails if that node is not the leader; on the follower side,
50/// [`RaftControlPlane::write`](crate::RaftControlPlane) forwards the [`Cmd`] to this endpoint on
51/// the leader. The leader commits it locally here and returns a [`CmdResult`].
52/// This is what makes "register against any node" writes land.
53async fn cluster_write(State(raft): State<MocraRaft>, body: Bytes) -> Vec<u8> {
54    let cmd: Cmd = match rmp_serde::from_slice(&body) {
55        Ok(c) => c,
56        Err(e) => {
57            let out: Result<CmdResult, String> = Err(format!("decode cmd: {e}"));
58            return rmp_serde::to_vec(&out).unwrap_or_default();
59        }
60    };
61    let out: Result<CmdResult, String> = raft
62        .client_write(cmd)
63        .await
64        .map(|r| r.data)
65        .map_err(|e| e.to_string());
66    rmp_serde::to_vec(&out).unwrap_or_default()
67}
68
69/// Cluster join: adds the new node as a **learner** (plan A: a worker starts out as a learner;
70/// promoting it to a voter is an explicit operation).
71/// Should be sent to the leader; a non-leader returns openraft's ForwardToLeader error
72/// (serialized into the result).
73async fn join(State(raft): State<MocraRaft>, body: Bytes) -> Vec<u8> {
74    let req: JoinRequest = match rmp_serde::from_slice(&body) {
75        Ok(r) => r,
76        Err(_) => return Vec::new(),
77    };
78    let res = raft
79        .add_learner(req.node_id, BasicNode::new(req.addr), true)
80        .await;
81    let out: Result<(), String> = res.map(|_| ()).map_err(|e| e.to_string());
82    rmp_serde::to_vec(&out).unwrap_or_default()
83}
84
85async fn append(State(raft): State<MocraRaft>, body: Bytes) -> Vec<u8> {
86    let req: AppendEntriesRequest<TypeConfig> = match rmp_serde::from_slice(&body) {
87        Ok(r) => r,
88        Err(_) => return Vec::new(),
89    };
90    let res = raft.append_entries(req).await;
91    rmp_serde::to_vec(&res).unwrap_or_default()
92}
93
94async fn vote(State(raft): State<MocraRaft>, body: Bytes) -> Vec<u8> {
95    let req: VoteRequest<NodeId> = match rmp_serde::from_slice(&body) {
96        Ok(r) => r,
97        Err(_) => return Vec::new(),
98    };
99    let res = raft.vote(req).await;
100    rmp_serde::to_vec(&res).unwrap_or_default()
101}
102
103async fn snapshot(State(raft): State<MocraRaft>, body: Bytes) -> Vec<u8> {
104    let req: InstallSnapshotRequest<TypeConfig> = match rmp_serde::from_slice(&body) {
105        Ok(r) => r,
106        Err(_) => return Vec::new(),
107    };
108    let res = raft.install_snapshot(req).await;
109    rmp_serde::to_vec(&res).unwrap_or_default()
110}
111
112// ============ Client ============
113
114/// RaftNetwork factory: holds one shared reqwest client.
115#[derive(Clone)]
116pub struct HttpNetwork {
117    client: reqwest::Client,
118}
119
120impl HttpNetwork {
121    pub fn new() -> Self {
122        Self {
123            client: reqwest::Client::new(),
124        }
125    }
126}
127
128impl Default for HttpNetwork {
129    fn default() -> Self {
130        Self::new()
131    }
132}
133
134/// A connection to a peer node.
135pub struct HttpConnection {
136    client: reqwest::Client,
137    target: NodeId,
138    addr: String,
139}
140
141impl HttpConnection {
142    /// POSTs raw bytes to one of the peer's endpoints and returns the response bytes (only
143    /// network errors are produced).
144    async fn post(&self, path: &str, body: Vec<u8>) -> Result<Vec<u8>, NetworkError> {
145        let url = format!("http://{}{}", self.addr, path);
146        let resp = self
147            .client
148            .post(url)
149            .body(body)
150            .send()
151            .await
152            .map_err(|e| NetworkError::new(&e))?;
153        let bytes = resp.bytes().await.map_err(|e| NetworkError::new(&e))?;
154        Ok(bytes.to_vec())
155    }
156}
157
158impl RaftNetworkFactory<TypeConfig> for HttpNetwork {
159    type Network = HttpConnection;
160
161    async fn new_client(&mut self, target: NodeId, node: &Node) -> Self::Network {
162        HttpConnection {
163            client: self.client.clone(),
164            target,
165            addr: node.addr.clone(),
166        }
167    }
168}
169
170impl RaftNetwork<TypeConfig> for HttpConnection {
171    async fn append_entries(
172        &mut self,
173        req: AppendEntriesRequest<TypeConfig>,
174        _option: RPCOption,
175    ) -> Result<AppendEntriesResponse<NodeId>, RPCError<NodeId, Node, RaftError<NodeId>>> {
176        let body = rmp_serde::to_vec(&req).map_err(|e| RPCError::Network(NetworkError::new(&e)))?;
177        let bytes = self
178            .post("/raft/append", body)
179            .await
180            .map_err(RPCError::Network)?;
181        let res: Result<AppendEntriesResponse<NodeId>, RaftError<NodeId>> =
182            rmp_serde::from_slice(&bytes).map_err(|e| RPCError::Network(NetworkError::new(&e)))?;
183        res.map_err(|e| RPCError::RemoteError(RemoteError::new(self.target, e)))
184    }
185
186    async fn vote(
187        &mut self,
188        req: VoteRequest<NodeId>,
189        _option: RPCOption,
190    ) -> Result<VoteResponse<NodeId>, RPCError<NodeId, Node, RaftError<NodeId>>> {
191        let body = rmp_serde::to_vec(&req).map_err(|e| RPCError::Network(NetworkError::new(&e)))?;
192        let bytes = self
193            .post("/raft/vote", body)
194            .await
195            .map_err(RPCError::Network)?;
196        let res: Result<VoteResponse<NodeId>, RaftError<NodeId>> =
197            rmp_serde::from_slice(&bytes).map_err(|e| RPCError::Network(NetworkError::new(&e)))?;
198        res.map_err(|e| RPCError::RemoteError(RemoteError::new(self.target, e)))
199    }
200
201    async fn install_snapshot(
202        &mut self,
203        req: InstallSnapshotRequest<TypeConfig>,
204        _option: RPCOption,
205    ) -> Result<
206        InstallSnapshotResponse<NodeId>,
207        RPCError<NodeId, Node, RaftError<NodeId, InstallSnapshotError>>,
208    > {
209        let body = rmp_serde::to_vec(&req).map_err(|e| RPCError::Network(NetworkError::new(&e)))?;
210        let bytes = self
211            .post("/raft/snapshot", body)
212            .await
213            .map_err(RPCError::Network)?;
214        let res: Result<InstallSnapshotResponse<NodeId>, RaftError<NodeId, InstallSnapshotError>> =
215            rmp_serde::from_slice(&bytes).map_err(|e| RPCError::Network(NetworkError::new(&e)))?;
216        res.map_err(|e| RPCError::RemoteError(RemoteError::new(self.target, e)))
217    }
218}
mocra_cluster/raft_http.rs

mocra_cluster/
raft_http.rs
