zebrad 6.2.2

use std::{
    borrow::Cow,
    net::SocketAddr,
    path::{Path, PathBuf},
    process::Stdio,
    sync::atomic::{AtomicU32, Ordering},
    time::{Duration, Instant},
};

use color_eyre::eyre::{eyre, Report};
use tokio::{
    io::{AsyncBufReadExt, BufReader},
    process::{Child, Command},
    sync::watch,
    task::JoinHandle,
    time::{sleep, timeout},
};
use tracing::{debug, error, info, warn};

use zebra_chain::parameters::NetworkKind;

use super::{effective_zcashd_datadir, ensure_zcashd_datadir, resolve_zcashd_datadir_path, Config};

const SUPERVISOR_ACTIVE_METRIC: &str = "zcashd.compat.supervisor.active";
const SUPERVISOR_DISABLED_METRIC: &str = "zcashd.compat.supervisor.disabled";
const SUPERVISOR_EXHAUSTED_METRIC: &str = "zcashd.compat.supervisor.exhausted";

/// The pid of the currently running supervised zcashd child, or `0` when none.
///
/// On SIGINT/SIGTERM, the tokio runtime drops the whole `start()` future — and
/// the supervisor task with it — without ever running the supervisor's
/// graceful-shutdown path. The child is spawned without `kill_on_drop`, so it
/// would be silently orphaned. This pid lets
/// [`terminate_abandoned_zcashd`] clean up synchronously after the runtime has
/// shut down.
static SUPERVISED_ZCASHD_PID: AtomicU32 = AtomicU32::new(0);

/// The full configuration used by the zcashd-compat supervisor task.
#[derive(Clone, Debug)]
pub struct SupervisorConfig {
    /// Path to the `zcashd` binary.
    pub zcashd_path: PathBuf,
    /// Datadir for `zcashd`.
    pub zcashd_datadir: PathBuf,
    /// Zebra's legacy P2P listen address, passed to zcashd as `-connect` so the
    /// sidecar peers only with the local Zebra node.
    pub zebra_p2p_addr: SocketAddr,
    /// Any extra user-provided arguments.
    pub extra_args: Vec<String>,
    /// Active Zebra network kind.
    pub network: NetworkKind,
    /// Delay before first spawn.
    pub startup_delay: std::time::Duration,
    /// Restart backoff.
    pub restart_backoff: Duration,
    /// Maximum restart backoff.
    pub restart_backoff_max: Duration,
    /// Child uptime that resets the consecutive restart count.
    pub restart_reset_after: Duration,
    /// Grace period after SIGTERM.
    pub shutdown_grace_period: Duration,
}

impl SupervisorConfig {
    /// Builds a runtime supervisor config from `zebrad` and `[zcashd_compat]` settings.
    pub fn new(
        zcashd_compat: &Config,
        zcashd_path: PathBuf,
        state_cache_dir: &Path,
        network: NetworkKind,
        zebra_p2p_addr: SocketAddr,
    ) -> Self {
        let extra_args = zcashd_compat.zcashd_extra_args.clone();
        let zcashd_datadir = resolve_zcashd_datadir_path(
            &effective_zcashd_datadir(zcashd_compat, state_cache_dir),
            &extra_args,
        );

        Self {
            zcashd_path,
            zcashd_datadir,
            zebra_p2p_addr,
            extra_args,
            network,
            startup_delay: zcashd_compat.startup_delay,
            restart_backoff: zcashd_compat.restart_backoff,
            restart_backoff_max: zcashd_compat.restart_backoff_max,
            restart_reset_after: zcashd_compat.restart_reset_after,
            shutdown_grace_period: zcashd_compat.shutdown_grace_period,
        }
    }

