Start the Raft event loop and RPC server. Must be called after `SharedState` is constructed (needs the WAL and dispatcher for the `SpscCommitApplier`). Moves the `MultiRaft` out of `handle.multi_raft` into the `RaftLoop`; must be called **exactly once** per handle.
(
handle: &ClusterHandle,
shared: Arc<SharedState>,
_data_dir: &std::path::Path,
shutdown_rx: tokio::sync::watch::Receiver<bool>,
transport_tuning: &ClusterTransportTuning,
)
| 35 | /// `handle.multi_raft` into the `RaftLoop`; must be called **exactly |
| 36 | /// once** per handle. |
| 37 | pub fn start_raft( |
| 38 | handle: &ClusterHandle, |
| 39 | shared: Arc<SharedState>, |
| 40 | _data_dir: &std::path::Path, |
| 41 | shutdown_rx: tokio::sync::watch::Receiver<bool>, |
| 42 | transport_tuning: &ClusterTransportTuning, |
| 43 | ) -> crate::Result<tokio::sync::watch::Receiver<bool>> { |
| 44 | // Move the MultiRaft constructed by `start_cluster` into this |
| 45 | // function. Rebuilding it here from the routing table would lose |
| 46 | // learner membership for joining nodes and would double-open |
| 47 | // per-group redb log files. |
| 48 | let mut multi_raft = handle |
| 49 | .multi_raft |
| 50 | .lock() |
| 51 | .unwrap_or_else(|p| p.into_inner()) |
| 52 | .take() |
| 53 | .ok_or_else(|| crate::Error::Config { |
| 54 | detail: "start_raft called twice: cluster multi_raft already consumed".into(), |
| 55 | })?; |
| 56 | |
| 57 | let sequencer_peers: Vec<u64> = { |
| 58 | let topo = handle.topology.read().unwrap_or_else(|p| p.into_inner()); |
| 59 | topo.all_nodes() |
| 60 | .filter(|node| node.node_id != handle.node_id && node.state.receives_log()) |
| 61 | .map(|node| node.node_id) |
| 62 | .collect() |
| 63 | }; |
| 64 | multi_raft |
| 65 | .add_group(SEQUENCER_GROUP_ID, sequencer_peers) |
| 66 | .map_err(|e| crate::Error::Config { |
| 67 | detail: format!("sequencer raft group add: {e}"), |
| 68 | })?; |
| 69 | |
| 70 | // Build the propose tracker and distributed applier. |
| 71 | // |
| 72 | // The tracker is wired with the per-group apply watermark |
| 73 | // registry so every `tracker.complete(group_id, idx, _)` call |
| 74 | // also bumps the watcher — coupling the "data applied on this |
| 75 | // node" signal to the single source of truth that proposers |
| 76 | // and cross-node visibility waits both consume. |
| 77 | let tracker = |
| 78 | Arc::new(ProposeTracker::new().with_group_watchers(handle.group_watchers.clone())); |
| 79 | let (dist_applier, apply_rx) = create_distributed_applier(tracker.clone()); |
| 80 | let dist_applier = Arc::new(dist_applier); |
| 81 | let calvin_completion_registry = CalvinCompletionRegistry::new(); |
| 82 | let sequencer_state_machine = Arc::new(Mutex::new(SequencerStateMachine::new( |
| 83 | std::collections::HashMap::new(), |
| 84 | Arc::clone(&calvin_completion_registry), |
| 85 | ))); |
| 86 | let calvin_read_result_senders = Arc::new(Mutex::new(std::collections::BTreeMap::< |
| 87 | u32, |
| 88 | tokio::sync::mpsc::Sender<ReadResultEvent>, |
| 89 | >::new())); |
| 90 | |
| 91 | // Install the propose tracker so CP dispatch paths can await commit. |
| 92 | if shared.propose_tracker.set(tracker.clone()).is_err() { |
| 93 | tracing::warn!("propose_tracker already set — start_raft appears to have run twice"); |
| 94 | } |
no test coverage detected