redis_server_wrapper/
sentinel.rs

1//! Redis Sentinel topology management built on `RedisServer`.
2
3use std::collections::HashMap;
4use std::fs;
5use std::path::PathBuf;
6use std::time::Duration;
7use std::time::{SystemTime, UNIX_EPOCH};
8
9use tokio::process::Command;
10
11use crate::cli::RedisCli;
12use crate::error::{Error, Result};
13use crate::server::{RedisServer, RedisServerHandle, SavePolicy};
14
15/// Builder for a Redis Sentinel topology.
16///
17/// # Example
18///
19/// ```no_run
20/// use redis_server_wrapper::RedisSentinel;
21///
22/// # async fn example() {
23/// let sentinel = RedisSentinel::builder()
24///     .master_name("mymaster")
25///     .master_port(6390)
26///     .replicas(2)
27///     .sentinels(3)
28///     .start()
29///     .await
30///     .unwrap();
31///
32/// assert!(sentinel.is_healthy().await);
33/// # }
34/// ```
35pub struct RedisSentinelBuilder {
36    master_name: String,
37    master_port: u16,
38    num_replicas: u16,
39    replica_base_port: u16,
40    num_sentinels: u16,
41    sentinel_base_port: u16,
42    quorum: u16,
43    bind: String,
44    logfile: Option<String>,
45    save: Option<SavePolicy>,
46    appendonly: Option<bool>,
47    down_after_ms: u64,
48    failover_timeout_ms: u64,
49    tls_port: Option<u16>,
50    tls_cert_file: Option<PathBuf>,
51    tls_key_file: Option<PathBuf>,
52    tls_ca_cert_file: Option<PathBuf>,
53    tls_ca_cert_dir: Option<PathBuf>,
54    tls_auth_clients: Option<bool>,
55    tls_replication: Option<bool>,
56    extra: HashMap<String, String>,
57    redis_server_bin: String,
58    redis_cli_bin: String,
59    monitored_masters: Vec<MonitoredMaster>,
60}
61
62#[derive(Clone, Debug, PartialEq, Eq)]
63struct MonitoredMaster {
64    name: String,
65    host: String,
66    port: u16,
67    expected_replicas: u16,
68}
69
70impl RedisSentinelBuilder {
71    /// Set the name of the monitored master (default: `"mymaster"`).
72    pub fn master_name(mut self, name: impl Into<String>) -> Self {
73        self.master_name = name.into();
74        self
75    }
76
77    /// Set the master's port (default: `6390`).
78    pub fn master_port(mut self, port: u16) -> Self {
79        self.master_port = port;
80        self
81    }
82
83    /// Set the number of replicas to start (default: `2`).
84    pub fn replicas(mut self, n: u16) -> Self {
85        self.num_replicas = n;
86        self
87    }
88
89    /// Set the base port for replica nodes (default: `6391`).
90    ///
91    /// Replicas are assigned consecutive ports starting at this value.
92    pub fn replica_base_port(mut self, port: u16) -> Self {
93        self.replica_base_port = port;
94        self
95    }
96
97    /// Set the number of sentinel processes to start (default: `3`).
98    pub fn sentinels(mut self, n: u16) -> Self {
99        self.num_sentinels = n;
100        self
101    }
102
103    /// Set the base port for sentinel processes (default: `26389`).
104    ///
105    /// Sentinels are assigned consecutive ports starting at this value.
106    pub fn sentinel_base_port(mut self, port: u16) -> Self {
107        self.sentinel_base_port = port;
108        self
109    }
110
111    /// Set the quorum count — how many sentinels must agree before a failover is triggered (default: `2`).
112    pub fn quorum(mut self, q: u16) -> Self {
113        self.quorum = q;
114        self
115    }
116
117    /// Set the bind address for all processes in the topology (default: `"127.0.0.1"`).
118    pub fn bind(mut self, bind: impl Into<String>) -> Self {
119        self.bind = bind.into();
120        self
121    }
122
123    /// Set the log file path for all processes in the topology.
124    pub fn logfile(mut self, path: impl Into<String>) -> Self {
125        self.logfile = Some(path.into());
126        self
127    }
128
129    /// Set the `down-after-milliseconds` threshold for all monitored masters (default: `5000`).
130    ///
131    /// A master is considered down after it fails to respond within this many milliseconds.
132    pub fn down_after_ms(mut self, ms: u64) -> Self {
133        self.down_after_ms = ms;
134        self
135    }
136
137    /// Set the `failover-timeout` for all monitored masters in milliseconds (default: `10000`).
138    pub fn failover_timeout_ms(mut self, ms: u64) -> Self {
139        self.failover_timeout_ms = ms;
140        self
141    }
142
143    /// Set the RDB save policy for all data-bearing processes in the topology.
144    ///
145    /// `true` omits the `save` directive (Redis defaults apply).
146    /// `false` emits `save ""` to disable RDB entirely.
147    pub fn save(mut self, save: bool) -> Self {
148        self.save = Some(if save {
149            SavePolicy::Default
150        } else {
151            SavePolicy::Disabled
152        });
153        self
154    }
155
156    /// Set a custom RDB save schedule for all data-bearing processes in the topology.
157    pub fn save_schedule(mut self, schedule: Vec<(u64, u64)>) -> Self {
158        self.save = Some(SavePolicy::Custom(schedule));
159        self
160    }
161
162    /// Enable or disable AOF persistence for all data-bearing processes in the topology.
163    ///
164    /// When not set, the builder defaults to `appendonly yes` for the master
165    /// and replicas.
166    pub fn appendonly(mut self, appendonly: bool) -> Self {
167        self.appendonly = Some(appendonly);
168        self
169    }
170
171    // -- TLS directives --
172
173    /// Set the TLS listening port for the master and replica nodes.
174    pub fn tls_port(mut self, port: u16) -> Self {
175        self.tls_port = Some(port);
176        self
177    }
178
179    /// Set the TLS certificate file path for all nodes.
180    pub fn tls_cert_file(mut self, path: impl Into<PathBuf>) -> Self {
181        self.tls_cert_file = Some(path.into());
182        self
183    }
184
185    /// Set the TLS private key file path for all nodes.
186    pub fn tls_key_file(mut self, path: impl Into<PathBuf>) -> Self {
187        self.tls_key_file = Some(path.into());
188        self
189    }
190
191    /// Set the TLS CA certificate file path for all nodes.
192    pub fn tls_ca_cert_file(mut self, path: impl Into<PathBuf>) -> Self {
193        self.tls_ca_cert_file = Some(path.into());
194        self
195    }
196
197    /// Set the TLS CA certificate directory for all nodes.
198    pub fn tls_ca_cert_dir(mut self, path: impl Into<PathBuf>) -> Self {
199        self.tls_ca_cert_dir = Some(path.into());
200        self
201    }
202
203    /// Require TLS client authentication for all nodes.
204    pub fn tls_auth_clients(mut self, auth: bool) -> Self {
205        self.tls_auth_clients = Some(auth);
206        self
207    }
208
209    /// Use TLS for replication traffic between nodes.
210    pub fn tls_replication(mut self, enable: bool) -> Self {
211        self.tls_replication = Some(enable);
212        self
213    }
214
215    /// Set an arbitrary config directive for all processes in the topology.
216    pub fn extra(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
217        self.extra.insert(key.into(), value.into());
218        self
219    }
220
221    /// Set a custom `redis-server` binary path.
222    pub fn redis_server_bin(mut self, bin: impl Into<String>) -> Self {
223        self.redis_server_bin = bin.into();
224        self
225    }
226
227    /// Set a custom `redis-cli` binary path.
228    pub fn redis_cli_bin(mut self, bin: impl Into<String>) -> Self {
229        self.redis_cli_bin = bin.into();
230        self
231    }
232
233    /// Add an additional master for the sentinels to monitor.
234    ///
235    /// The builder-managed topology still creates the primary master configured by
236    /// [`Self::master_name`] and [`Self::master_port`]. Additional monitored
237    /// masters are expected to already be running.
238    pub fn monitor(mut self, name: impl Into<String>, host: impl Into<String>, port: u16) -> Self {
239        self.monitored_masters.push(MonitoredMaster {
240            name: name.into(),
241            host: host.into(),
242            port,
243            expected_replicas: 0,
244        });
245        self
246    }
247
248    /// Add an additional master and the minimum number of replicas expected for it.
249    pub fn monitor_with_replicas(
250        mut self,
251        name: impl Into<String>,
252        host: impl Into<String>,
253        port: u16,
254        expected_replicas: u16,
255    ) -> Self {
256        self.monitored_masters.push(MonitoredMaster {
257            name: name.into(),
258            host: host.into(),
259            port,
260            expected_replicas,
261        });
262        self
263    }
264
265    /// Whether TLS is configured (cert + key present).
266    fn has_tls(&self) -> bool {
267        self.tls_cert_file.is_some() && self.tls_key_file.is_some()
268    }
269
270    /// Apply TLS flags to a CLI instance based on builder config.
271    fn apply_tls_to_cli(&self, mut cli: RedisCli) -> RedisCli {
272        if self.has_tls() {
273            cli = cli.tls(true);
274            if let Some(ref ca) = self.tls_ca_cert_file {
275                cli = cli.cacert(ca);
276            } else {
277                cli = cli.insecure(true);
278            }
279            if let Some(ref cert) = self.tls_cert_file {
280                cli = cli.cert(cert);
281            }
282            if let Some(ref key) = self.tls_key_file {
283                cli = cli.key(key);
284            }
285        }
286        cli
287    }
288
289    /// Apply TLS config to a server builder.
290    fn apply_tls_to_server(&self, mut server: RedisServer) -> RedisServer {
291        if let Some(port) = self.tls_port {
292            server = server.tls_port(port);
293        }
294        if let Some(ref path) = self.tls_cert_file {
295            server = server.tls_cert_file(path);
296        }
297        if let Some(ref path) = self.tls_key_file {
298            server = server.tls_key_file(path);
299        }
300        if let Some(ref path) = self.tls_ca_cert_file {
301            server = server.tls_ca_cert_file(path);
302        }
303        if let Some(ref path) = self.tls_ca_cert_dir {
304            server = server.tls_ca_cert_dir(path);
305        }
306        if let Some(v) = self.tls_auth_clients {
307            server = server.tls_auth_clients(v);
308        }
309        if let Some(v) = self.tls_replication {
310            server = server.tls_replication(v);
311        }
312        server
313    }
314
315    fn replica_ports(&self) -> impl Iterator<Item = u16> {
316        let base = self.replica_base_port;
317        let n = self.num_replicas;
318        (0..n).map(move |i| base + i)
319    }
320
321    fn sentinel_ports(&self) -> impl Iterator<Item = u16> {
322        let base = self.sentinel_base_port;
323        let n = self.num_sentinels;
324        (0..n).map(move |i| base + i)
325    }
326
327    /// Start the full topology: master, replicas, sentinels.
328    pub async fn start(self) -> Result<RedisSentinelHandle> {
329        let mut monitored_masters = Vec::with_capacity(1 + self.monitored_masters.len());
330        monitored_masters.push(MonitoredMaster {
331            name: self.master_name.clone(),
332            host: self.bind.clone(),
333            port: self.master_port,
334            expected_replicas: self.num_replicas,
335        });
336        monitored_masters.extend(self.monitored_masters.iter().cloned());
337
338        // Kill leftover processes.
339        let cli_for_shutdown = |port: u16| {
340            let cli = self.apply_tls_to_cli(
341                RedisCli::new()
342                    .bin(&self.redis_cli_bin)
343                    .host(&self.bind)
344                    .port(port),
345            );
346            cli.shutdown();
347        };
348        cli_for_shutdown(self.master_port);
349        for port in self.replica_ports() {
350            cli_for_shutdown(port);
351        }
352        for port in self.sentinel_ports() {
353            cli_for_shutdown(port);
354        }
355        tokio::time::sleep(Duration::from_millis(500)).await;
356
357        let unique = SystemTime::now()
358            .duration_since(UNIX_EPOCH)
359            .map(|duration| duration.as_nanos())
360            .unwrap_or(0);
361        let base_dir = std::env::temp_dir().join(format!(
362            "redis-sentinel-wrapper-{}-{}",
363            std::process::id(),
364            unique
365        ));
366        fs::create_dir_all(&base_dir)?;
367
368        // 1. Start master.
369        let appendonly = self.appendonly.unwrap_or(true);
370        let mut master = RedisServer::new()
371            .port(self.master_port)
372            .bind(&self.bind)
373            .dir(base_dir.join("master"))
374            .appendonly(appendonly)
375            .redis_server_bin(&self.redis_server_bin)
376            .redis_cli_bin(&self.redis_cli_bin);
377        master = self.apply_tls_to_server(master);
378        if let Some(ref logfile) = self.logfile {
379            master = master.logfile(logfile.clone());
380        }
381        if let Some(ref save) = self.save {
382            match save {
383                SavePolicy::Disabled => master = master.save(false),
384                SavePolicy::Default => master = master.save(true),
385                SavePolicy::Custom(pairs) => master = master.save_schedule(pairs.clone()),
386            }
387        }
388        for (key, value) in &self.extra {
389            master = master.extra(key.clone(), value.clone());
390        }
391        let master = master.start().await?;
392
393        // 2. Start replicas.
394        let mut replicas = Vec::new();
395        for port in self.replica_ports() {
396            let mut replica = RedisServer::new()
397                .port(port)
398                .bind(&self.bind)
399                .dir(base_dir.join(format!("replica-{port}")))
400                .appendonly(appendonly)
401                .replicaof(self.bind.clone(), self.master_port)
402                .redis_server_bin(&self.redis_server_bin)
403                .redis_cli_bin(&self.redis_cli_bin);
404            replica = self.apply_tls_to_server(replica);
405            if let Some(ref logfile) = self.logfile {
406                replica = replica.logfile(logfile.clone());
407            }
408            if let Some(ref save) = self.save {
409                match save {
410                    SavePolicy::Disabled => replica = replica.save(false),
411                    SavePolicy::Default => replica = replica.save(true),
412                    SavePolicy::Custom(pairs) => {
413                        replica = replica.save_schedule(pairs.clone());
414                    }
415                }
416            }
417            for (key, value) in &self.extra {
418                replica = replica.extra(key.clone(), value.clone());
419            }
420            let replica = replica.start().await?;
421            replicas.push(replica);
422        }
423
424        // Let replication link up.
425        tokio::time::sleep(Duration::from_secs(1)).await;
426
427        // 3. Start sentinels.
428        let mut sentinel_handles = Vec::new();
429        for port in self.sentinel_ports() {
430            let dir = base_dir.join(format!("sentinel-{port}"));
431            fs::create_dir_all(&dir)?;
432            let conf_path = dir.join("sentinel.conf");
433            let logfile = self
434                .logfile
435                .as_deref()
436                .map(str::to_owned)
437                .unwrap_or_else(|| format!("{}/sentinel.log", dir.display()));
438            let mut conf = format!(
439                "port {port}\n\
440                 bind {bind}\n\
441                 daemonize yes\n\
442                 pidfile \"{dir}/sentinel.pid\"\n\
443                 logfile \"{logfile}\"\n\
444                 dir \"{dir}\"\n",
445                port = port,
446                bind = self.bind,
447                dir = dir.display(),
448                logfile = logfile,
449            );
450            for master in &monitored_masters {
451                conf.push_str(&format!(
452                    "sentinel monitor {name} {host} {master_port} {quorum}\n\
453                     sentinel down-after-milliseconds {name} {down_after}\n\
454                     sentinel failover-timeout {name} {failover_timeout}\n\
455                     sentinel parallel-syncs {name} 1\n",
456                    name = master.name,
457                    host = master.host,
458                    master_port = master.port,
459                    quorum = self.quorum,
460                    down_after = self.down_after_ms,
461                    failover_timeout = self.failover_timeout_ms,
462                ));
463            }
464            // TLS directives for sentinels.
465            if let Some(ref path) = self.tls_cert_file {
466                conf.push_str(&format!("tls-cert-file \"{}\"\n", path.display()));
467            }
468            if let Some(ref path) = self.tls_key_file {
469                conf.push_str(&format!("tls-key-file \"{}\"\n", path.display()));
470            }
471            if let Some(ref path) = self.tls_ca_cert_file {
472                conf.push_str(&format!("tls-ca-cert-file \"{}\"\n", path.display()));
473            }
474            if let Some(ref path) = self.tls_ca_cert_dir {
475                conf.push_str(&format!("tls-ca-cert-dir \"{}\"\n", path.display()));
476            }
477            if let Some(tls_port) = self.tls_port {
478                conf.push_str(&format!("tls-port {tls_port}\n"));
479            }
480            if let Some(v) = self.tls_auth_clients {
481                conf.push_str(&format!(
482                    "tls-auth-clients {}\n",
483                    if v { "yes" } else { "no" }
484                ));
485            }
486            if let Some(v) = self.tls_replication {
487                conf.push_str(&format!(
488                    "tls-replication {}\n",
489                    if v { "yes" } else { "no" }
490                ));
491            }
492            for (key, value) in &self.extra {
493                conf.push_str(&format!("{key} {value}\n"));
494            }
495            fs::write(&conf_path, conf)?;
496
497            let status = Command::new(&self.redis_server_bin)
498                .arg(&conf_path)
499                .arg("--sentinel")
500                .stdout(std::process::Stdio::null())
501                .stderr(std::process::Stdio::null())
502                .status()
503                .await?;
504
505            if !status.success() {
506                return Err(Error::SentinelStart { port });
507            }
508
509            let cli = self.apply_tls_to_cli(
510                RedisCli::new()
511                    .bin(&self.redis_cli_bin)
512                    .host(&self.bind)
513                    .port(port),
514            );
515            cli.wait_for_ready(Duration::from_secs(10)).await?;
516
517            let pid_path = dir.join("sentinel.pid");
518            let pid: u32 = {
519                let deadline = std::time::Instant::now() + std::time::Duration::from_secs(1);
520                loop {
521                    if let Ok(s) = fs::read_to_string(&pid_path)
522                        && let Ok(p) = s.trim().parse::<u32>()
523                    {
524                        break p;
525                    }
526                    if std::time::Instant::now() >= deadline {
527                        return Err(Error::SentinelStart { port });
528                    }
529                    tokio::time::sleep(std::time::Duration::from_millis(50)).await;
530                }
531            };
532
533            sentinel_handles.push((port, pid, cli));
534        }
535
536        // Wait for sentinels to discover each other.
537        tokio::time::sleep(Duration::from_secs(2)).await;
538
539        Ok(RedisSentinelHandle {
540            master,
541            replicas,
542            sentinel_ports: sentinel_handles.iter().map(|(p, _, _)| *p).collect(),
543            sentinel_pids: sentinel_handles.iter().map(|(_, pid, _)| *pid).collect(),
544            master_name: self.master_name,
545            bind: self.bind,
546            redis_cli_bin: self.redis_cli_bin,
547            num_sentinels: self.num_sentinels,
548            monitored_masters,
549            tls: TlsConfig {
550                cert_file: self.tls_cert_file,
551                key_file: self.tls_key_file,
552                ca_cert_file: self.tls_ca_cert_file,
553            },
554        })
555    }
556}
557
558/// TLS configuration snapshot stored in the handle for building CLI instances.
559#[derive(Clone, Debug, Default)]
560struct TlsConfig {
561    cert_file: Option<PathBuf>,
562    key_file: Option<PathBuf>,
563    ca_cert_file: Option<PathBuf>,
564}
565
566impl TlsConfig {
567    fn has_tls(&self) -> bool {
568        self.cert_file.is_some() && self.key_file.is_some()
569    }
570
571    fn apply(&self, mut cli: RedisCli) -> RedisCli {
572        if self.has_tls() {
573            cli = cli.tls(true);
574            if let Some(ref ca) = self.ca_cert_file {
575                cli = cli.cacert(ca);
576            } else {
577                cli = cli.insecure(true);
578            }
579            if let Some(ref cert) = self.cert_file {
580                cli = cli.cert(cert);
581            }
582            if let Some(ref key) = self.key_file {
583                cli = cli.key(key);
584            }
585        }
586        cli
587    }
588}
589
590/// A running Redis Sentinel topology. Stops everything on Drop.
591pub struct RedisSentinelHandle {
592    master: RedisServerHandle,
593    #[allow(dead_code)] // Kept alive for Drop cleanup
594    replicas: Vec<RedisServerHandle>,
595    sentinel_ports: Vec<u16>,
596    sentinel_pids: Vec<u32>,
597    master_name: String,
598    bind: String,
599    redis_cli_bin: String,
600    num_sentinels: u16,
601    monitored_masters: Vec<MonitoredMaster>,
602    tls: TlsConfig,
603}
604
605/// Entry point for building a Redis Sentinel topology.
606///
607/// Call [`RedisSentinel::builder`] to obtain a [`RedisSentinelBuilder`], then
608/// configure it and call [`RedisSentinelBuilder::start`] to launch the topology.
609pub struct RedisSentinel;
610
611impl RedisSentinel {
612    /// Create a new sentinel builder with defaults.
613    pub fn builder() -> RedisSentinelBuilder {
614        RedisSentinelBuilder {
615            master_name: "mymaster".into(),
616            master_port: 6390,
617            num_replicas: 2,
618            replica_base_port: 6391,
619            num_sentinels: 3,
620            sentinel_base_port: 26389,
621            quorum: 2,
622            bind: "127.0.0.1".into(),
623            logfile: None,
624            save: None,
625            appendonly: None,
626            down_after_ms: 5000,
627            failover_timeout_ms: 10000,
628            tls_port: None,
629            tls_cert_file: None,
630            tls_key_file: None,
631            tls_ca_cert_file: None,
632            tls_ca_cert_dir: None,
633            tls_auth_clients: None,
634            tls_replication: None,
635            extra: HashMap::new(),
636            redis_server_bin: "redis-server".into(),
637            redis_cli_bin: "redis-cli".into(),
638            monitored_masters: Vec::new(),
639        }
640    }
641}
642
643impl RedisSentinelHandle {
644    /// The master's address as "host:port". Kept consistent with
645    /// [`RedisServerHandle::addr`](crate::server::RedisServerHandle::addr) and
646    /// [`RedisClusterHandle::addr`](crate::cluster::RedisClusterHandle::addr).
647    pub fn addr(&self) -> String {
648        self.master.addr()
649    }
650
651    /// The master's address. Alias for [`Self::addr`].
652    pub fn master_addr(&self) -> String {
653        self.master.addr()
654    }
655
656    /// Get a `RedisCli` for the seed sentinel (first sentinel process).
657    pub fn cli(&self) -> RedisCli {
658        self.tls.apply(
659            RedisCli::new()
660                .bin(&self.redis_cli_bin)
661                .host(&self.bind)
662                .port(self.sentinel_ports[0]),
663        )
664    }
665
666    /// All monitored master names.
667    pub fn monitored_master_names(&self) -> Vec<&str> {
668        self.monitored_masters
669            .iter()
670            .map(|master| master.name.as_str())
671            .collect()
672    }
673
674    /// All monitored master addresses.
675    pub fn monitored_master_addrs(&self) -> Vec<String> {
676        self.monitored_masters
677            .iter()
678            .map(|master| format!("{}:{}", master.host, master.port))
679            .collect()
680    }
681
682    /// The PIDs of all processes in the topology (master, replicas, sentinels).
683    pub fn pids(&self) -> Vec<u32> {
684        let mut pids = Vec::with_capacity(1 + self.replicas.len() + self.sentinel_pids.len());
685        pids.push(self.master.pid());
686        for replica in &self.replicas {
687            pids.push(replica.pid());
688        }
689        pids.extend_from_slice(&self.sentinel_pids);
690        pids
691    }
692
693    /// All sentinel addresses.
694    pub fn sentinel_addrs(&self) -> Vec<String> {
695        self.sentinel_ports
696            .iter()
697            .map(|p| format!("{}:{}", self.bind, p))
698            .collect()
699    }
700
701    /// The monitored master name.
702    pub fn master_name(&self) -> &str {
703        &self.master_name
704    }
705
706    /// Query a sentinel for the primary monitored master's status.
707    ///
708    /// Iterates over the sentinel processes until one responds, then runs
709    /// `SENTINEL MASTER <name>` and returns the result as a flat key/value map.
710    ///
711    /// Common keys in the returned map include `"ip"`, `"port"`, `"flags"`,
712    /// `"num-slaves"`, and `"num-other-sentinels"`.
713    ///
714    /// Returns [`Error::NoReachableSentinel`] if no sentinel responds.
715    pub async fn poke(&self) -> Result<HashMap<String, String>> {
716        self.poke_master(&self.master_name).await
717    }
718
719    /// Query a sentinel for a specific monitored master's status.
720    ///
721    /// Like [`poke`](Self::poke) but targets `master_name` instead of the
722    /// primary master configured for this topology.
723    pub async fn poke_master(&self, master_name: &str) -> Result<HashMap<String, String>> {
724        for port in &self.sentinel_ports {
725            let cli = self.tls.apply(
726                RedisCli::new()
727                    .bin(&self.redis_cli_bin)
728                    .host(&self.bind)
729                    .port(*port),
730            );
731            if let Ok(raw) = cli.run(&["SENTINEL", "MASTER", master_name]).await {
732                return Ok(parse_flat_kv(&raw));
733            }
734        }
735        Err(Error::NoReachableSentinel)
736    }
737
738    /// Check if the topology is healthy.
739    pub async fn is_healthy(&self) -> bool {
740        for master in &self.monitored_masters {
741            let Ok(info) = self.poke_master(&master.name).await else {
742                return false;
743            };
744            let flags = info.get("flags").map(|s| s.as_str()).unwrap_or("");
745            let num_slaves: u64 = info
746                .get("num-slaves")
747                .and_then(|v| v.parse().ok())
748                .unwrap_or(0);
749            let num_sentinels: u64 = info
750                .get("num-other-sentinels")
751                .and_then(|v| v.parse().ok())
752                .unwrap_or(0)
753                + 1;
754            if flags != "master"
755                || num_slaves < master.expected_replicas as u64
756                || num_sentinels < self.num_sentinels as u64
757            {
758                return false;
759            }
760        }
761        true
762    }
763
764    /// Wait until the topology is healthy or timeout.
765    pub async fn wait_for_healthy(&self, timeout: Duration) -> Result<()> {
766        let start = std::time::Instant::now();
767        loop {
768            if self.is_healthy().await {
769                return Ok(());
770            }
771            if start.elapsed() > timeout {
772                return Err(Error::Timeout {
773                    message: "sentinel topology did not become healthy in time".into(),
774                });
775            }
776            tokio::time::sleep(Duration::from_millis(500)).await;
777        }
778    }
779
780    /// Stop everything via an escalating shutdown strategy.
781    ///
782    /// 1. Sends `SHUTDOWN NOSAVE` to each sentinel process.
783    /// 2. Waits 500ms for them to exit.
784    /// 3. For each sentinel PID that is still alive, calls [`crate::process::force_kill`].
785    /// 4. Calls [`crate::process::kill_by_port`] for each sentinel port as a safety net.
786    ///
787    /// Replicas and the master are stopped by their own handles' [`Drop`] impls,
788    /// which also use the escalating strategy.
789    pub fn stop(&self) {
790        // Step 1: graceful shutdown for each sentinel.
791        for port in &self.sentinel_ports {
792            self.tls
793                .apply(
794                    RedisCli::new()
795                        .bin(&self.redis_cli_bin)
796                        .host(&self.bind)
797                        .port(*port),
798                )
799                .shutdown();
800        }
801        // Step 2: grace period.
802        std::thread::sleep(std::time::Duration::from_millis(500));
803        // Step 3: force kill any sentinels still alive.
804        for pid in &self.sentinel_pids {
805            if crate::process::pid_alive(*pid) {
806                crate::process::force_kill(*pid);
807            }
808        }
809        // Step 4: port cleanup as safety net.
810        for port in &self.sentinel_ports {
811            crate::process::kill_by_port(*port);
812        }
813        // Replicas and master stopped by their handles' Drop.
814    }
815}
816
817impl Drop for RedisSentinelHandle {
818    fn drop(&mut self) {
819        self.stop();
820    }
821}
822
823/// Parse alternating key/value lines from sentinel output.
824fn parse_flat_kv(raw: &str) -> HashMap<String, String> {
825    let lines: Vec<&str> = raw.lines().map(|l| l.trim()).collect();
826    let mut map = HashMap::new();
827    let mut i = 0;
828    while i + 1 < lines.len() {
829        map.insert(lines[i].to_string(), lines[i + 1].to_string());
830        i += 2;
831    }
832    map
833}
834
835#[cfg(test)]
836mod tests {
837    use super::*;
838
839    #[test]
840    fn builder_defaults() {
841        let b = RedisSentinel::builder();
842        assert_eq!(b.master_port, 6390);
843        assert_eq!(b.num_replicas, 2);
844        assert_eq!(b.num_sentinels, 3);
845        assert_eq!(b.quorum, 2);
846        assert!(b.logfile.is_none());
847        assert!(b.extra.is_empty());
848        assert!(b.monitored_masters.is_empty());
849    }
850
851    #[test]
852    fn builder_chain() {
853        let b = RedisSentinel::builder()
854            .master_name("custom")
855            .master_port(6500)
856            .replicas(1)
857            .sentinels(5)
858            .quorum(3)
859            .logfile("/tmp/sentinel.log")
860            .extra("maxmemory", "10mb")
861            .monitor("backup", "127.0.0.1", 6501);
862        assert_eq!(b.master_name, "custom");
863        assert_eq!(b.master_port, 6500);
864        assert_eq!(b.num_replicas, 1);
865        assert_eq!(b.num_sentinels, 5);
866        assert_eq!(b.quorum, 3);
867        assert_eq!(b.logfile.as_deref(), Some("/tmp/sentinel.log"));
868        assert_eq!(b.extra.get("maxmemory").map(String::as_str), Some("10mb"));
869        assert_eq!(b.monitored_masters.len(), 1);
870        assert_eq!(
871            b.monitored_masters[0],
872            MonitoredMaster {
873                name: "backup".into(),
874                host: "127.0.0.1".into(),
875                port: 6501,
876                expected_replicas: 0,
877            }
878        );
879    }
880
881    #[test]
882    fn parse_sentinel_output() {
883        let raw = "name\nmymaster\nip\n127.0.0.1\nport\n6380\n";
884        let map = parse_flat_kv(raw);
885        assert_eq!(map.get("name").unwrap(), "mymaster");
886        assert_eq!(map.get("ip").unwrap(), "127.0.0.1");
887        assert_eq!(map.get("port").unwrap(), "6380");
888    }
889}
redis_server_wrapper/sentinel.rs

redis_server_wrapper/
sentinel.rs
