nodedb_cluster/raft_loop/handle_rpc/
dispatch.rs

1// SPDX-License-Identifier: BUSL-1.1
2
3//! The single `impl RaftRpcHandler for RaftLoop` block. Thin dispatch only —
4//! each method delegates to a helper defined in a sibling module
5//! ([`super::consensus`], [`super::membership`], [`super::plan_dispatch`],
6//! [`super::shuffle_calvin`]).
7
8use crate::error::{ClusterError, Result};
9use crate::forward::{ChunkSink, PlanExecutor};
10use crate::rpc_codec::{
11    AssignSurrogateRequest, AssignSurrogateResponse, ExecuteRequest, RaftRpc,
12    ReleaseReservationRequest, ReleaseReservationResponse, ReserveReadRequest, ReserveReadResponse,
13    ShuffleAggregateConsumeRequest, ShuffleAggregateConsumeResponse, ShuffleConsumeRequest,
14    ShuffleConsumeResponse, ShuffleProduceRequest, ShuffleProduceResponse, ShufflePushRequest,
15    SubmitCalvinInboxRequest, SubmitCalvinInboxResponse, SubmitCalvinTxnRequest,
16    SubmitCalvinTxnResponse, TypedClusterError,
17};
18use crate::transport::RaftRpcHandler;
19use nodedb_raft::message::TimeoutNowRequest;
20
21use super::super::loop_core::{CommitApplier, RaftLoop};
22
23impl<A: CommitApplier, P: PlanExecutor> RaftRpcHandler for RaftLoop<A, P> {
24    async fn handle_rpc(&self, rpc: RaftRpc) -> Result<RaftRpc> {
25        match rpc {
26            // Raft consensus RPCs — lock MultiRaft (sync, never across await).
27            RaftRpc::AppendEntriesRequest(req) => self.handle_append_entries_rpc(req),
28            RaftRpc::RequestVoteRequest(req) => self.handle_request_vote_rpc(req),
29            RaftRpc::InstallSnapshotRequest(req) => self.handle_install_snapshot_rpc(req).await,
30            // Cluster join — full orchestration in `super::join`.
31            RaftRpc::JoinRequest(req) => Ok(RaftRpc::JoinResponse(self.join_flow(req).await)),
32            // Health check.
33            RaftRpc::Ping(req) => self.handle_ping_rpc(req),
34            // Topology broadcast.
35            RaftRpc::TopologyUpdate(update) => self.handle_topology_update_rpc(update),
36            // Physical-plan execution (C-β) — execute locally via the PlanExecutor,
37            // skipping SQL re-planning entirely.
38            RaftRpc::ExecuteRequest(req) => self.handle_execute_rpc(req).await,
39            // Metadata-group proposal forwarding.
40            RaftRpc::MetadataProposeRequest(req) => self.handle_metadata_propose_rpc(req),
41            // Data-group proposal forwarding.
42            RaftRpc::DataProposeRequest(req) => self.handle_data_propose_rpc(req),
43            // VShardEnvelope — dispatch to registered handler (Event Plane, etc.).
44            RaftRpc::VShardEnvelope(bytes) => self.handle_vshard_envelope_rpc(bytes).await,
45            other => Err(ClusterError::Transport {
46                detail: format!("unexpected request type in RPC handler: {other:?}"),
47            }),
48        }
49    }
50
51    // Streaming physical-plan execution (L4) — delegate to the PlanExecutor's
52    // streaming path. The transport drives the multi-frame chunk/end envelope
53    // writes; this just runs the plan and feeds `sink`.
54    async fn handle_rpc_streaming(
55        &self,
56        req: ExecuteRequest,
57        sink: impl ChunkSink,
58    ) -> Option<TypedClusterError> {
59        self.handle_rpc_streaming_impl(req, sink).await
60    }
61
62    async fn on_shuffle_request(&self, req: ShufflePushRequest) {
63        self.on_shuffle_request_impl(req).await
64    }
65
66    async fn on_shuffle_chunk(
67        &self,
68        shuffle_id: u64,
69        part: u32,
70        side: u8,
71        payload: Vec<u8>,
72    ) -> Result<()> {
73        self.on_shuffle_chunk_impl(shuffle_id, part, side, payload)
74            .await
75    }
76
77    async fn on_shuffle_end(
78        &self,
79        shuffle_id: u64,
80        part: u32,
81        side: u8,
82        error: Option<TypedClusterError>,
83    ) {
84        self.on_shuffle_end_impl(shuffle_id, part, side, error)
85            .await
86    }
87
88    async fn on_shuffle_produce(&self, req: ShuffleProduceRequest) -> ShuffleProduceResponse {
89        self.on_shuffle_produce_impl(req).await
90    }
91
92    async fn on_shuffle_consume(&self, req: ShuffleConsumeRequest) -> ShuffleConsumeResponse {
93        self.on_shuffle_consume_impl(req).await
94    }
95
96    async fn on_shuffle_aggregate(
97        &self,
98        req: ShuffleAggregateConsumeRequest,
99    ) -> ShuffleAggregateConsumeResponse {
100        self.on_shuffle_aggregate_impl(req).await
101    }
102
103    async fn on_assign_surrogate(&self, req: AssignSurrogateRequest) -> AssignSurrogateResponse {
104        self.on_assign_surrogate_impl(req).await
105    }
106
107    async fn on_submit_calvin_txn(&self, req: SubmitCalvinTxnRequest) -> SubmitCalvinTxnResponse {
108        self.on_submit_calvin_txn_impl(req).await
109    }
110
111    async fn on_submit_calvin_inbox(
112        &self,
113        req: SubmitCalvinInboxRequest,
114    ) -> SubmitCalvinInboxResponse {
115        self.on_submit_calvin_inbox_impl(req).await
116    }
117
118    async fn on_reserve_read(&self, req: ReserveReadRequest) -> ReserveReadResponse {
119        self.on_reserve_read_impl(req).await
120    }
121
122    async fn on_release_reservation(
123        &self,
124        req: ReleaseReservationRequest,
125    ) -> ReleaseReservationResponse {
126        self.on_release_reservation_impl(req).await
127    }
128
129    async fn on_timeout_now(&self, req: TimeoutNowRequest) {
130        self.on_timeout_now_impl(req).await
131    }
132}
nodedb_cluster/raft_loop/handle_rpc/dispatch.rs

nodedb_cluster/raft_loop/handle_rpc/
dispatch.rs