    /// Builds the zcashd command-line arguments.
    pub fn command_args(&self) -> Vec<String> {
        let mut args = vec![format!(
            "-datadir={}",
            self.zcashd_datadir.to_string_lossy()
        )];

        match self.network {
            NetworkKind::Mainnet => {}
            NetworkKind::Testnet => args.push("-testnet".to_string()),
            NetworkKind::Regtest => {
                args.push("-regtest".to_string());
                // Zebra skips proof-of-work on regtest, so its mined blocks
                // carry null Equihash solutions that stock zcashd validation
                // would reject with a peer ban.
                args.push("-regtestacceptunvalidatedpow".to_string());
            }
        }

        // Always include -printtoconsole and filter it out from extra_args
        args.push("-printtoconsole".to_string());
        args.extend(
            self.extra_args
                .iter()
                .filter(|arg| arg.as_str() != "-printtoconsole")
                .cloned(),
        );

        // zcashd peers only with the local Zebra node: `-connect` pins the
        // single outbound peer, and zcashd itself then soft-disables DNS
        // seeding, inbound listening, and discovery. `-daemon=0` keeps zcashd
        // in the foreground so the supervisor's tracked child *is* the real
        // process: a `daemon=1` left in the operator's zcash.conf (or passed in
        // extra_args) would otherwise fork, let the tracked parent exit
        // "successfully", and leave the supervisor unable to signal or reap the
        // real daemon while it respawns fresh parents. The explicit flags are
        // defense in depth against operator zcash.conf values. They come after
        // extra_args because zcashd takes the *last* occurrence of a
        // single-valued command-line argument, and a command-line value
        // overrides zcash.conf. Multi-valued peer-selection options
        // (-connect/-addnode/-seednode) accumulate instead, so
        // [`reject_peer_selection_extra_args`] refuses them at startup.
        args.push(format!("-connect={}", self.zebra_p2p_addr));
        args.push("-listen=0".to_string());
        args.push("-dnsseed=0".to_string());
        args.push("-listenonion=0".to_string());
        args.push("-discover=0".to_string());
        args.push("-daemon=0".to_string());

        args
    }
}

/// zcashd options that add P2P peers and accumulate across the command line,
/// so the supervisor's own `-connect` cannot override them.
const PEER_SELECTION_OPTIONS: &[&str] = &["connect", "addnode", "seednode"];

/// Rejects `zcashd_extra_args` entries that would change which peers the
/// supervised zcashd talks to.
///
/// The P2P sidecar must connect only to the local Zebra node. Unlike
/// single-valued boolean flags, every `-connect`/`-addnode`/`-seednode`
/// occurrence adds a peer, and negated forms (`-noconnect`) clobber the
/// supervisor's pinned `-connect`, so both are refused instead of overridden.
///
/// # Errors
///
/// Returns an error naming the first offending argument.
pub fn reject_peer_selection_extra_args(extra_args: &[String]) -> Result<(), Report> {
    for arg in extra_args {
        let name = arg
            .trim_start_matches('-')
            .split('=')
            .next()
            .unwrap_or_default()
            .to_ascii_lowercase();
        let name = name.strip_prefix("no").unwrap_or(&name);

        if PEER_SELECTION_OPTIONS.contains(&name) {
            return Err(eyre!(
                "zcashd_compat.zcashd_extra_args contains {arg:?}: peer-selection options are not \
                 allowed because the zcashd P2P sidecar must connect only to the local Zebra node"
            ));
        }
    }

    Ok(())
}

