1use std::{
2 borrow::Cow,
3 net::SocketAddr,
4 path::{Path, PathBuf},
5 process::Stdio,
6 sync::atomic::{AtomicU32, Ordering},
7 time::{Duration, Instant},
8};
9
10use color_eyre::eyre::{eyre, Report};
11use tokio::{
12 io::{AsyncBufReadExt, BufReader},
13 process::{Child, Command},
14 sync::watch,
15 task::JoinHandle,
16 time::{sleep, timeout},
17};
18use tracing::{debug, error, info, warn};
19
20use zebra_chain::parameters::NetworkKind;
21
22use super::{effective_zcashd_datadir, ensure_zcashd_datadir, resolve_zcashd_datadir_path, Config};
23
24const SUPERVISOR_ACTIVE_METRIC: &str = "zcashd.compat.supervisor.active";
25const SUPERVISOR_DISABLED_METRIC: &str = "zcashd.compat.supervisor.disabled";
26const SUPERVISOR_EXHAUSTED_METRIC: &str = "zcashd.compat.supervisor.exhausted";
27
28static SUPERVISED_ZCASHD_PID: AtomicU32 = AtomicU32::new(0);
37
38#[derive(Clone, Debug)]
40pub struct SupervisorConfig {
41 pub zcashd_path: PathBuf,
43 pub zcashd_datadir: PathBuf,
45 pub zebra_p2p_addr: SocketAddr,
48 pub extra_args: Vec<String>,
50 pub network: NetworkKind,
52 pub startup_delay: std::time::Duration,
54 pub restart_backoff: Duration,
56 pub restart_backoff_max: Duration,
58 pub restart_reset_after: Duration,
60 pub shutdown_grace_period: Duration,
62}
63
64impl SupervisorConfig {
65 pub fn new(
67 zcashd_compat: &Config,
68 zcashd_path: PathBuf,
69 state_cache_dir: &Path,
70 network: NetworkKind,
71 zebra_p2p_addr: SocketAddr,
72 ) -> Self {
73 let extra_args = zcashd_compat.zcashd_extra_args.clone();
74 let zcashd_datadir = resolve_zcashd_datadir_path(
75 &effective_zcashd_datadir(zcashd_compat, state_cache_dir),
76 &extra_args,
77 );
78
79 Self {
80 zcashd_path,
81 zcashd_datadir,
82 zebra_p2p_addr,
83 extra_args,
84 network,
85 startup_delay: zcashd_compat.startup_delay,
86 restart_backoff: zcashd_compat.restart_backoff,
87 restart_backoff_max: zcashd_compat.restart_backoff_max,
88 restart_reset_after: zcashd_compat.restart_reset_after,
89 shutdown_grace_period: zcashd_compat.shutdown_grace_period,
90 }
91 }
92
93 pub fn command_args(&self) -> Vec<String> {
95 let mut args = vec![format!(
96 "-datadir={}",
97 self.zcashd_datadir.to_string_lossy()
98 )];
99
100 match self.network {
101 NetworkKind::Mainnet => {}
102 NetworkKind::Testnet => args.push("-testnet".to_string()),
103 NetworkKind::Regtest => {
104 args.push("-regtest".to_string());
105 args.push("-regtestacceptunvalidatedpow".to_string());
109 }
110 }
111
112 args.push("-printtoconsole".to_string());
114 args.extend(
115 self.extra_args
116 .iter()
117 .filter(|arg| arg.as_str() != "-printtoconsole")
118 .cloned(),
119 );
120
121 args.push(format!("-connect={}", self.zebra_p2p_addr));
136 args.push("-listen=0".to_string());
137 args.push("-dnsseed=0".to_string());
138 args.push("-listenonion=0".to_string());
139 args.push("-discover=0".to_string());
140 args.push("-daemon=0".to_string());
141
142 args
143 }
144}
145
146const PEER_SELECTION_OPTIONS: &[&str] = &["connect", "addnode", "seednode"];
149
150pub fn reject_peer_selection_extra_args(extra_args: &[String]) -> Result<(), Report> {
162 for arg in extra_args {
163 let name = arg
164 .trim_start_matches('-')
165 .split('=')
166 .next()
167 .unwrap_or_default()
168 .to_ascii_lowercase();
169 let name = name.strip_prefix("no").unwrap_or(&name);
170
171 if PEER_SELECTION_OPTIONS.contains(&name) {
172 return Err(eyre!(
173 "zcashd_compat.zcashd_extra_args contains {arg:?}: peer-selection options are not \
174 allowed because the zcashd P2P sidecar must connect only to the local Zebra node"
175 ));
176 }
177 }
178
179 Ok(())
180}
181
182pub async fn run(
194 config: SupervisorConfig,
195 mut shutdown_rx: watch::Receiver<bool>,
196) -> Result<(), Report> {
197 reject_peer_selection_extra_args(&config.extra_args)?;
198 if config.restart_backoff.is_zero() || config.restart_backoff_max.is_zero() {
203 return Err(eyre!(
204 "zcashd_compat.restart_backoff and zcashd_compat.restart_backoff_max must both be \
205 greater than zero to avoid a hot zcashd restart loop"
206 ));
207 }
208 ensure_zcashd_datadir(&config.zcashd_datadir, &config.extra_args)?;
209 set_supervision_active_metrics();
210
211 if wait_for_delay_or_shutdown(config.startup_delay, &mut shutdown_rx).await {
212 info!("zcashd-compat supervisor received shutdown during startup delay");
213 set_supervision_inactive_metrics();
214 return Ok(());
215 }
216
217 let mut consecutive_restart_count = 0u32;
218
219 loop {
220 if *shutdown_rx.borrow() {
221 info!("zcashd-compat supervisor received shutdown before spawn");
222 set_supervision_inactive_metrics();
223 return Ok(());
224 }
225
226 let mut child = match spawn_zcashd(&config) {
227 Ok(child) => child,
228 Err(error) => {
229 consecutive_restart_count = consecutive_restart_count.saturating_add(1);
230 warn!(
231 %error,
232 restart_count = consecutive_restart_count,
233 "failed to spawn zcashd-compat zcashd child, retrying after backoff"
234 );
235
236 let restart_delay = restart_backoff_delay(
237 config.restart_backoff,
238 config.restart_backoff_max,
239 consecutive_restart_count,
240 );
241 if wait_for_delay_or_shutdown(restart_delay, &mut shutdown_rx).await {
242 info!("zcashd-compat supervisor received shutdown during spawn retry backoff");
243 set_supervision_inactive_metrics();
244 return Ok(());
245 }
246 continue;
247 }
248 };
249 let child_started_at = Instant::now();
250 SUPERVISED_ZCASHD_PID.store(child.id().unwrap_or(0), Ordering::SeqCst);
251 info!(
252 path = %config.zcashd_path.display(),
253 datadir = %config.zcashd_datadir.display(),
254 connect = %config.zebra_p2p_addr,
255 "started zcashd-compat zcashd child"
256 );
257
258 let child_result = wait_for_child_or_shutdown(&mut child, &mut shutdown_rx).await;
259 match child_result {
260 ChildOutcome::ShutdownRequested => {
261 info!(
262 pid = ?child.id(),
263 grace_period = ?config.shutdown_grace_period,
264 "zcashd-compat supervisor received shutdown request; terminating zcashd child"
265 );
266 let terminate_result =
267 terminate_child(&mut child, config.shutdown_grace_period).await;
268 if terminate_result.is_ok() {
269 SUPERVISED_ZCASHD_PID.store(0, Ordering::SeqCst);
273 }
274 terminate_result?;
275 info!("zcashd-compat zcashd child stopped on shutdown");
276 set_supervision_inactive_metrics();
277 return Ok(());
278 }
279 ChildOutcome::Exited(status) => {
280 SUPERVISED_ZCASHD_PID.store(0, Ordering::SeqCst);
281 let child_uptime = child_started_at.elapsed();
282 if should_reset_restart_count(child_uptime, config.restart_reset_after) {
283 info!(
284 ?status,
285 child_uptime_secs = child_uptime.as_secs(),
286 restart_reset_after_secs = config.restart_reset_after.as_secs(),
287 previous_restart_count = consecutive_restart_count,
288 "zcashd-compat zcashd child had healthy uptime, resetting restart count"
289 );
290 consecutive_restart_count = 0;
291 }
292
293 consecutive_restart_count = consecutive_restart_count.saturating_add(1);
294 warn!(
295 ?status,
296 restart_count = consecutive_restart_count,
297 child_uptime_secs = child_uptime.as_secs(),
298 "zcashd-compat zcashd child exited before shutdown, restarting"
299 );
300
301 let restart_delay = restart_backoff_delay(
302 config.restart_backoff,
303 config.restart_backoff_max,
304 consecutive_restart_count,
305 );
306 if wait_for_delay_or_shutdown(restart_delay, &mut shutdown_rx).await {
307 info!("zcashd-compat supervisor received shutdown during restart backoff");
308 set_supervision_inactive_metrics();
309 return Ok(());
310 }
311 }
312 ChildOutcome::WaitFailed => {
313 warn!(
318 "failed waiting on zcashd-compat zcashd child; \
319 terminating it before restart"
320 );
321 if let Err(error) = terminate_child(&mut child, config.shutdown_grace_period).await
322 {
323 set_supervision_inactive_metrics();
329 return Err(eyre!(
330 "could not reap zcashd-compat zcashd after a wait error ({error}); \
331 stopping supervision to avoid running two instances on one datadir"
332 ));
333 }
334 SUPERVISED_ZCASHD_PID.store(0, Ordering::SeqCst);
335
336 consecutive_restart_count = consecutive_restart_count.saturating_add(1);
337 let restart_delay = restart_backoff_delay(
338 config.restart_backoff,
339 config.restart_backoff_max,
340 consecutive_restart_count,
341 );
342 if wait_for_delay_or_shutdown(restart_delay, &mut shutdown_rx).await {
343 info!("zcashd-compat supervisor received shutdown during restart backoff");
344 set_supervision_inactive_metrics();
345 return Ok(());
346 }
347 }
348 }
349 }
350}
351
352pub fn terminate_abandoned_zcashd(shutdown_grace_period: Duration) {
365 let pid = SUPERVISED_ZCASHD_PID.swap(0, Ordering::SeqCst);
366 if pid == 0 {
367 return;
368 }
369
370 #[cfg(unix)]
371 {
372 use nix::{
373 sys::{
374 signal::{kill, Signal::SIGKILL, Signal::SIGTERM},
375 wait::{waitpid, WaitPidFlag, WaitStatus},
376 },
377 unistd::Pid,
378 };
379
380 let Ok(pid) = i32::try_from(pid) else {
381 warn!(
384 pid,
385 "abandoned zcashd-compat zcashd pid does not fit in i32"
386 );
387 return;
388 };
389 let pid = Pid::from_raw(pid);
390
391 let child_has_exited = || match waitpid(pid, Some(WaitPidFlag::WNOHANG)) {
398 Ok(WaitStatus::StillAlive) => false,
399 Ok(_) | Err(_) => true,
401 };
402
403 if child_has_exited() {
406 return;
407 }
408
409 info!(
410 %pid,
411 grace_period = ?shutdown_grace_period,
412 "terminating zcashd-compat zcashd child abandoned by runtime shutdown"
413 );
414 if kill(pid, SIGTERM).is_err() || child_has_exited() {
415 return;
416 }
417
418 const POLL_INTERVAL: Duration = Duration::from_millis(200);
419 let deadline = Instant::now() + shutdown_grace_period;
420 while Instant::now() < deadline {
421 std::thread::sleep(POLL_INTERVAL);
422 if child_has_exited() {
423 info!(%pid, "abandoned zcashd-compat zcashd child exited after SIGTERM");
424 return;
425 }
426 }
427
428 warn!(
429 %pid,
430 grace_period = ?shutdown_grace_period,
431 "abandoned zcashd-compat zcashd child did not exit within the grace period; \
432 sending SIGKILL as a last resort"
433 );
434 let _ = kill(pid, SIGKILL);
435 let _ = waitpid(pid, None);
436 }
437
438 #[cfg(not(unix))]
439 {
440 let _ = (pid, shutdown_grace_period);
441 }
442}
443
444pub fn set_supervision_config_disabled_metrics() {
446 metrics::gauge!(SUPERVISOR_ACTIVE_METRIC).set(0.0);
447 metrics::gauge!(SUPERVISOR_DISABLED_METRIC).set(1.0);
448 metrics::gauge!(SUPERVISOR_EXHAUSTED_METRIC).set(0.0);
449}
450
451pub fn set_supervision_unexpectedly_disabled_metrics() {
453 metrics::gauge!(SUPERVISOR_ACTIVE_METRIC).set(0.0);
454 metrics::gauge!(SUPERVISOR_DISABLED_METRIC).set(1.0);
455}
456
457fn set_supervision_active_metrics() {
458 metrics::gauge!(SUPERVISOR_ACTIVE_METRIC).set(1.0);
459 metrics::gauge!(SUPERVISOR_DISABLED_METRIC).set(0.0);
460 metrics::gauge!(SUPERVISOR_EXHAUSTED_METRIC).set(0.0);
461}
462
463fn set_supervision_inactive_metrics() {
464 metrics::gauge!(SUPERVISOR_ACTIVE_METRIC).set(0.0);
465}
466
467fn should_reset_restart_count(child_uptime: Duration, restart_reset_after: Duration) -> bool {
469 restart_reset_after != Duration::ZERO && child_uptime >= restart_reset_after
470}
471
472fn restart_backoff_delay(
474 base_delay: Duration,
475 max_delay: Duration,
476 restart_count: u32,
477) -> Duration {
478 if base_delay == Duration::ZERO || restart_count <= 1 {
479 return base_delay.min(max_delay);
480 }
481
482 let multiplier = 1u32
483 .checked_shl(restart_count.saturating_sub(1))
484 .unwrap_or(u32::MAX);
485 base_delay.saturating_mul(multiplier).min(max_delay)
486}
487
488fn spawn_zcashd(config: &SupervisorConfig) -> Result<Child, Report> {
502 let args = config.command_args();
503
504 let mut command = Command::new(&config.zcashd_path);
505 command
506 .args(args)
507 .stdout(Stdio::piped())
508 .stderr(Stdio::piped())
509 .stdin(Stdio::null())
510 .kill_on_drop(false);
511 #[cfg(unix)]
512 command.process_group(0);
513
514 let mut child = command
515 .spawn()
516 .map_err(|err| eyre!("failed to spawn zcashd-compat zcashd process: {err}"))?;
517
518 if let Some(stdout) = child.stdout.take() {
519 spawn_log_task(stdout, "stdout");
520 }
521 if let Some(stderr) = child.stderr.take() {
522 spawn_log_task(stderr, "stderr");
523 }
524
525 Ok(child)
526}
527
528fn spawn_log_task<T>(stream: T, stream_name: &'static str) -> JoinHandle<()>
530where
531 T: tokio::io::AsyncRead + Unpin + Send + 'static,
532{
533 tokio::spawn(async move {
534 let mut reader = BufReader::new(stream).lines();
535
536 while let Ok(Some(line)) = reader.next_line().await {
537 let line = sanitize_child_log_line(&line);
538
539 if stream_name == "stderr" {
540 error!(target: "zcashd_compat.zcashd", stream = stream_name, "{line}");
541 } else {
542 info!(target: "zcashd_compat.zcashd", stream = stream_name, "{line}");
543 }
544 }
545 })
546}
547
548fn sanitize_child_log_line(line: &str) -> Cow<'_, str> {
550 let has_escape_or_control = line
554 .bytes()
555 .any(|byte| byte == 0x1b || byte >= 0x80 || (byte.is_ascii_control() && byte != b'\t'));
556
557 if !has_escape_or_control {
558 return Cow::Borrowed(line);
559 }
560
561 let mut output = String::with_capacity(line.len());
562 let mut chars = line.chars().peekable();
563
564 enum ParseState {
565 Normal,
566 Escape,
567 Csi,
568 Osc,
569 }
570
571 let mut state = ParseState::Normal;
572
573 while let Some(ch) = chars.next() {
574 match state {
575 ParseState::Normal => {
576 if ch == '\u{1b}' {
577 state = ParseState::Escape;
578 } else if !(ch.is_control() && ch != '\t') {
579 output.push(ch);
580 }
581 }
582 ParseState::Escape => {
583 state = match ch {
584 '[' => ParseState::Csi,
585 ']' => ParseState::Osc,
586 _ => ParseState::Normal,
587 };
588 }
589 ParseState::Csi => {
590 if ('@'..='~').contains(&ch) {
591 state = ParseState::Normal;
592 }
593 }
594 ParseState::Osc => {
595 if ch == '\u{7}' {
596 state = ParseState::Normal;
597 } else if ch == '\u{1b}' && chars.peek() == Some(&'\\') {
598 let _ = chars.next();
599 state = ParseState::Normal;
600 }
601 }
602 }
603 }
604
605 Cow::Owned(output)
606}
607
608enum ChildOutcome {
609 ShutdownRequested,
610 Exited(std::process::ExitStatus),
611 WaitFailed,
614}
615
616async fn wait_for_delay_or_shutdown(
618 delay: std::time::Duration,
619 shutdown_rx: &mut watch::Receiver<bool>,
620) -> bool {
621 if *shutdown_rx.borrow() {
622 return true;
623 }
624
625 if delay == std::time::Duration::ZERO {
626 return false;
627 }
628
629 let delay = sleep(delay);
630 tokio::pin!(delay);
631
632 loop {
633 tokio::select! {
634 () = &mut delay => return false,
635 changed = shutdown_rx.changed() => {
636 if changed.is_err() {
637 debug!("zcashd-compat shutdown sender dropped");
638 return true;
639 }
640
641 if *shutdown_rx.borrow_and_update() {
642 return true;
643 }
644 }
645 }
646 }
647}
648
649async fn wait_for_child_or_shutdown(
655 child: &mut Child,
656 shutdown_rx: &mut watch::Receiver<bool>,
657) -> ChildOutcome {
658 tokio::select! {
659 changed = shutdown_rx.changed() => {
660 if changed.is_err() {
661 debug!("zcashd-compat shutdown sender dropped");
662 }
663 ChildOutcome::ShutdownRequested
664 }
665 exited = child.wait() => {
666 match exited {
667 Ok(status) => ChildOutcome::Exited(status),
668 Err(error) => {
669 error!(?error, "failed waiting on zcashd-compat zcashd child");
670 ChildOutcome::WaitFailed
671 }
672 }
673 }
674 }
675}
676
677async fn terminate_child(
686 child: &mut Child,
687 shutdown_grace_period: std::time::Duration,
688) -> Result<(), Report> {
689 let pid = child.id();
690
691 #[cfg(unix)]
692 {
693 use nix::{
694 sys::signal::{kill, Signal::SIGTERM},
695 unistd::Pid,
696 };
697
698 if let Some(id) = pid.and_then(|id| i32::try_from(id).ok()) {
701 info!(
702 pid = id,
703 grace_period = ?shutdown_grace_period,
704 "sending SIGTERM to zcashd-compat zcashd child"
705 );
706 if let Err(error) = kill(Pid::from_raw(id), SIGTERM) {
707 warn!(
708 pid = id,
709 ?error,
710 "failed to send SIGTERM to zcashd-compat zcashd child"
711 );
712 }
713 } else {
714 warn!("zcashd-compat zcashd child has no usable process id; cannot send SIGTERM");
715 }
716 }
717
718 let start = std::time::Instant::now();
719 let wait_result = timeout(shutdown_grace_period, child.wait()).await;
720 match wait_result {
721 Ok(Ok(_status)) => {
722 info!(
723 ?pid,
724 elapsed = ?start.elapsed(),
725 "zcashd-compat zcashd exited cleanly after SIGTERM"
726 );
727 Ok(())
728 }
729 Ok(Err(error)) => Err(eyre!(
730 "failed waiting for zcashd-compat zcashd shutdown: {error}"
731 )),
732 Err(_timeout) => {
733 warn!(
734 ?pid,
735 grace_period = ?shutdown_grace_period,
736 "zcashd-compat zcashd did not exit after SIGTERM, sending kill; \
737 an interrupted shutdown can lose un-flushed chainstate"
738 );
739 child
740 .start_kill()
741 .map_err(|err| eyre!("failed to kill zcashd-compat zcashd child: {err}"))?;
742 let _ = child.wait().await;
743 Ok(())
744 }
745 }
746}
747
748pub fn is_command_resolvable(path: &Path) -> bool {
753 if path.components().count() > 1 {
754 return is_executable(path);
755 }
756
757 std::env::var_os("PATH").is_some_and(|path_var| {
758 std::env::split_paths(&path_var)
759 .map(|dir| dir.join(path))
760 .any(|candidate| candidate.exists() && is_executable(&candidate))
761 })
762}
763
764fn is_executable(path: &Path) -> bool {
769 if !path.is_file() {
770 return false;
771 }
772
773 #[cfg(unix)]
774 {
775 use std::os::unix::fs::PermissionsExt;
776 path.metadata()
777 .map(|metadata| (metadata.permissions().mode() & 0o111) != 0)
778 .unwrap_or(false)
779 }
780
781 #[cfg(not(unix))]
782 {
783 use std::ffi::OsStr;
784
785 let extension = path.extension().and_then(OsStr::to_str).unwrap_or_default();
786 return matches!(
787 extension.to_ascii_lowercase().as_str(),
788 "exe" | "cmd" | "bat" | "com"
789 );
790 }
791}
792
793#[cfg(test)]
794mod tests {
795 use std::{path::PathBuf, time::Duration};
796
797 use tokio::sync::watch;
798 use zebra_chain::parameters::NetworkKind;
799
800 use super::{
801 reject_peer_selection_extra_args, restart_backoff_delay, should_reset_restart_count,
802 wait_for_delay_or_shutdown, SupervisorConfig,
803 };
804
805 fn test_supervisor_config(extra_args: Vec<String>) -> SupervisorConfig {
806 SupervisorConfig {
807 zcashd_path: PathBuf::from("zcashd"),
808 zcashd_datadir: PathBuf::from("/tmp/zcashd-compat-datadir"),
809 zebra_p2p_addr: "127.0.0.1:18233".parse().expect("valid socket address"),
810 extra_args,
811 network: NetworkKind::Regtest,
812 startup_delay: Duration::from_secs(1),
813 restart_backoff: Duration::from_secs(2),
814 restart_backoff_max: Duration::from_secs(5 * 60),
815 restart_reset_after: Duration::from_secs(60 * 60),
816 shutdown_grace_period: Duration::from_secs(300),
817 }
818 }
819
820 #[test]
821 fn command_args_pin_zcashd_to_zebra_p2p() {
822 let config = test_supervisor_config(vec!["-debug=1".to_string()]);
823
824 let args = config.command_args();
825
826 assert!(args.contains(&"-datadir=/tmp/zcashd-compat-datadir".to_string()));
827 assert!(args.contains(&"-regtest".to_string()));
828 assert!(args.contains(&"-regtestacceptunvalidatedpow".to_string()));
829 assert!(args.contains(&"-connect=127.0.0.1:18233".to_string()));
830 assert!(args.contains(&"-listen=0".to_string()));
831 assert!(args.contains(&"-dnsseed=0".to_string()));
832 assert!(args.contains(&"-listenonion=0".to_string()));
833 assert!(args.contains(&"-discover=0".to_string()));
834 assert!(args.contains(&"-daemon=0".to_string()));
835 assert!(args.contains(&"-printtoconsole".to_string()));
836 assert!(args.contains(&"-debug=1".to_string()));
837 assert!(
838 !args.iter().any(|arg| arg.starts_with("-zebra-compat")),
839 "P2P sidecar must not pass RPC-ingest flags: {args:?}"
840 );
841
842 let debug_idx = args
845 .iter()
846 .position(|a| a == "-debug=1")
847 .expect("extra arg present");
848 for forced in [
849 "-listen=0",
850 "-dnsseed=0",
851 "-listenonion=0",
852 "-discover=0",
853 "-daemon=0",
854 ] {
855 let forced_idx = args
856 .iter()
857 .position(|a| a == forced)
858 .expect("forced flag present");
859 assert!(
860 forced_idx > debug_idx,
861 "{forced} must come after extra_args"
862 );
863 }
864 }
865
866 #[test]
867 fn peer_selection_extra_args_are_rejected() {
868 for arg in [
869 "-connect=1.2.3.4:8233",
870 "--connect=1.2.3.4",
871 "-addnode=1.2.3.4",
872 "-seednode=1.2.3.4",
873 "-noconnect",
874 "-CONNECT=1.2.3.4",
875 ] {
876 let _rejected = reject_peer_selection_extra_args(&[arg.to_string()])
877 .expect_err("peer-selection extra args must be rejected");
878 }
879
880 reject_peer_selection_extra_args(&[
881 "-debug=1".to_string(),
882 "-rpcport=18232".to_string(),
883 "-maxconnections=8".to_string(),
884 ])
885 .expect("non-peer-selection extra args are allowed");
886 }
887
888 #[test]
889 fn restart_count_resets_after_healthy_uptime() {
890 assert!(should_reset_restart_count(
891 Duration::from_secs(60 * 60),
892 Duration::from_secs(60 * 60)
893 ));
894 assert!(should_reset_restart_count(
895 Duration::from_secs(60 * 60 + 1),
896 Duration::from_secs(60 * 60)
897 ));
898 }
899
900 #[test]
901 fn restart_count_does_not_reset_before_threshold() {
902 assert!(!should_reset_restart_count(
903 Duration::from_secs(60 * 60 - 1),
904 Duration::from_secs(60 * 60)
905 ));
906 assert!(!should_reset_restart_count(
907 Duration::from_secs(60 * 60),
908 Duration::ZERO
909 ));
910 }
911
912 #[test]
913 fn restart_backoff_is_exponential_from_base_delay() {
914 let base_delay = Duration::from_secs(2);
915 let max_delay = Duration::from_secs(60);
916
917 assert_eq!(restart_backoff_delay(base_delay, max_delay, 0), base_delay);
918 assert_eq!(restart_backoff_delay(base_delay, max_delay, 1), base_delay);
919 assert_eq!(
920 restart_backoff_delay(base_delay, max_delay, 2),
921 Duration::from_secs(4)
922 );
923 assert_eq!(
924 restart_backoff_delay(base_delay, max_delay, 3),
925 Duration::from_secs(8)
926 );
927 }
928
929 #[test]
930 fn restart_backoff_is_capped() {
931 let delay = restart_backoff_delay(Duration::from_secs(2), Duration::from_secs(10), 10);
932
933 assert_eq!(delay, Duration::from_secs(10));
934 }
935
936 #[test]
937 fn restart_backoff_caps_saturated_delay() {
938 let delay = restart_backoff_delay(Duration::MAX, Duration::from_secs(10), u32::MAX);
939
940 assert_eq!(delay, Duration::from_secs(10));
941 }
942
943 #[tokio::test]
944 async fn delay_wait_returns_on_shutdown_request() {
945 let (shutdown_tx, mut shutdown_rx) = watch::channel(false);
946
947 let wait = tokio::spawn(async move {
948 wait_for_delay_or_shutdown(Duration::from_secs(60), &mut shutdown_rx).await
949 });
950
951 shutdown_tx
952 .send(true)
953 .expect("shutdown receiver exists because wait task owns it");
954
955 let was_shutdown = tokio::time::timeout(Duration::from_secs(1), wait)
956 .await
957 .expect("interruptible delay should complete promptly")
958 .expect("wait task should not panic");
959
960 assert!(was_shutdown);
961 }
962
963 #[tokio::test]
964 async fn delay_wait_returns_on_dropped_shutdown_sender() {
965 let (shutdown_tx, mut shutdown_rx) = watch::channel(false);
966
967 let wait = tokio::spawn(async move {
968 wait_for_delay_or_shutdown(Duration::from_secs(60), &mut shutdown_rx).await
969 });
970
971 drop(shutdown_tx);
972
973 let was_shutdown = tokio::time::timeout(Duration::from_secs(1), wait)
974 .await
975 .expect("interruptible delay should complete promptly")
976 .expect("wait task should not panic");
977
978 assert!(was_shutdown);
979 }
980
981 #[test]
982 fn sanitize_child_log_line_strips_ansi_csi_sequences() {
983 let line = "\u{1b}[32mINFO\u{1b}[0m ProcessNewTrustedBlockBatch";
984 let sanitized = super::sanitize_child_log_line(line);
985
986 assert_eq!(sanitized, "INFO ProcessNewTrustedBlockBatch");
987 }
988
989 #[test]
990 fn sanitize_child_log_line_removes_control_chars() {
991 let line = "good\u{0}text\u{8}\tkeeps-tab";
992 let sanitized = super::sanitize_child_log_line(line);
993
994 assert_eq!(sanitized, "goodtext\tkeeps-tab");
995 }
996
997 #[test]
998 fn sanitize_child_log_line_keeps_clean_lines_unchanged() {
999 let line = "UpdateTip: new best hash=abc height=42";
1000 let sanitized = super::sanitize_child_log_line(line);
1001
1002 assert_eq!(sanitized, line);
1003 }
1004
1005 #[cfg(unix)]
1008 #[tokio::test]
1009 async fn terminate_child_waits_for_graceful_exit() {
1010 let mut child = tokio::process::Command::new("/bin/sleep")
1011 .arg("60")
1012 .kill_on_drop(false)
1013 .spawn()
1014 .expect("sleep is available on unix test hosts");
1015
1016 let start = std::time::Instant::now();
1017 super::terminate_child(&mut child, Duration::from_secs(30))
1018 .await
1019 .expect("terminate_child should succeed for a SIGTERM-compliant child");
1020
1021 assert!(
1022 start.elapsed() < Duration::from_secs(30),
1023 "child should exit on SIGTERM well before the grace period"
1024 );
1025 }
1026
1027 #[cfg(unix)]
1030 #[tokio::test(start_paused = true)]
1031 async fn terminate_child_kills_after_grace_period() {
1032 use std::process::Stdio;
1033
1034 let mut child = tokio::process::Command::new("/bin/sh")
1035 .args(["-c", "trap '' TERM; while read _; do :; done"])
1036 .stdin(Stdio::piped())
1037 .kill_on_drop(true)
1038 .spawn()
1039 .expect("sh is available on unix test hosts");
1040
1041 tokio::task::yield_now().await;
1044 std::thread::sleep(std::time::Duration::from_millis(200));
1045
1046 super::terminate_child(&mut child, Duration::from_secs(5))
1047 .await
1048 .expect("terminate_child should fall back to SIGKILL");
1049
1050 let status = child
1051 .try_wait()
1052 .expect("child status should be queryable after terminate_child");
1053 assert!(
1054 status.is_some(),
1055 "child must have been reaped after the SIGKILL fallback"
1056 );
1057 }
1058}