krabka_raft/controller/
startup.rs

1//! Cluster formation and controller start-up: the [`Controller`] factory that
2//! validates the requested bootstrap mode against the on-disk log, binds the
3//! controller listener, opens or recovers the engine, and the on-disk state
4//! probe the broker binary shares with that validation.
5
6#[cfg(target_os = "wasi")]
7use std::future;
8use {
9    super::{
10        ControllerHandle, checkpoint::load_latest_checkpoint,
11        feature_check::unsupported_feature_level,
12    },
13    crate::{
14        config::{
15            BootstrapMode, ControllerConfig, UnstableFeatureVersions, metadata_partition_dir,
16        },
17        error::RaftError,
18        kraft::{Activation, KraftController, core::QuorumTimeouts},
19        network::{OutboundDialer, PlaintextDialer, RealPeerSender},
20        server,
21        snapshot::SnapshotReader,
22    },
23    krabka_kraft_core::snapshot_fetch::MetadataSnapshotFetchMax,
24    std::{
25        collections::BTreeMap,
26        fs, io,
27        net::SocketAddr,
28        path::Path,
29        sync::{self, Arc},
30    },
31    tokio::{
32        net::TcpListener,
33        sync::{Mutex, watch},
34    },
35    tokio_util::sync::CancellationToken,
36    tracing::info,
37    uuid::Uuid,
38};
39
40/// Zero-sized factory for [`ControllerHandle`]s.
41pub struct Controller;
42
43impl Controller {
44    /// Start a controller node, open the listener, and begin participating in
45    /// the quorum.
46    ///
47    /// `bootstrap_mode` governs cluster formation: `Bootstrap` seeds a fresh
48    /// quorum from `initial_voters`; `Join`/`Rejoin` recover or wait. Mismatches
49    /// between mode and on-disk log state return [`RaftError::Startup`].
50    ///
51    /// # Errors
52    /// Returns an error if configuration, storage recovery, or startup fails.
53    pub async fn start(config: ControllerConfig) -> Result<ControllerHandle, RaftError> {
54        Self::start_with_listener(config, None).await
55    }
56
57    /// Like [`Self::start`], but adopts a caller-supplied, already-bound
58    /// controller listener instead of binding `controller_listen_addr` itself.
59    /// On native targets, the controller reports the listener's actual local
60    /// address. On WASI, it reports `config.controller_listen_addr`, which
61    /// should match the adopted socket because its address cannot be queried.
62    ///
63    /// # Errors
64    /// Returns an error if the listener, storage, or controller cannot start.
65    #[tracing::instrument(
66        level = "info",
67        skip_all,
68        fields(node = config.node_id.0, mode = ?config.bootstrap_mode),
69        err
70    )]
71    pub async fn start_with_listener(
72        config: ControllerConfig,
73        prebound: Option<TcpListener>,
74    ) -> Result<ControllerHandle, RaftError> {
75        let metadata_snapshot_fetch_max =
76            MetadataSnapshotFetchMax::new(config.metadata_snapshot_fetch_max)
77                .map_err(RaftError::Startup)?;
78
79        // First-boot orchestration validates mode against on-disk log state.
80        // The metadata log, its checkpoints and the quorum-state file all live
81        // in the `__cluster_metadata-0` directory under `log_dir`, as Kafka's
82        // do under `metadata.log.dir`.
83        let data_dir = metadata_partition_dir(&config.log_dir);
84        let log_exists = metadata_log_nonempty(&data_dir);
85        let snapshot_voters = load_latest_checkpoint(&data_dir)
86            .and_then(|(_, bytes)| SnapshotReader::read(&bytes).ok())
87            .and_then(|snapshot| snapshot.control_state.map(|state| state.voters))
88            .unwrap_or_default();
89        let voters = if config.initial_voters.is_empty() {
90            snapshot_voters
91        } else {
92            config.initial_voters.clone()
93        };
94        let bootstrap_mode = effective_bootstrap_mode(
95            config.bootstrap_mode,
96            voters.is_empty(),
97            config.auto_join || !config.bootstrap_servers.is_empty(),
98        );
99        match (bootstrap_mode, log_exists) {
100            (BootstrapMode::Bootstrap, false) => {
101                if voters.is_empty() {
102                    return Err(RaftError::Startup(
103                        "Bootstrap mode requires a non-empty initial_voters set".into(),
104                    ));
105                }
106            }
107            (BootstrapMode::Join, false) | (BootstrapMode::Rejoin, true) => {}
108            (BootstrapMode::Bootstrap, true) => {
109                return Err(RaftError::Startup(
110                    "Bootstrap mode requires empty raft log; existing log indicates an already-initialized broker — use Rejoin".into(),
111                ));
112            }
113            (BootstrapMode::Rejoin, false) => {
114                return Err(RaftError::Startup(
115                    "Rejoin mode requires non-empty raft log; this broker has no on-disk state — use Bootstrap or Join".into(),
116                ));
117            }
118            (BootstrapMode::Join, true) => {
119                return Err(RaftError::Startup(
120                    "Join mode requires empty raft log; this broker has on-disk state — use Rejoin"
121                        .into(),
122                ));
123            }
124        }
125
126        // The controller listener binds before the engine opens, so a port
127        // that another process holds fails the start with no engine left
128        // running behind it.
129        let bind_error = |source| RaftError::ControllerListenerBind {
130            addr: config.controller_listen_addr,
131            source,
132        };
133        let listener = match prebound {
134            Some(l) => l,
135            None => bind_controller_listener(config.controller_listen_addr)
136                .await
137                .map_err(bind_error)?,
138        };
139        let actual_addr =
140            listener_address(&listener, config.controller_listen_addr).map_err(bind_error)?;
141
142        let cluster_id = config.cluster_id.unwrap_or_else(Uuid::nil);
143        let dialer: Arc<dyn OutboundDialer> = config
144            .dialer
145            .clone()
146            .unwrap_or_else(|| Arc::new(PlaintextDialer));
147
148        // The peer sender starts from the bootstrap view. The engine replaces
149        // it immediately when it replays a dynamic voter control record.
150        let peers = Arc::new(RealPeerSender::new(
151            voters.clone(),
152            &config.bootstrap_servers,
153            config.client_id.clone(),
154            Arc::clone(&dialer),
155            config.client_dispatch_queue_capacity,
156            config.client_frame_max,
157        ));
158
159        // Build / recover the engine. `Join` nodes with an empty log + empty
160        // voter set sit unattached; `Bootstrap` seeds the static voter set.
161        let engine = KraftController::open(
162            data_dir.clone(),
163            config.node_id,
164            cluster_id,
165            config.directory_id,
166            voters.clone(),
167            QuorumTimeouts {
168                election: config.election_timeout,
169                fetch: config.fetch_timeout,
170            },
171            config.heartbeat_interval,
172            config.controller_fetch_miss_limit,
173            config.metadata_raft_command_queue_capacity,
174            config.metadata_raft_fetch_max,
175            peers,
176            config.snapshot_interval_records,
177            config.max_bytes_between_snapshots,
178            config.max_snapshot_interval,
179            metadata_snapshot_fetch_max,
180            config.metadata_log,
181            Activation {
182                bootstrap_records: config.bootstrap_records.clone(),
183                default_min_insync_replicas: config.default_min_insync_replicas,
184            },
185            Arc::clone(&config.event_queue_recorder),
186            |image| {
187                if let Some(refusal) =
188                    unsupported_feature_level(image, config.unstable_feature_versions)
189                {
190                    return Err(RaftError::FatalFault(refusal));
191                }
192                Ok(())
193            },
194        )?;
195
196        let shutdown = CancellationToken::new();
197        let leader_rx = engine.watch_leader();
198        let listener_task = tokio::spawn(server::run(
199            listener,
200            engine.clone(),
201            shutdown.clone(),
202            config.handshake.clone(),
203            config.shard_router.clone(),
204            config.admin_router.clone(),
205            (
206                server::Unstable {
207                    api_versions: config.unstable_api_versions,
208                },
209                config.listener_limits,
210            ),
211        ));
212        let (fatal_tx, fatal) = watch::channel(None);
213        tokio::spawn(stop_on_fatal_fault(
214            engine.clone(),
215            shutdown.clone(),
216            fatal_tx,
217            config.unstable_feature_versions,
218        ));
219        info!(
220            node_id = config.node_id.0,
221            addr = %actual_addr,
222            "controller started"
223        );
224
225        Ok(ControllerHandle {
226            engine,
227            leader: leader_rx,
228            shutdown,
229            fatal,
230            listener_task: Mutex::new(Some(listener_task)),
231            data_dir,
232            client_id: config.client_id.clone(),
233            client_dispatch_queue_capacity: config.client_dispatch_queue_capacity,
234            client_frame_max: config.client_frame_max,
235            self_node_id: config.node_id,
236            voters,
237            staged_learners: sync::Mutex::new(BTreeMap::new()),
238            dialer,
239            controller_bound_addr: actual_addr,
240        })
241    }
242}
243
244/// The mode a node starts in once it knows whether it has initial voters.
245///
246/// A fresh node with no voters, neither configured nor in its bootstrap
247/// checkpoint, is a controller formatted with `--no-initial-controllers`.
248/// Kafka starts it as an observer that finds the leader through
249/// `controller.quorum.bootstrap.servers`. Auto-join or `kafka-metadata-quorum
250/// add-controller` makes it a voter later. So `Bootstrap` becomes `Join` for a
251/// node that can reach the quorum: it has a bootstrap server, or it auto-joins,
252/// which needs one. A node with neither stays in `Bootstrap`, which refuses
253/// the empty voter set.
254fn effective_bootstrap_mode(
255    requested: BootstrapMode,
256    no_initial_voters: bool,
257    reaches_quorum: bool,
258) -> BootstrapMode {
259    if requested == BootstrapMode::Bootstrap && no_initial_voters && reaches_quorum {
260        BootstrapMode::Join
261    } else {
262        requested
263    }
264}
265
266/// Stops the controller over a fatal fault, as Kafka's fatal fault handler
267/// does. Three faults are fatal:
268///
269/// - The replay of a feature level that the controller does not support, as
270///   Kafka's fatal fault on a `FeatureControlManager` replay exception. The
271///   engine publishes each image it applies, so a node that follows a leader
272///   which finalized a level above its own range sees it here and stops
273///   serving, and never runs at a level it did not advertise.
274/// - The failure of the metadata log directory: a write to it that returned
275///   an I/O error. Kafka shuts the node down when its metadata log directory
276///   fails (KIP-858), and the engine has already stopped taking part in the
277///   quorum.
278/// - A controller activation that failed: a new leader of an empty log that
279///   could not write the bootstrap records. Kafka's `QuorumController` gives
280///   that failure to its `fatalFaultHandler` as `exception while completing
281///   controller activation`, and the engine has already stopped.
282///
283/// The fault goes out on `fatal` first, as [`ControllerHandle::watch_fatal`]
284/// documents, so that a process hosting the controller can halt over it, as
285/// Kafka's `ProcessTerminatingFaultHandler` does. It is in place before any
286/// later submit can fail with [`RaftError::Shutdown`].
287async fn stop_on_fatal_fault(
288    engine: KraftController,
289    shutdown: CancellationToken,
290    fatal: watch::Sender<Option<String>>,
291    unstable: UnstableFeatureVersions,
292) {
293    let mut images = engine.watch_image();
294    let mut faults = engine.watch_fault();
295    loop {
296        let (refusal, engine_stopped) = tokio::select! {
297            () = shutdown.cancelled() => return,
298            changed = images.changed() => match changed {
299                Ok(()) => (
300                    unsupported_feature_level(&images.borrow_and_update(), unstable),
301                    false,
302                ),
303                Err(_) => (None, true),
304            },
305            changed = faults.changed() => (None, changed.is_err()),
306        };
307        if let Some(refusal) = refusal {
308            tracing::error!(%refusal, "controller stopping: it replayed an unsupported feature level");
309            fatal.send_replace(Some(refusal));
310        } else if let Some(fault) = current(&faults) {
311            // The engine publishes the fault and then stops, so the fault can
312            // arrive together with the end of the image channel.
313            tracing::error!(%fault, "controller stopping over a fault of its engine");
314            fatal.send_replace(Some(fault));
315        } else if engine_stopped {
316            return;
317        } else {
318            continue;
319        }
320        shutdown.cancel();
321        engine.halt().await;
322        return;
323    }
324}
325
326/// The value `receiver` holds now, with its borrow already released.
327fn current<T: Clone>(receiver: &watch::Receiver<T>) -> T { receiver.borrow().clone() }
328
329/// Bind the controller listener on `addr`.
330#[cfg(not(target_os = "wasi"))]
331async fn bind_controller_listener(addr: SocketAddr) -> io::Result<TcpListener> {
332    TcpListener::bind(addr).await
333}
334
335/// WASI preview 1 has no `bind`: the embedder adopts a preopened socket and
336/// passes it to [`Controller::start_with_listener`].
337#[cfg(target_os = "wasi")]
338fn bind_controller_listener(addr: SocketAddr) -> future::Ready<io::Result<TcpListener>> {
339    future::ready(Err(io::Error::new(
340        io::ErrorKind::Unsupported,
341        format!(
342            "cannot bind the controller listener on {addr}: this platform has no bind; \
343             pass a preopened listener to Controller::start_with_listener"
344        ),
345    )))
346}
347
348/// The address `listener` is bound to.
349///
350/// WASI preview 1 has no `getsockname`, so there the configured address
351/// stands in. [`Controller::start_with_listener`] requires the two to be
352/// equal.
353fn listener_address(listener: &TcpListener, configured: SocketAddr) -> io::Result<SocketAddr> {
354    if cfg!(target_os = "wasi") {
355        Ok(configured)
356    } else {
357        listener.local_addr()
358    }
359}
360
361/// True when the metadata partition directory `dir` already holds durable
362/// raft state, which a node that ran before leaves. A quorum-state file or a
363/// log segment that holds bytes is such state.
364///
365/// `dir` is the metadata partition directory,
366/// `<metadata.log.dir>/__cluster_metadata-0`. The broker binary's
367/// `detect_bootstrap_mode` calls this so its Bootstrap/Rejoin choice can never
368/// disagree with [`Controller::start_with_listener`]'s mode validation.
369/// `KraftController::open` creates an empty active segment before the first
370/// election, so an empty segment is not state: a node killed mid-election,
371/// with that segment but no `quorum-state` yet, reads as un-formatted and
372/// re-Bootstraps rather than dying with "Rejoin requires non-empty raft log".
373#[must_use]
374pub fn metadata_log_nonempty(dir: &Path) -> bool {
375    let qs = dir.join("quorum-state");
376    if qs.exists() {
377        return true;
378    }
379    fs::read_dir(dir).is_ok_and(|entries| {
380        entries.flatten().any(|e| {
381            e.path().extension().is_some_and(|ext| ext == "log")
382                && e.metadata().is_ok_and(|metadata| metadata.len() > 0)
383        })
384    })
385}
386
387#[cfg(test)]
388mod tests {
389    use {
390        super::*,
391        crate::{
392            controller::test_support::{
393                bootstrap_controller, joining_controller, submit_change_with_timeout,
394                wait_for_leader,
395            },
396            metadata_partition_dir, test_support,
397            types::NodeId,
398        },
399        std::{fs, io::ErrorKind, iter, time::Duration},
400        tempfile::TempDir,
401        tokio::{net::TcpListener, time},
402    };
403
404    #[yare::parameterized(
405        empty_directory = { None, false },
406        quorum_state_file = { Some(("quorum-state", b"state".as_slice())), true },
407        log_segment = { Some(("00000000000000000000.log", b"log".as_slice())), true },
408        empty_log_segment = { Some(("00000000000000000000.log", b"".as_slice())), false },
409        non_log_extension = { Some(("00000000000000000000.txt", b"log".as_slice())), false },
410    )]
411    fn metadata_log_nonempty_detects_quorum_state_and_log_segments_only(
412        file: Option<(&str, &[u8])>,
413        expected: bool,
414    ) {
415        let dir = TempDir::new().unwrap();
416        if let Some((name, contents)) = file {
417            fs::write(dir.path().join(name), contents).unwrap();
418        }
419        assert2::assert!(metadata_log_nonempty(dir.path()) == expected);
420    }
421
422    /// A write to the metadata log directory that fails stops the controller
423    /// with a fatal fault, as Kafka shuts a node down when its metadata log
424    /// directory fails (KIP-858).
425    #[tokio::test]
426    async fn a_failed_metadata_log_directory_stops_the_controller() {
427        let dir = TempDir::new().unwrap();
428        let ctrl = Controller::start(ControllerConfig::for_tests(
429            NodeId(1),
430            dir.path().to_path_buf(),
431        ))
432        .await
433        .expect("start");
434        wait_for_leader(&ctrl).await;
435        // The committed offset is written to this file on every advance. A
436        // directory in its place makes the next write fail with an I/O error.
437        let high_watermark = metadata_partition_dir(dir.path()).join("high-watermark");
438        let _ = fs::remove_file(&high_watermark);
439        fs::create_dir(&high_watermark).unwrap();
440
441        let _ = time::timeout(
442            Duration::from_secs(5),
443            ctrl.submit_change(vec![krabka_metadata::MetadataRecord::V1Topic(
444                test_support::single_partition_topic("after-the-failure", Uuid::new_v4()),
445            )]),
446        )
447        .await;
448        let mut fatal = ctrl.watch_fatal();
449        let fault = time::timeout(Duration::from_secs(5), fatal.wait_for(Option::is_some))
450            .await
451            .expect("the controller publishes a fatal fault")
452            .expect("the fault channel stays open until the fault")
453            .clone();
454
455        assert2::check!(fault.is_some_and(|fault| fault.starts_with(&format!(
456            "the metadata log directory {} has failed: ",
457            dir.path().display()
458        ))));
459        ctrl.shutdown().await;
460    }
461
462    /// A controller whose activation records its own leader refuses stops
463    /// with a fatal fault, as Kafka's `QuorumController` halts the process
464    /// with `exception while completing controller activation`. A process
465    /// that hosts it reads the fault from `watch_fatal`.
466    #[tokio::test]
467    async fn a_refused_activation_stops_the_controller() {
468        let dir = TempDir::new().unwrap();
469        let ctrl = Controller::start(ControllerConfig {
470            bootstrap_records: vec![
471                krabka_metadata::MetadataRecord::V1FeatureLevel(
472                    krabka_metadata::FeatureLevelRecord {
473                        name: "metadata.version".into(),
474                        level: crate::LATEST_PRODUCTION_METADATA_VERSION,
475                    },
476                ),
477                // No record creates the topic, so the leader refuses this one.
478                krabka_metadata::MetadataRecord::V1Partition(
479                    test_support::single_replica_partition(
480                        "missing",
481                        krabka_ids::PartitionIndex(0),
482                        NodeId(1),
483                    ),
484                ),
485            ],
486            ..ControllerConfig::for_tests(NodeId(1), dir.path().to_path_buf())
487        })
488        .await
489        .expect("start");
490        let mut fatal = ctrl.watch_fatal();
491        let fault = time::timeout(Duration::from_secs(10), fatal.wait_for(Option::is_some))
492            .await
493            .expect("the controller publishes a fatal fault")
494            .expect("the fault channel stays open until the fault")
495            .clone();
496
497        assert2::check!(
498            fault.as_deref()
499                == Some(
500                    "exception while completing controller activation: metadata: unknown topic 'missing'"
501                )
502        );
503        ctrl.shutdown().await;
504    }
505
506    #[tokio::test]
507    async fn bootstrap_on_non_empty_log_errors() {
508        let (dir, ctrl) = bootstrap_controller("first bootstrap ok").await;
509        // Drive a commit so the log is non-empty on the second boot.
510        wait_for_leader(&ctrl).await;
511        submit_change_with_timeout(
512            &ctrl,
513            vec![krabka_metadata::MetadataRecord::V1Topic(
514                test_support::single_partition_topic("seed", Uuid::new_v4()),
515            )],
516            "bootstrap seed",
517        )
518        .await
519        .expect("submit");
520        ctrl.shutdown().await;
521
522        let cfg2 = ControllerConfig {
523            bootstrap_mode: BootstrapMode::Bootstrap,
524            ..ControllerConfig::for_tests(NodeId(1), dir.path().to_path_buf())
525        };
526        match Controller::start(cfg2).await {
527            Err(err) => assert2::assert!(matches!(err, RaftError::Startup(_))),
528            Ok(ctrl) => {
529                ctrl.shutdown().await;
530                panic!("Bootstrap on existing log must error but succeeded");
531            }
532        }
533    }
534
535    /// A controller listener whose port another socket holds fails the start
536    /// with a listener bind error that names the address and the cause. It is
537    /// not a storage error: the log directory is fine.
538    #[tokio::test]
539    async fn a_taken_controller_port_fails_with_a_listener_bind_error() {
540        let taken = TcpListener::bind("127.0.0.1:0").await.expect("bind");
541        let addr = taken.local_addr().expect("local address");
542        let dir = TempDir::new().unwrap();
543        let cfg = ControllerConfig {
544            controller_listen_addr: addr,
545            ..ControllerConfig::for_tests(NodeId(1), dir.path().to_path_buf())
546        };
547
548        let Err(error) = Controller::start(cfg).await else {
549            panic!("a controller started on a port another socket holds");
550        };
551
552        let message = error.to_string();
553        assert2::assert!(let RaftError::ControllerListenerBind { addr: failed, source } = error);
554        assert2::check!(failed == addr);
555        assert2::check!(source.kind() == ErrorKind::AddrInUse);
556        assert2::check!(
557            message == format!("unable to bind the controller listener to {addr}: {source}")
558        );
559    }
560
561    #[tokio::test]
562    async fn rejoin_on_empty_log_errors() {
563        let dir = TempDir::new().unwrap();
564        let cfg = ControllerConfig {
565            bootstrap_mode: BootstrapMode::Rejoin,
566            ..ControllerConfig::for_tests(NodeId(1), dir.path().to_path_buf())
567        };
568        match Controller::start(cfg).await {
569            Err(err) => assert2::assert!(matches!(err, RaftError::Startup(_))),
570            Ok(ctrl) => {
571                ctrl.shutdown().await;
572                panic!("Rejoin on empty log must error but succeeded");
573            }
574        }
575    }
576
577    #[tokio::test]
578    async fn join_on_empty_log_starts_unattached() {
579        let (_dir, ctrl) = joining_controller("Join on empty log starts ok").await;
580        // Without voters this node never elects.
581        assert2::assert!(ctrl.watch_leader().borrow().is_none());
582        ctrl.shutdown().await;
583    }
584
585    #[tokio::test]
586    async fn bootstrap_with_empty_voters_and_auto_join_falls_back_to_join() {
587        let dir = TempDir::new().unwrap();
588        let mut cfg = ControllerConfig::for_tests(NodeId(1), dir.path().to_path_buf());
589        cfg.bootstrap_mode = BootstrapMode::Bootstrap;
590        cfg.initial_voters = krabka_metadata::VoterSet::from_voters(iter::empty());
591        cfg.auto_join = true;
592        let ctrl = Controller::start(cfg)
593            .await
594            .expect("Bootstrap with empty voters and auto_join should fall back to Join");
595        ctrl.shutdown().await;
596    }
597
598    /// Only a fresh node with no voters and a way to reach the quorum swaps
599    /// `Bootstrap` for `Join`; every other request stands.
600    #[test]
601    fn a_voterless_node_that_reaches_the_quorum_joins_it() {
602        use BootstrapMode::{Bootstrap, Join, Rejoin};
603        // (requested, no initial voters, reaches the quorum, effective)
604        let cases = [
605            (Bootstrap, true, true, Join),
606            (Bootstrap, true, false, Bootstrap),
607            (Bootstrap, false, true, Bootstrap),
608            (Bootstrap, false, false, Bootstrap),
609            (Join, true, true, Join),
610            (Join, false, false, Join),
611            (Rejoin, true, true, Rejoin),
612            (Rejoin, false, false, Rejoin),
613        ];
614        let effective: Vec<_> = cases
615            .iter()
616            .map(|&(requested, no_voters, reaches, _)| {
617                effective_bootstrap_mode(requested, no_voters, reaches)
618            })
619            .collect();
620        let expected: Vec<_> = cases.iter().map(|&(.., mode)| mode).collect();
621        assert2::assert!(effective == expected);
622    }
623
624    /// A controller formatted with `--no-initial-controllers` and pointed at
625    /// the quorum through bootstrap servers starts as an observer, as Kafka's
626    /// does, rather than refusing its empty voter set.
627    #[tokio::test]
628    async fn bootstrap_with_empty_voters_and_bootstrap_servers_falls_back_to_join() {
629        let dir = TempDir::new().unwrap();
630        let cfg = ControllerConfig {
631            bootstrap_mode: BootstrapMode::Bootstrap,
632            initial_voters: krabka_metadata::VoterSet::from_voters(iter::empty()),
633            auto_join: false,
634            bootstrap_servers: vec!["127.0.0.1:1".to_owned()],
635            ..ControllerConfig::for_tests(NodeId(1), dir.path().to_path_buf())
636        };
637        let ctrl = Controller::start(cfg)
638            .await
639            .expect("a voterless controller with bootstrap servers starts as an observer");
640        assert2::assert!(ctrl.watch_leader().borrow().is_none());
641        ctrl.shutdown().await;
642    }
643
644    async fn check_empty_voter_startup_refused(
645        mode: BootstrapMode,
646        auto_join: bool,
647        message: &str,
648    ) {
649        let dir = TempDir::new().unwrap();
650        let cfg = ControllerConfig {
651            bootstrap_mode: mode,
652            initial_voters: krabka_metadata::VoterSet::from_voters(iter::empty()),
653            auto_join,
654            ..ControllerConfig::for_tests(NodeId(1), dir.path().to_path_buf())
655        };
656        let res = Controller::start(cfg).await;
657        assert2::assert!(matches!(res, Err(RaftError::Startup(ref msg)) if msg.contains(message)));
658    }
659
660    #[tokio::test]
661    async fn bootstrap_with_empty_voters_and_no_auto_join_errors() {
662        check_empty_voter_startup_refused(BootstrapMode::Bootstrap, false, "initial_voters set")
663            .await;
664    }
665
666    #[tokio::test]
667    async fn rejoin_with_auto_join_on_empty_log_errors() {
668        check_empty_voter_startup_refused(
669            BootstrapMode::Rejoin,
670            true,
671            "Rejoin mode requires non-empty",
672        )
673        .await;
674    }
675}
krabka_raft/controller/startup.rs

krabka_raft/controller/
startup.rs