/// Runs the zcashd-compat zcashd supervisor until shutdown.
///
/// The supervisor keeps restarting `zcashd` exits that happen before Zebra
/// shutdown, using capped exponential backoff. Spawn failures use the same
/// backoff, so a binary that is briefly missing or unspawnable (for example
/// during an upgrade, or under transient resource pressure) does not
/// permanently end supervision.
///
/// # Errors
///
/// Returns an error if shutdown handling fails.
pub async fn run(
    config: SupervisorConfig,
    mut shutdown_rx: watch::Receiver<bool>,
) -> Result<(), Report> {
    reject_peer_selection_extra_args(&config.extra_args)?;
    // A zero base or maximum backoff collapses `restart_backoff_delay` to zero,
    // so a zcashd that fails to spawn or crashes immediately would be respawned
    // in a tight loop that burns CPU and floods the logs. Require both to be
    // positive so the capped exponential backoff stays meaningful.
    if config.restart_backoff.is_zero() || config.restart_backoff_max.is_zero() {
        return Err(eyre!(
            "zcashd_compat.restart_backoff and zcashd_compat.restart_backoff_max must both be \
             greater than zero to avoid a hot zcashd restart loop"
        ));
    }
    ensure_zcashd_datadir(&config.zcashd_datadir, &config.extra_args)?;
    set_supervision_active_metrics();

    if wait_for_delay_or_shutdown(config.startup_delay, &mut shutdown_rx).await {
        info!("zcashd-compat supervisor received shutdown during startup delay");
        set_supervision_inactive_metrics();
        return Ok(());
    }

    let mut consecutive_restart_count = 0u32;

    loop {
        if *shutdown_rx.borrow() {
            info!("zcashd-compat supervisor received shutdown before spawn");
            set_supervision_inactive_metrics();
            return Ok(());
        }

        let mut child = match spawn_zcashd(&config) {
            Ok(child) => child,
            Err(error) => {
                consecutive_restart_count = consecutive_restart_count.saturating_add(1);
                warn!(
                    %error,
                    restart_count = consecutive_restart_count,
                    "failed to spawn zcashd-compat zcashd child, retrying after backoff"
                );

                let restart_delay = restart_backoff_delay(
                    config.restart_backoff,
                    config.restart_backoff_max,
                    consecutive_restart_count,
                );
                if wait_for_delay_or_shutdown(restart_delay, &mut shutdown_rx).await {
                    info!("zcashd-compat supervisor received shutdown during spawn retry backoff");
                    set_supervision_inactive_metrics();
                    return Ok(());
                }
                continue;
            }
        };
        let child_started_at = Instant::now();
        SUPERVISED_ZCASHD_PID.store(child.id().unwrap_or(0), Ordering::SeqCst);
        info!(
            path = %config.zcashd_path.display(),
            datadir = %config.zcashd_datadir.display(),
            connect = %config.zebra_p2p_addr,
            "started zcashd-compat zcashd child"
        );

        let child_result = wait_for_child_or_shutdown(&mut child, &mut shutdown_rx).await;
        match child_result {
            ChildOutcome::ShutdownRequested => {
                info!(
                    pid = ?child.id(),
                    grace_period = ?config.shutdown_grace_period,
                    "zcashd-compat supervisor received shutdown request; terminating zcashd child"
                );
                let terminate_result =
                    terminate_child(&mut child, config.shutdown_grace_period).await;
                if terminate_result.is_ok() {
                    // Only forget the pid once the child's exit is confirmed:
                    // on errors it may still be running, and the pid arms the
                    // post-runtime `terminate_abandoned_zcashd` cleanup.
                    SUPERVISED_ZCASHD_PID.store(0, Ordering::SeqCst);
                }
                terminate_result?;
                info!("zcashd-compat zcashd child stopped on shutdown");
                set_supervision_inactive_metrics();
                return Ok(());
            }
            ChildOutcome::Exited(status) => {
                SUPERVISED_ZCASHD_PID.store(0, Ordering::SeqCst);
                let child_uptime = child_started_at.elapsed();
                if should_reset_restart_count(child_uptime, config.restart_reset_after) {
                    info!(
                        ?status,
                        child_uptime_secs = child_uptime.as_secs(),
                        restart_reset_after_secs = config.restart_reset_after.as_secs(),
                        previous_restart_count = consecutive_restart_count,
                        "zcashd-compat zcashd child had healthy uptime, resetting restart count"
                    );
                    consecutive_restart_count = 0;
                }

                consecutive_restart_count = consecutive_restart_count.saturating_add(1);
                warn!(
                    ?status,
                    restart_count = consecutive_restart_count,
                    child_uptime_secs = child_uptime.as_secs(),
                    "zcashd-compat zcashd child exited before shutdown, restarting"
                );

                let restart_delay = restart_backoff_delay(
                    config.restart_backoff,
                    config.restart_backoff_max,
                    consecutive_restart_count,
                );
                if wait_for_delay_or_shutdown(restart_delay, &mut shutdown_rx).await {
                    info!("zcashd-compat supervisor received shutdown during restart backoff");
                    set_supervision_inactive_metrics();
                    return Ok(());
                }
            }
            ChildOutcome::WaitFailed => {
                // A `child.wait()` error does not prove the process exited, so
                // reap it before doing anything else: spawning a replacement
                // while the original may still be alive would run two zcashd
                // against one datadir and corrupt wallet.dat.
                warn!(
                    "failed waiting on zcashd-compat zcashd child; \
                     terminating it before restart"
                );
                if let Err(error) = terminate_child(&mut child, config.shutdown_grace_period).await
                {
                    // The child's state is unknown and reaping was not
                    // confirmed, so refuse to start a second instance. The
                    // tracked pid stays armed for the post-runtime
                    // `terminate_abandoned_zcashd` cleanup that runs once
                    // zebrad exits.
                    set_supervision_inactive_metrics();
                    return Err(eyre!(
                        "could not reap zcashd-compat zcashd after a wait error ({error}); \
                         stopping supervision to avoid running two instances on one datadir"
                    ));
                }
                SUPERVISED_ZCASHD_PID.store(0, Ordering::SeqCst);

                consecutive_restart_count = consecutive_restart_count.saturating_add(1);
                let restart_delay = restart_backoff_delay(
                    config.restart_backoff,
                    config.restart_backoff_max,
                    consecutive_restart_count,
                );
                if wait_for_delay_or_shutdown(restart_delay, &mut shutdown_rx).await {
                    info!("zcashd-compat supervisor received shutdown during restart backoff");
                    set_supervision_inactive_metrics();
                    return Ok(());
                }
            }
        }
    }
}

/// Terminates a supervised zcashd child that the supervisor task never got to
/// shut down, blocking the calling thread.
///
/// On SIGINT/SIGTERM, the tokio runtime cancels the supervisor task without
/// polling it again, so its graceful-shutdown path never runs and the child
/// (spawned without `kill_on_drop`) is orphaned. Call this after the runtime
/// has shut down: it sends SIGTERM, waits up to `shutdown_grace_period` for the
/// child to exit, and SIGKILLs it as a last resort — the same sequence as the
/// supervisor's own [`terminate_child`].
///
/// Does nothing when no supervised child is running, or on non-Unix targets
/// (managed zcashd is only supported on Linux).
pub fn terminate_abandoned_zcashd(shutdown_grace_period: Duration) {
    let pid = SUPERVISED_ZCASHD_PID.swap(0, Ordering::SeqCst);
    if pid == 0 {
        return;
    }

    #[cfg(unix)]
    {
        use nix::{
            sys::{
                signal::{kill, Signal::SIGKILL, Signal::SIGTERM},
                wait::{waitpid, WaitPidFlag, WaitStatus},
            },
            unistd::Pid,
        };

        let Ok(pid) = i32::try_from(pid) else {
            // Linux pids fit in i32; a value that doesn't cannot be signalled
            // safely (a wrapped negative pid would target a process group).
            warn!(
                pid,
                "abandoned zcashd-compat zcashd pid does not fit in i32"
            );
            return;
        };
        let pid = Pid::from_raw(pid);

        // Returns true once the child has exited. zebrad is the child's
        // parent, and its dropped `Child` handle no longer reaps it, so an
        // exited child stays a zombie (where `kill(pid, 0)` still succeeds)
        // until this `waitpid` reaps it. `waitpid` only matches our own
        // children, so a recycled pid belonging to another process reports
        // as exited instead of being signalled.
        let child_has_exited = || match waitpid(pid, Some(WaitPidFlag::WNOHANG)) {
            Ok(WaitStatus::StillAlive) => false,
            // Reaped here, already reaped, or never ours: nothing left running.
            Ok(_) | Err(_) => true,
        };

        // Check before signalling: if the child already exited and was reaped,
        // its pid may have been recycled by an unrelated process.
        if child_has_exited() {
            return;
        }

        info!(
            %pid,
            grace_period = ?shutdown_grace_period,
            "terminating zcashd-compat zcashd child abandoned by runtime shutdown"
        );
        if kill(pid, SIGTERM).is_err() || child_has_exited() {
            return;
        }

        const POLL_INTERVAL: Duration = Duration::from_millis(200);
        let deadline = Instant::now() + shutdown_grace_period;
        while Instant::now() < deadline {
            std::thread::sleep(POLL_INTERVAL);
            if child_has_exited() {
                info!(%pid, "abandoned zcashd-compat zcashd child exited after SIGTERM");
                return;
            }
        }

        warn!(
            %pid,
            grace_period = ?shutdown_grace_period,
            "abandoned zcashd-compat zcashd child did not exit within the grace period; \
             sending SIGKILL as a last resort"
        );
        let _ = kill(pid, SIGKILL);
        let _ = waitpid(pid, None);
    }

    #[cfg(not(unix))]
    {
        let _ = (pid, shutdown_grace_period);
    }
}

/// Sets metrics for zcashd-compat mode when zcashd supervision is intentionally disabled.
pub fn set_supervision_config_disabled_metrics() {
    metrics::gauge!(SUPERVISOR_ACTIVE_METRIC).set(0.0);
    metrics::gauge!(SUPERVISOR_DISABLED_METRIC).set(1.0);
    metrics::gauge!(SUPERVISOR_EXHAUSTED_METRIC).set(0.0);
}

/// Sets metrics for zcashd-compat mode when supervision has unexpectedly stopped.
pub fn set_supervision_unexpectedly_disabled_metrics() {
    metrics::gauge!(SUPERVISOR_ACTIVE_METRIC).set(0.0);
    metrics::gauge!(SUPERVISOR_DISABLED_METRIC).set(1.0);
}

fn set_supervision_active_metrics() {
    metrics::gauge!(SUPERVISOR_ACTIVE_METRIC).set(1.0);
    metrics::gauge!(SUPERVISOR_DISABLED_METRIC).set(0.0);
    metrics::gauge!(SUPERVISOR_EXHAUSTED_METRIC).set(0.0);
}

fn set_supervision_inactive_metrics() {
    metrics::gauge!(SUPERVISOR_ACTIVE_METRIC).set(0.0);
}

/// Returns `true` when a child ran long enough to make previous failures stale.
fn should_reset_restart_count(child_uptime: Duration, restart_reset_after: Duration) -> bool {
    restart_reset_after != Duration::ZERO && child_uptime >= restart_reset_after
}

/// Calculates capped exponential restart backoff from the base delay and consecutive exit count.
fn restart_backoff_delay(
    base_delay: Duration,
    max_delay: Duration,
    restart_count: u32,
) -> Duration {
    if base_delay == Duration::ZERO || restart_count <= 1 {
        return base_delay.min(max_delay);
    }

    let multiplier = 1u32
        .checked_shl(restart_count.saturating_sub(1))
        .unwrap_or(u32::MAX);
    base_delay.saturating_mul(multiplier).min(max_delay)
}

/// Spawns `zcashd` with zcashd-compat arguments and connects child output streams.
///
/// `kill_on_drop` is intentionally disabled: a dropped child handle (zebrad
/// panic, supervisor task abort) must not SIGKILL a zcashd that may be flushing
/// its chainstate and wallet. An abandoned zcashd finishes any SIGTERM-initiated
/// shutdown on its own, or keeps running until stopped externally; `init` reaps
/// it once zebrad exits. The child also runs in its own process group so
/// group-wide terminal signals aimed at zebrad cannot kill zcashd uncleanly;
/// [`terminate_child`] remains the only path that force-kills it.
///
/// # Errors
///
/// Returns an error if the child process cannot be spawned.
fn spawn_zcashd(config: &SupervisorConfig) -> Result<Child, Report> {
    let args = config.command_args();

    let mut command = Command::new(&config.zcashd_path);
    command
        .args(args)
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .stdin(Stdio::null())
        .kill_on_drop(false);
    #[cfg(unix)]
    command.process_group(0);

    let mut child = command
        .spawn()
        .map_err(|err| eyre!("failed to spawn zcashd-compat zcashd process: {err}"))?;

    if let Some(stdout) = child.stdout.take() {
        spawn_log_task(stdout, "stdout");
    }
    if let Some(stderr) = child.stderr.take() {
        spawn_log_task(stderr, "stderr");
    }

    Ok(child)
}

/// Forwards a child output stream into Zebra logs under `zcashd_compat.zcashd`.
fn spawn_log_task<T>(stream: T, stream_name: &'static str) -> JoinHandle<()>
where
    T: tokio::io::AsyncRead + Unpin + Send + 'static,
{
    tokio::spawn(async move {
        let mut reader = BufReader::new(stream).lines();

        while let Ok(Some(line)) = reader.next_line().await {
            let line = sanitize_child_log_line(&line);

            if stream_name == "stderr" {
                error!(target: "zcashd_compat.zcashd", stream = stream_name, "{line}");
            } else {
                info!(target: "zcashd_compat.zcashd", stream = stream_name, "{line}");
            }
        }
    })
}

/// Returns a sanitized log line with ANSI escape/control noise removed.
fn sanitize_child_log_line(line: &str) -> Cow<'_, str> {
    // Take the slow path for any non-ASCII byte too: C1 controls like U+009B
    // (single-byte CSI) are multi-byte in UTF-8 and would slip through an
    // ASCII-only gate, but are stripped by the `is_control` filter below.
    let has_escape_or_control = line
        .bytes()
        .any(|byte| byte == 0x1b || byte >= 0x80 || (byte.is_ascii_control() && byte != b'\t'));

    if !has_escape_or_control {
        return Cow::Borrowed(line);
    }

    let mut output = String::with_capacity(line.len());
    let mut chars = line.chars().peekable();

    enum ParseState {
        Normal,
        Escape,
        Csi,
        Osc,
    }

    let mut state = ParseState::Normal;

    while let Some(ch) = chars.next() {
        match state {
            ParseState::Normal => {
                if ch == '\u{1b}' {
                    state = ParseState::Escape;
                } else if !(ch.is_control() && ch != '\t') {
                    output.push(ch);
                }
            }
            ParseState::Escape => {
                state = match ch {
                    '[' => ParseState::Csi,
                    ']' => ParseState::Osc,
                    _ => ParseState::Normal,
                };
            }
            ParseState::Csi => {
                if ('@'..='~').contains(&ch) {
                    state = ParseState::Normal;
                }
            }
            ParseState::Osc => {
                if ch == '\u{7}' {
                    state = ParseState::Normal;
                } else if ch == '\u{1b}' && chars.peek() == Some(&'\\') {
                    let _ = chars.next();
                    state = ParseState::Normal;
                }
            }
        }
    }

    Cow::Owned(output)
}

enum ChildOutcome {
    ShutdownRequested,
    Exited(std::process::ExitStatus),
    /// Waiting on the child returned an error, so it is unknown whether the
    /// process actually exited. The supervisor must reap it before restarting.
    WaitFailed,
}

/// Waits for `delay` to elapse, returning `true` if shutdown is requested first.
async fn wait_for_delay_or_shutdown(
    delay: std::time::Duration,
    shutdown_rx: &mut watch::Receiver<bool>,
) -> bool {
    if *shutdown_rx.borrow() {
        return true;
    }

    if delay == std::time::Duration::ZERO {
        return false;
    }

    let delay = sleep(delay);
    tokio::pin!(delay);

    loop {
        tokio::select! {
            () = &mut delay => return false,
            changed = shutdown_rx.changed() => {
                if changed.is_err() {
                    debug!("zcashd-compat shutdown sender dropped");
                    return true;
                }

                if *shutdown_rx.borrow_and_update() {
                    return true;
                }
            }
        }
    }
}

/// Waits until either a shutdown request arrives or the child exits.
///
/// If waiting on the child fails, returns [`ChildOutcome::WaitFailed`] so the
/// supervisor reaps the possibly-still-running child before restarting, rather
/// than assuming it exited.
async fn wait_for_child_or_shutdown(
    child: &mut Child,
    shutdown_rx: &mut watch::Receiver<bool>,
) -> ChildOutcome {
    tokio::select! {
        changed = shutdown_rx.changed() => {
            if changed.is_err() {
                debug!("zcashd-compat shutdown sender dropped");
            }
            ChildOutcome::ShutdownRequested
        }
        exited = child.wait() => {
            match exited {
                Ok(status) => ChildOutcome::Exited(status),
                Err(error) => {
                    error!(?error, "failed waiting on zcashd-compat zcashd child");
                    ChildOutcome::WaitFailed
                }
            }
        }
    }
}

/// Attempts graceful termination of the zcashd-compat child process.
///
/// On Unix, this sends SIGTERM first. If the process has not exited after
/// `shutdown_grace_period`, it is force-killed.
///
/// # Errors
///
/// Returns an error if waiting for process termination fails.
async fn terminate_child(
    child: &mut Child,
    shutdown_grace_period: std::time::Duration,
) -> Result<(), Report> {
    let pid = child.id();

    #[cfg(unix)]
    {
        use nix::{
            sys::signal::{kill, Signal::SIGTERM},
            unistd::Pid,
        };

        // Linux pids fit in i32; a value that doesn't cannot be signalled
        // safely (a wrapped negative pid would target a process group).
        if let Some(id) = pid.and_then(|id| i32::try_from(id).ok()) {
            info!(
                pid = id,
                grace_period = ?shutdown_grace_period,
                "sending SIGTERM to zcashd-compat zcashd child"
            );
            if let Err(error) = kill(Pid::from_raw(id), SIGTERM) {
                warn!(
                    pid = id,
                    ?error,
                    "failed to send SIGTERM to zcashd-compat zcashd child"
                );
            }
        } else {
            warn!("zcashd-compat zcashd child has no usable process id; cannot send SIGTERM");
        }
    }

    let start = std::time::Instant::now();
    let wait_result = timeout(shutdown_grace_period, child.wait()).await;
    match wait_result {
        Ok(Ok(_status)) => {
            info!(
                ?pid,
                elapsed = ?start.elapsed(),
                "zcashd-compat zcashd exited cleanly after SIGTERM"
            );
            Ok(())
        }
        Ok(Err(error)) => Err(eyre!(
            "failed waiting for zcashd-compat zcashd shutdown: {error}"
        )),
        Err(_timeout) => {
            warn!(
                ?pid,
                grace_period = ?shutdown_grace_period,
                "zcashd-compat zcashd did not exit after SIGTERM, sending kill; \
                 an interrupted shutdown can lose un-flushed chainstate"
            );
            child
                .start_kill()
                .map_err(|err| eyre!("failed to kill zcashd-compat zcashd child: {err}"))?;
            let _ = child.wait().await;
            Ok(())
        }
    }
}

/// Returns `true` if the given command path is resolvable as an executable.
///
/// Paths containing separators are validated directly, while bare command names
/// are searched in `PATH`.
pub fn is_command_resolvable(path: &Path) -> bool {
    if path.components().count() > 1 {
        return is_executable(path);
    }

    std::env::var_os("PATH").is_some_and(|path_var| {
        std::env::split_paths(&path_var)
            .map(|dir| dir.join(path))
            .any(|candidate| candidate.exists() && is_executable(&candidate))
    })
}

/// Returns `true` when `path` points to an executable regular file.
///
/// On Unix this checks execute mode bits. On non-Unix targets this checks
/// common executable filename extensions.
fn is_executable(path: &Path) -> bool {
    if !path.is_file() {
        return false;
    }

    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        path.metadata()
            .map(|metadata| (metadata.permissions().mode() & 0o111) != 0)
            .unwrap_or(false)
    }

    #[cfg(not(unix))]
    {
        use std::ffi::OsStr;

        let extension = path.extension().and_then(OsStr::to_str).unwrap_or_default();
        return matches!(
            extension.to_ascii_lowercase().as_str(),
            "exe" | "cmd" | "bat" | "com"
        );
    }
}

#[cfg(test)]
mod tests {
    use std::{path::PathBuf, time::Duration};

    use tokio::sync::watch;
    use zebra_chain::parameters::NetworkKind;

    use super::{
        reject_peer_selection_extra_args, restart_backoff_delay, should_reset_restart_count,
        wait_for_delay_or_shutdown, SupervisorConfig,
    };

    fn test_supervisor_config(extra_args: Vec<String>) -> SupervisorConfig {
        SupervisorConfig {
            zcashd_path: PathBuf::from("zcashd"),
            zcashd_datadir: PathBuf::from("/tmp/zcashd-compat-datadir"),
            zebra_p2p_addr: "127.0.0.1:18233".parse().expect("valid socket address"),
            extra_args,
            network: NetworkKind::Regtest,
            startup_delay: Duration::from_secs(1),
            restart_backoff: Duration::from_secs(2),
            restart_backoff_max: Duration::from_secs(5 * 60),
            restart_reset_after: Duration::from_secs(60 * 60),
            shutdown_grace_period: Duration::from_secs(300),
        }
    }

    #[test]
    fn command_args_pin_zcashd_to_zebra_p2p() {
        let config = test_supervisor_config(vec!["-debug=1".to_string()]);

        let args = config.command_args();

        assert!(args.contains(&"-datadir=/tmp/zcashd-compat-datadir".to_string()));
        assert!(args.contains(&"-regtest".to_string()));
        assert!(args.contains(&"-regtestacceptunvalidatedpow".to_string()));
        assert!(args.contains(&"-connect=127.0.0.1:18233".to_string()));
        assert!(args.contains(&"-listen=0".to_string()));
        assert!(args.contains(&"-dnsseed=0".to_string()));
        assert!(args.contains(&"-listenonion=0".to_string()));
        assert!(args.contains(&"-discover=0".to_string()));
        assert!(args.contains(&"-daemon=0".to_string()));
        assert!(args.contains(&"-printtoconsole".to_string()));
        assert!(args.contains(&"-debug=1".to_string()));
        assert!(
            !args.iter().any(|arg| arg.starts_with("-zebra-compat")),
            "P2P sidecar must not pass RPC-ingest flags: {args:?}"
        );

        // zcashd takes the last occurrence of a single-valued argument, so the
        // forced P2P pinning flags must come after operator extra_args.
        let debug_idx = args
            .iter()
            .position(|a| a == "-debug=1")
            .expect("extra arg present");
        for forced in [
            "-listen=0",
            "-dnsseed=0",
            "-listenonion=0",
            "-discover=0",
            "-daemon=0",
        ] {
            let forced_idx = args
                .iter()
                .position(|a| a == forced)
                .expect("forced flag present");
            assert!(
                forced_idx > debug_idx,
                "{forced} must come after extra_args"
            );
        }
    }

    #[test]
    fn peer_selection_extra_args_are_rejected() {
        for arg in [
            "-connect=1.2.3.4:8233",
            "--connect=1.2.3.4",
            "-addnode=1.2.3.4",
            "-seednode=1.2.3.4",
            "-noconnect",
            "-CONNECT=1.2.3.4",
        ] {
            let _rejected = reject_peer_selection_extra_args(&[arg.to_string()])
                .expect_err("peer-selection extra args must be rejected");
        }

        reject_peer_selection_extra_args(&[
            "-debug=1".to_string(),
            "-rpcport=18232".to_string(),
            "-maxconnections=8".to_string(),
        ])
        .expect("non-peer-selection extra args are allowed");
    }

    #[test]
    fn restart_count_resets_after_healthy_uptime() {
        assert!(should_reset_restart_count(
            Duration::from_secs(60 * 60),
            Duration::from_secs(60 * 60)
        ));
        assert!(should_reset_restart_count(
            Duration::from_secs(60 * 60 + 1),
            Duration::from_secs(60 * 60)
        ));
    }

    #[test]
    fn restart_count_does_not_reset_before_threshold() {
        assert!(!should_reset_restart_count(
            Duration::from_secs(60 * 60 - 1),
            Duration::from_secs(60 * 60)
        ));
        assert!(!should_reset_restart_count(
            Duration::from_secs(60 * 60),
            Duration::ZERO
        ));
    }

    #[test]
    fn restart_backoff_is_exponential_from_base_delay() {
        let base_delay = Duration::from_secs(2);
        let max_delay = Duration::from_secs(60);

        assert_eq!(restart_backoff_delay(base_delay, max_delay, 0), base_delay);
        assert_eq!(restart_backoff_delay(base_delay, max_delay, 1), base_delay);
        assert_eq!(
            restart_backoff_delay(base_delay, max_delay, 2),
            Duration::from_secs(4)
        );
        assert_eq!(
            restart_backoff_delay(base_delay, max_delay, 3),
            Duration::from_secs(8)
        );
    }

    #[test]
    fn restart_backoff_is_capped() {
        let delay = restart_backoff_delay(Duration::from_secs(2), Duration::from_secs(10), 10);

        assert_eq!(delay, Duration::from_secs(10));
    }

    #[test]
    fn restart_backoff_caps_saturated_delay() {
        let delay = restart_backoff_delay(Duration::MAX, Duration::from_secs(10), u32::MAX);

        assert_eq!(delay, Duration::from_secs(10));
    }

    #[tokio::test]
    async fn delay_wait_returns_on_shutdown_request() {
        let (shutdown_tx, mut shutdown_rx) = watch::channel(false);

        let wait = tokio::spawn(async move {
            wait_for_delay_or_shutdown(Duration::from_secs(60), &mut shutdown_rx).await
        });

        shutdown_tx
            .send(true)
            .expect("shutdown receiver exists because wait task owns it");

        let was_shutdown = tokio::time::timeout(Duration::from_secs(1), wait)
            .await
            .expect("interruptible delay should complete promptly")
            .expect("wait task should not panic");

        assert!(was_shutdown);
    }

    #[tokio::test]
    async fn delay_wait_returns_on_dropped_shutdown_sender() {
        let (shutdown_tx, mut shutdown_rx) = watch::channel(false);

        let wait = tokio::spawn(async move {
            wait_for_delay_or_shutdown(Duration::from_secs(60), &mut shutdown_rx).await
        });

        drop(shutdown_tx);

        let was_shutdown = tokio::time::timeout(Duration::from_secs(1), wait)
            .await
            .expect("interruptible delay should complete promptly")
            .expect("wait task should not panic");

        assert!(was_shutdown);
    }

    #[test]
    fn sanitize_child_log_line_strips_ansi_csi_sequences() {
        let line = "\u{1b}[32mINFO\u{1b}[0m ProcessNewTrustedBlockBatch";
        let sanitized = super::sanitize_child_log_line(line);

        assert_eq!(sanitized, "INFO ProcessNewTrustedBlockBatch");
    }

    #[test]
    fn sanitize_child_log_line_removes_control_chars() {
        let line = "good\u{0}text\u{8}\tkeeps-tab";
        let sanitized = super::sanitize_child_log_line(line);

        assert_eq!(sanitized, "goodtext\tkeeps-tab");
    }

    #[test]
    fn sanitize_child_log_line_keeps_clean_lines_unchanged() {
        let line = "UpdateTip: new best hash=abc height=42";
        let sanitized = super::sanitize_child_log_line(line);

        assert_eq!(sanitized, line);
    }

    /// A child that exits on SIGTERM within the grace period is never SIGKILLed,
    /// so its shutdown flush cannot be interrupted.
    #[cfg(unix)]
    #[tokio::test]
    async fn terminate_child_waits_for_graceful_exit() {
        let mut child = tokio::process::Command::new("/bin/sleep")
            .arg("60")
            .kill_on_drop(false)
            .spawn()
            .expect("sleep is available on unix test hosts");

        let start = std::time::Instant::now();
        super::terminate_child(&mut child, Duration::from_secs(30))
            .await
            .expect("terminate_child should succeed for a SIGTERM-compliant child");

        assert!(
            start.elapsed() < Duration::from_secs(30),
            "child should exit on SIGTERM well before the grace period"
        );
    }

    /// A child that ignores SIGTERM is force-killed only after the full grace
    /// period elapses.
    #[cfg(unix)]
    #[tokio::test(start_paused = true)]
    async fn terminate_child_kills_after_grace_period() {
        use std::process::Stdio;

        let mut child = tokio::process::Command::new("/bin/sh")
            .args(["-c", "trap '' TERM; while read _; do :; done"])
            .stdin(Stdio::piped())
            .kill_on_drop(true)
            .spawn()
            .expect("sh is available on unix test hosts");

        // Give the shell a moment of real time to install the TERM trap before
        // SIGTERM is sent; the paused clock only skips tokio timers.
        tokio::task::yield_now().await;
        std::thread::sleep(std::time::Duration::from_millis(200));

        super::terminate_child(&mut child, Duration::from_secs(5))
            .await
            .expect("terminate_child should fall back to SIGKILL");

        let status = child
            .try_wait()
            .expect("child status should be queryable after terminate_child");
        assert!(
            status.is_some(),
            "child must have been reaped after the SIGKILL fallback"
        );
    }
}